diff --git a/bench/ensemble_vs_pytester.py b/bench/ensemble_vs_pytester.py new file mode 100644 index 00000000000..c4f3432d690 --- /dev/null +++ b/bench/ensemble_vs_pytester.py @@ -0,0 +1,176 @@ +"""Compare the cost of the pytester harness against ``_pytest.ensemble``. + +pytest's own test suite is dominated by pytester-based tests: they write a +module to disk and run a full session against it. ``_pytest.ensemble`` runs +tests from in-memory objects in a reduced, hermetic configuration. This +script measures what that difference costs per run. + +The arms are, from heaviest to lightest: + +``subprocess`` + ``makepyfile`` + ``runpytest_subprocess`` - a whole new interpreter. +``inprocess`` + ``makepyfile`` + ``runpytest_inprocess`` - full config, capture and + terminal, with the rendered output captured for matching. +``inline`` + ``makepyfile`` + ``inline_run`` - same session, but no terminal + reporting or output parsing. +``ensemble`` + ``run_tests`` on in-memory functions - no rootdir discovery, no config + files, no conftests, no plugin autoload, no capture, no terminal. +``makepyfile`` + Only writes the module, without running anything, so that the file + materialization cost can be subtracted from the pytester arms. + +Each pytester iteration writes a *fresh* module, so it pays assertion +rewriting, compilation and import every time - which is exactly what a +pytester-based test does. The ensemble arm's functions are compiled once, +because an ensemble's sources are plain ``def`` statements in the test body. + +Run with:: + + python bench/ensemble_vs_pytester.py + pytest bench/ensemble_vs_pytester.py -s # equivalent +""" + +from __future__ import annotations + +from collections.abc import Callable +import contextlib +import io +from pathlib import Path +import time + +from _pytest.ensemble import run_tests +from _pytest.pytester import Pytester + + +#: Number of test functions per run. +SIZES = (1, 10, 100) + +#: Iterations per arm. The subprocess arm is orders of magnitude slower, so +#: it gets fewer; the numbers are still stable enough to compare magnitudes. +ITERATIONS = {"subprocess": 3} +DEFAULT_ITERATIONS = 10 + + +def _source(count: int) -> str: + return "\n".join(f"def test_{i}():\n assert {i} == {i}" for i in range(count)) + + +def _functions(count: int) -> list[Callable[[], None]]: + """The in-memory equivalent of :func:`_source`. + + Built once per size, outside the timed loop: an ensemble's sources are + plain ``def`` statements in the test body, compiled when the test + module is imported, not materialized per run. + """ + namespace: dict[str, Callable[[], None]] = {} + exec(compile(_source(count), "", "exec"), namespace) + return [namespace[f"test_{i}"] for i in range(count)] + + +def _files(path: Path) -> int: + """Files below *path*, ignoring bytecode caches.""" + return sum( + 1 + for p in path.rglob("*") + if p.is_file() and "__pycache__" not in p.parts and p.suffix != ".pyc" + ) + + +def _measure( + run: Callable[[int], None], iterations: int, path: Path +) -> tuple[float, float]: + """Return (seconds, files created) per iteration. + + Terminal output of the measured runs is redirected away so it does not + pollute the report; the rendering cost itself is still paid. + """ + with contextlib.redirect_stdout(io.StringIO()): + run(0) # warm up: let lazy imports happen outside the measurement + before = _files(path) + start = time.perf_counter() + for i in range(1, iterations + 1): + run(i) + elapsed = time.perf_counter() - start + return elapsed / iterations, (_files(path) - before) / iterations + + +def _arms(pytester: Pytester, count: int) -> dict[str, Callable[[int], None]]: + source = _source(count) + functions = _functions(count) + + def write(tag: str, i: int) -> str: + name = f"test_{tag}_{count}_{i}" + pytester.makepyfile(**{name: source}) + return f"{name}.py" + + def subprocess_(i: int) -> None: + result = pytester.runpytest_subprocess(write("sub", i)) + assert result.ret == 0, result.outlines + + def inprocess(i: int) -> None: + result = pytester.runpytest_inprocess(write("inp", i)) + result.assert_outcomes(passed=count) + + def inline(i: int) -> None: + rec = pytester.inline_run(write("inl", i)) + rec.assertoutcome(passed=count) + + def ensemble(i: int) -> None: + record = run_tests(*functions, rootpath=pytester.path) + record.assert_outcomes(passed=count) + + def makepyfile(i: int) -> None: + write("raw", i) + + return { + "subprocess": subprocess_, + "inprocess": inprocess, + "inline": inline, + "ensemble": ensemble, + "makepyfile": makepyfile, + } + + +def test_ensemble_vs_pytester(pytester: Pytester) -> None: + """Run with ``-s`` and read the tables; only the file count is asserted.""" + print() + by_size: dict[int, dict[str, tuple[float, float]]] = {} + for count in SIZES: + results = {} + for name, run in _arms(pytester, count).items(): + iterations = ITERATIONS.get(name, DEFAULT_ITERATIONS) + results[name] = _measure(run, iterations, pytester.path) + by_size[count] = results + + baseline = results["ensemble"][0] + print(f"\n{count} test function(s) per run") + print(f" {'arm':<12} {'per run':>10} {'files':>8} {'vs ensemble':>13}") + for name, (seconds, files) in results.items(): + ratio = f"{seconds / baseline:.1f}x" if baseline else "n/a" + print(f" {name:<12} {seconds * 1000:>8.2f}ms {files:>8.0f} {ratio:>13}") + + # Split the cost in two: what every run pays regardless of size, and + # what each additional test function adds. A two point estimate over the + # smallest and largest size is enough to tell the two apart. + low, high = min(SIZES), max(SIZES) + print(f"\nfixed cost per run vs marginal cost per test ({low} -> {high} tests)") + print(f" {'arm':<12} {'fixed':>10} {'per test':>12}") + for name in by_size[low]: + fixed = by_size[low][name][0] + marginal = (by_size[high][name][0] - fixed) / (high - low) + print(f" {name:<12} {fixed * 1000:>8.2f}ms {marginal * 1000:>10.3f}ms") + + print("\n(file counts exclude __pycache__; ensemble writes nothing at all)") + + # The timings are informational, but "no disk at all" is the design claim. + for count, results in by_size.items(): + assert results["ensemble"][1] == 0, f"ensemble wrote files at size {count}" + + +if __name__ == "__main__": + import pytest + + raise SystemExit(pytest.main([__file__, "-s", "-q", "-p", "no:randomly"])) diff --git a/changelog/14809.bugfix.rst b/changelog/14809.bugfix.rst new file mode 100644 index 00000000000..88776de9f4f --- /dev/null +++ b/changelog/14809.bugfix.rst @@ -0,0 +1,4 @@ +:meth:`pytest.Class.from_parent` no longer silently discards an ``obj`` argument. +The parameter was accepted and then dropped, so the collected class was always +looked up by name on the parent's object; a passed ``obj`` is now used as the +collector's object. diff --git a/changelog/14809.feature.rst b/changelog/14809.feature.rst new file mode 100644 index 00000000000..a41c538256b --- /dev/null +++ b/changelog/14809.feature.rst @@ -0,0 +1,15 @@ +Added the experimental internal ``_pytest.ensemble`` API for testing pytest with pytest: +build a hermetic nested configuration from declarative data (``ConfigSpec``/``configured()``), +collect test items from in-memory python objects instead of files on disk +(``build_module()``, ``collect_tests()``), run them through the standard runtest protocol, +and assert on typed report objects (``RunRecord``) instead of glob-matching terminal output. + +In support of this: + +* ``Config.parse()`` was split into behavior-preserving phase methods, so programmatic + config construction shares real code with command line parsing. +* ``Config.ArgsSource.SPEC`` marks args taken verbatim from a programmatic specification. +* ``_pytest.config.get_config()`` accepts an explicit invocation ``dir=``. +* ``PdbTrace`` wraps ``pytest_runtest_call`` rather than ``pytest_pyfunc_call``, so + ``--trace`` reaches unittest test cases through the normal hook instead of + ``_pytest.unittest`` calling into ``_pytest.debugging``. diff --git a/src/_pytest/assertion/__init__.py b/src/_pytest/assertion/__init__.py index a171633f320..30435dce493 100644 --- a/src/_pytest/assertion/__init__.py +++ b/src/_pytest/assertion/__init__.py @@ -26,6 +26,11 @@ from _pytest.main import Session +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def pytest_addoption(parser: Parser) -> None: group = parser.getgroup("debugconfig") group.addoption( diff --git a/src/_pytest/config/__init__.py b/src/_pytest/config/__init__.py index 7a7573de050..4a099fc4164 100644 --- a/src/_pytest/config/__init__.py +++ b/src/_pytest/config/__init__.py @@ -361,13 +361,14 @@ def get_config( plugins: Sequence[str | _PluggyPlugin] | None = None, *, prog: str | None = None, + dir: pathlib.Path | None = None, ) -> Config: # Subsequent calls to main will create a fresh instance. pluginmanager = PytestPluginManager() invocation_params = Config.InvocationParams( args=args or (), plugins=plugins, - dir=pathlib.Path.cwd(), + dir=dir if dir is not None else pathlib.Path.cwd(), ) config = Config(pluginmanager, invocation_params=invocation_params, prog=prog) @@ -1116,10 +1117,17 @@ class ArgsSource(enum.Enum): INCOVATION_DIR = INVOCATION_DIR # backwards compatibility alias #: 'testpaths' configuration value. TESTPATHS = enum.auto() + #: Programmatic specification; args are taken verbatim, without + #: filesystem-based fallbacks (experimental). + SPEC = enum.auto() # Set by cacheprovider plugin. cache: Cache + # The parsed command line namespace, including only options known at + # the time of parsing. Set during parse(). + known_args_namespace: argparse.Namespace + def __init__( self, pluginmanager: PytestPluginManager, @@ -1291,10 +1299,18 @@ def _ensure_unconfigure(self) -> None: self._cleanup_stack = contextlib.ExitStack() def get_terminal_writer(self) -> TerminalWriter: + """The writer the terminal plugin reports through. + + Without that plugin - as in a programmatically constructed config - + a plain writer over the current stdout is returned instead, so that + code needing to render something does not have to care whether + anything is reporting. + """ terminalreporter: TerminalReporter | None = self.pluginmanager.get_plugin( "terminalreporter" ) - assert terminalreporter is not None + if terminalreporter is None: + return create_terminal_writer(self) return terminalreporter._tw def pytest_cmdline_parse( @@ -1581,34 +1597,28 @@ def _get_unknown_ini_keys(self) -> set[str]: known_keys = self._parser._inidict.keys() | self._parser._ini_aliases.keys() return self._inicfg.keys() - known_keys - def parse(self, args: list[str], addopts: bool = True) -> None: - # Parse given cmdline arguments into this config object. - assert self.args == [], ( - "can only parse cmdline args at most once per Config object" - ) - - self.hook.pytest_addhooks.call_historic( - kwargs=dict(pluginmanager=self.pluginmanager) - ) + def _preparse_addopts(self, args: list[str]) -> None: + """Prepend options from the PYTEST_ADDOPTS environment variable (in place).""" + env_addopts = os.environ.get("PYTEST_ADDOPTS", "") + if len(env_addopts): + args[:] = ( + self._validate_args(shlex.split(env_addopts), "via PYTEST_ADDOPTS") + + args + ) - if addopts: - env_addopts = os.environ.get("PYTEST_ADDOPTS", "") - if len(env_addopts): - args[:] = ( - self._validate_args(shlex.split(env_addopts), "via PYTEST_ADDOPTS") - + args - ) + def _apply_rootdir( + self, + *, + rootpath: pathlib.Path, + inipath: pathlib.Path | None, + inicfg: ConfigDict, + ignored_config_files: Sequence[str], + ) -> None: + """Apply an already-determined rootdir/inifile setup to this config. - # At this point, self.option contains only defaults from the _processopt - # callback. - ns = self._parser.parse_known_args(args, namespace=copy.copy(self.option)) - rootpath, inipath, inicfg, ignored_config_files = determine_setup( - inifile=ns.inifilename, - override_ini=ns.override_ini, - args=ns.file_or_dir, - rootdir_cmd_arg=ns.rootdir or None, - invocation_dir=self.invocation_params.dir, - ) + Normally fed by :func:`determine_setup`, but usable with explicitly + constructed values as well (experimental). + """ self._rootpath = rootpath self._inipath = inipath self._ignored_config_files = ignored_config_files @@ -1616,6 +1626,8 @@ def parse(self, args: list[str], addopts: bool = True) -> None: self._parser.extra_info["rootdir"] = str(self.rootpath) self._parser.extra_info["inifile"] = str(self.inipath) + def _register_core_ini_options(self) -> None: + """Register the ini options which parse() itself consumes.""" self._parser.addini("addopts", "Extra command line options", "args") self._parser.addini("minversion", "Minimally required pytest version") self._parser.addini( @@ -1628,20 +1640,8 @@ def parse(self, args: list[str], addopts: bool = True) -> None: default=[], ) - if addopts: - args[:] = ( - self._validate_args(self.getini("addopts"), "via addopts config") + args - ) - - self.known_args_namespace = self._parser.parse_known_args( - args, namespace=copy.copy(self.option) - ) - if addopts: - # addopts may have added overrides (especially via OverrideIniAction). - # The thing can be endlessly circular but we only do one level (#14442). - if overrides := parse_override_ini(self.known_args_namespace.override_ini): - self._inicfg.update(overrides) - self._inicache.clear() + def _load_plugins_phase(self, args: list[str]) -> None: + """Load plugins from args, entry points and environment; reparse known args.""" self._checkversion() self._consider_importhook() self._configure_python_path() @@ -1681,6 +1681,8 @@ def parse(self, args: list[str], addopts: bool = True) -> None: self._validate_plugins() self._warn_about_skipped_plugins() + def _load_initial_conftests_phase(self, args: list[str]) -> None: + """Default confcutdir and fire the pytest_load_initial_conftests hook.""" if self.known_args_namespace.confcutdir is None: if self.inipath is not None: confcutdir = str(self.inipath.parent) @@ -1702,19 +1704,83 @@ def parse(self, args: list[str], addopts: bool = True) -> None: else: raise + def _finalize_parse(self, args: list[str], *, decide_args: bool = True) -> None: + """Fully parse args into self.option and decide the initial args. + + With ``decide_args=False`` (experimental), the positional args are + taken verbatim without the testpaths/invocation-dir fallbacks and + ``args_source`` is set to :attr:`ArgsSource.SPEC`. + """ + if not hasattr(self, "known_args_namespace"): + self.known_args_namespace = self._parser.parse_known_args( + args, namespace=copy.copy(self.option) + ) try: self._parser.parse(args, namespace=self.option) except PrintHelp: return - self.args, self.args_source = self._decide_args( - args=getattr(self.option, FILE_OR_DIR), - pyargs=self.option.pyargs, - testpaths=self.getini("testpaths"), + if decide_args: + self.args, self.args_source = self._decide_args( + args=getattr(self.option, FILE_OR_DIR), + pyargs=self.option.pyargs, + testpaths=self.getini("testpaths"), + invocation_dir=self.invocation_params.dir, + rootpath=self.rootpath, + warn=True, + ) + else: + self.args = list(getattr(self.option, FILE_OR_DIR)) + self.args_source = Config.ArgsSource.SPEC + + def parse(self, args: list[str], addopts: bool = True) -> None: + # Parse given cmdline arguments into this config object. + assert self.args == [], ( + "can only parse cmdline args at most once per Config object" + ) + + self.hook.pytest_addhooks.call_historic( + kwargs=dict(pluginmanager=self.pluginmanager) + ) + + if addopts: + self._preparse_addopts(args) + + # At this point, self.option contains only defaults from the _processopt + # callback. + ns = self._parser.parse_known_args(args, namespace=copy.copy(self.option)) + rootpath, inipath, inicfg, ignored_config_files = determine_setup( + inifile=ns.inifilename, + override_ini=ns.override_ini, + args=ns.file_or_dir, + rootdir_cmd_arg=ns.rootdir or None, invocation_dir=self.invocation_params.dir, - rootpath=self.rootpath, - warn=True, ) + self._apply_rootdir( + rootpath=rootpath, + inipath=inipath, + inicfg=inicfg, + ignored_config_files=ignored_config_files, + ) + self._register_core_ini_options() + + if addopts: + args[:] = ( + self._validate_args(self.getini("addopts"), "via addopts config") + args + ) + + self.known_args_namespace = self._parser.parse_known_args( + args, namespace=copy.copy(self.option) + ) + if addopts: + # addopts may have added overrides (especially via OverrideIniAction). + # The thing can be endlessly circular but we only do one level (#14442). + if overrides := parse_override_ini(self.known_args_namespace.override_ini): + self._inicfg.update(overrides) + self._inicache.clear() + self._load_plugins_phase(args) + self._load_initial_conftests_phase(args) + self._finalize_parse(args) def issue_config_time_warning(self, warning: Warning, stacklevel: int) -> None: """Issue and handle a warning during the "configure" stage. @@ -2232,17 +2298,23 @@ def create_terminal_writer( Every code which requires a TerminalWriter object and has access to a config object should use this function. + + The presentation options are read defensively: they are registered by the + terminal plugin, which a programmatically constructed config need not + load, and a writer is still useful without them. """ tw = TerminalWriter(file=file) - if config.option.color == "yes": + color = getattr(config.option, "color", None) + if color == "yes": tw.hasmarkup = True - elif config.option.color == "no": + elif color == "no": tw.hasmarkup = False - if config.option.code_highlight == "yes": + code_highlight = getattr(config.option, "code_highlight", None) + if code_highlight == "yes": tw.code_highlight = True - elif config.option.code_highlight == "no": + elif code_highlight == "no": tw.code_highlight = False return tw diff --git a/src/_pytest/debugging.py b/src/_pytest/debugging.py index b256f83c8bf..2374bf15800 100644 --- a/src/_pytest/debugging.py +++ b/src/_pytest/debugging.py @@ -22,11 +22,18 @@ from _pytest.config import PytestPluginManager from _pytest.config.argparsing import Parser from _pytest.config.exceptions import UsageError +from _pytest.nodes import Item from _pytest.nodes import Node +from _pytest.python import Function from _pytest.reports import BaseReport from _pytest.runner import CallInfo +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def _validate_usepdb_cls(value: str) -> tuple[str, str]: """Validate syntax of --pdbcls option.""" try: @@ -302,9 +309,15 @@ def pytest_internalerror(self, excinfo: ExceptionInfo[BaseException]) -> None: class PdbTrace: - @hookimpl(wrapper=True) - def pytest_pyfunc_call(self, pyfuncitem) -> Generator[None, object, object]: - wrap_pytest_function_for_tracing(pyfuncitem) + # trylast so this is the innermost wrapper, i.e. runs inside + # CaptureManager's - _init_pdb() suspends capturing, and an outer wrapper + # would have that undone again by the capture manager. + @hookimpl(wrapper=True, trylast=True) + def pytest_runtest_call(self, item: Item) -> Generator[None, object, object]: + # Not every item is function-backed (doctests, custom items), and + # TestCaseFunction never reaches pytest_pyfunc_call at all. + if isinstance(item, Function): + wrap_pytest_function_for_tracing(item) return (yield) @@ -327,23 +340,18 @@ def wrapper(*args, **kwargs) -> None: pyfuncitem.obj = wrapper -def maybe_wrap_pytest_function_for_tracing(pyfuncitem) -> None: - """Wrap the given pytestfunct item for tracing support if --trace was given in - the command line.""" - if pyfuncitem.config.getvalue("trace"): - wrap_pytest_function_for_tracing(pyfuncitem) - - def _enter_pdb( node: Node, excinfo: ExceptionInfo[BaseException], rep: BaseReport ) -> BaseReport: # XXX we reuse the TerminalReporter's terminalwriter # because this seems to avoid some encoding related troubles - # for not completely clear reasons. - tw = node.config.pluginmanager.getplugin("terminalreporter")._tw + # for not completely clear reasons. Falls back to a plain writer when + # nothing is reporting, rather than crashing on entry to the debugger. + tw = node.config.get_terminal_writer() tw.line() - showcapture = node.config.option.showcapture + # Registered by the terminal plugin; default to its default when absent. + showcapture = node.config.getoption("showcapture", "all") for sectionname, content in ( ("stdout", rep.capstdout), diff --git a/src/_pytest/ensemble/__init__.py b/src/_pytest/ensemble/__init__.py new file mode 100644 index 00000000000..fcf99973129 --- /dev/null +++ b/src/_pytest/ensemble/__init__.py @@ -0,0 +1,309 @@ +"""Nested configs and in-memory collection for pytest-under-pytest testing. + +EXPERIMENTAL: internal API, no backwards-compatibility guarantees. + +A pytest *ensemble* is a deliberately small pytest assembled from parts +handed to it, rather than a full session discovered from a filesystem: a +hermetic nested configuration built from declarative data +(:class:`ConfigSpec`), test items collected from in-memory python objects +instead of files on disk, run through the standard runtest protocol, with +typed report objects (:class:`RunRecord`) to assert on instead of +glob-matching rendered terminal output. + +Known limitations (by design, for now): + +* Ensembles never load conftest files; pass plugin objects via + ``ConfigSpec.extra_plugins`` instead. A plugin object is equivalent to a + conftest at the *rootdir*; nothing below rootdir is expressible, so + per-directory conftest scoping has no analogue. +* **Capture does not nest.** ``capture`` is not loaded by default, and + loading it via ``with_plugins("capture")`` registers a ``CaptureManager`` + without ever starting global capturing - ``pytest_load_initial_conftests`` + is where that happens, and an ensemble does not run it. Fixture-level + ``capsys``/``capfd`` inside the ensemble therefore work, but output from + the item itself is neither captured nor reported: **it escapes to the + stdout of whatever is running the ensemble.** Starting global capturing + here would redirect the process's streams underneath the host that is + already capturing them; making that safe is the stack-aware + ``CaptureManager`` work, not something this package can paper over. +* The ``terminal`` plugin is not loaded by default, so by default there is + no rendered output at all. ``capture_output=True`` loads it bound to a + private buffer, which is also what makes terminal-only options such as + ``--tb``, ``-v`` and ``--color`` available. Writing to a buffer of its + own counts as capturing, so the progress column renders - which in turn + means ``--capture=no`` cannot switch it off in an ensemble. +* Assertion *rewriting* is not applied to ensemble sources; sources defined + in the host test suite's own files are already rewritten by the host. The + assertion *explanation* is the ensemble's own, because the ``assertion`` + plugin is loaded by default - without it ``util._reprcompare`` would stay + bound to the host's. +* Sources without real code objects (``exec``'d, lambdas) degrade + ``reportinfo``/traceback quality. +* An item's location comes from its function's code object, so a source + written inline in a test body reports the *host* file - anything + rendering ``file:line`` names the host, and the terminal groups all such + items as one module. Use :func:`module_from_path` to run a script that + exists on disk as itself; its items then report their own path and line. +* Process-global warning filters active around the ensemble (e.g. the + host suite's ``filterwarnings = error``) are inherited; an ensemble's + own ``inicfg={"filterwarnings": [...]}`` takes precedence over them. + Warnings the ensemble itself raises do not escape: those from configure + and unconfigure are captured and reported on :attr:`RunRecord.warnings`. +""" + +from __future__ import annotations + +import contextlib +import dataclasses +import io +import pathlib +from typing import TYPE_CHECKING + +from _pytest.config import Config +from _pytest.ensemble.collection import build_module +from _pytest.ensemble.collection import collect_sources +from _pytest.ensemble.collection import DEFAULT_MODULE_NAME +from _pytest.ensemble.collection import EnsembleModule +from _pytest.ensemble.collection import module_from_path +from _pytest.ensemble.collection import running_session +from _pytest.ensemble.collection import Source +from _pytest.ensemble.config import ConfigSpec +from _pytest.ensemble.config import configured +from _pytest.ensemble.config import DEFAULT_PLUGINS +from _pytest.ensemble.results import ensure_recorder +from _pytest.ensemble.results import ItemRecord +from _pytest.ensemble.results import run_items +from _pytest.ensemble.results import RunRecord +from _pytest.main import Session +from _pytest.nodes import Collector +from _pytest.nodes import Item +from _pytest.reports import CollectReport + + +if TYPE_CHECKING: + from types import TracebackType + + from typing_extensions import Self + + +__all__ = [ + "DEFAULT_MODULE_NAME", + "DEFAULT_PLUGINS", + "ConfigSpec", + "Ensemble", + "EnsembleModule", + "ItemRecord", + "RunRecord", + "Source", + "build_module", + "collect_sources", + "collect_tests", + "configured", + "module_from_path", + "run_items", + "run_tests", + "running_session", +] + + +def _resolve_spec(spec: ConfigSpec | None, rootpath: pathlib.Path | None) -> ConfigSpec: + if spec is None: + spec = ConfigSpec(rootpath=rootpath) + elif rootpath is not None and spec.rootpath is None: + spec = spec.replace(rootpath=rootpath) + return spec + + +class Ensemble: + """A nested config plus a running session, over which sources can be + collected and run stepwise. + + Used as a context manager; the config is configured and the session + started on enter, and both are torn down on exit:: + + with Ensemble(test_fn, SomeTestClass, rootpath=tmp_path) as ensemble: + items = ensemble.collect() + record = ensemble.run() + record.assert_outcomes(passed=2) + + For the common one-shot cases use :func:`run_tests` or + :func:`collect_tests` instead. + """ + + #: The configured nested config; only available while entered. + config: Config + #: The started session; only available while entered. + session: Session + + def __init__( + self, + *sources: Source, + rootpath: pathlib.Path | None = None, + spec: ConfigSpec | None = None, + name: str = DEFAULT_MODULE_NAME, + capture_output: bool = False, + ) -> None: + self._spec = _resolve_spec(spec, rootpath) + self._sources = sources + self._name = name + self._collected = False + self._round = 0 + self._stack: contextlib.ExitStack | None = None + self._sink: io.StringIO | None = None + if capture_output: + self._sink = io.StringIO() + self._spec = self._spec.replace(output=self._sink) + if "terminal" not in self._spec.plugins: + self._spec = self._spec.with_plugins("terminal") + + @property + def output(self) -> str: + """What the terminal plugin has rendered so far, if capturing. + + Failure sections and the summary line are only written as the + session finishes, so read this after leaving the context manager to + get the whole report. + """ + return self._sink.getvalue() if self._sink is not None else "" + + def __enter__(self) -> Self: + if self._stack is not None: + raise RuntimeError("Ensemble is not reentrant") + stack = contextlib.ExitStack() + try: + self.config = stack.enter_context(configured(self._spec)) + self.session = stack.enter_context(running_session(self.config)) + except BaseException: + stack.close() + raise + self._stack = stack + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + assert self._stack is not None, "Ensemble was not entered" + stack, self._stack = self._stack, None + # Forward the exception rather than closing blind, so the session + # teardown sees that the body failed. Neither of the entered context + # managers suppresses, so the result is not propagated. + stack.__exit__(exc_type, exc, tb) + + @property + def collect_errors(self) -> list[CollectReport]: + """The collect reports that failed, if any. + + Collection failures do not raise, so an ensemble that collects + nothing because something blew up looks exactly like one that had + nothing to collect; this is how the two are told apart. + """ + recorder = ensure_recorder(self.config) + return [report for report in recorder.collect_reports if report.failed] + + def collect(self, *sources: Source) -> list[Item]: + """Collect the ensemble's sources (plus any given extra ones). + + Idempotent: calling it again without new sources returns the items + already collected, rather than registering the same collectors a + second time and reporting every test twice. + """ + if self._collected: + if not sources: + return list(self.session.items) + # Later rounds get their own synthesized module, so that loose + # sources do not land on a module path already in the tree. + self._round += 1 + return collect_sources( + self.session, *sources, name=f"{self._name}_{self._round}" + ) + self._collected = True + return collect_sources(self.session, *self._sources, *sources, name=self._name) + + def run(self, items: list[Item] | None = None) -> RunRecord: + """Run the collected items (collecting first if needed). + + Anything the session emits while tearing down - late warnings, the + rendered summary - necessarily arrives after this returns; see + :meth:`final_record`. + """ + if not self._collected: + self.collect() + record = run_items(self.session, items) + if self._sink is not None: + record = dataclasses.replace(record, output=self.output) + return record + + def final_record(self, record: RunRecord) -> RunRecord: + """Refresh *record* with everything session teardown produced. + + ``pytest_sessionfinish`` runs as the ensemble is left, and plugins + emit warnings (and the terminal writes its summary) from there. + A record built during :meth:`run` predates all of it, which makes + ``assert_outcomes(warnings=...)`` quietly wrong. + """ + refreshed = RunRecord.from_recorder( + ensure_recorder(self.config), config=self.config + ) + return dataclasses.replace( + refreshed, output=self.output, stopped=record.stopped + ) + + +def run_tests( + *sources: Source, + rootpath: pathlib.Path | None = None, + spec: ConfigSpec | None = None, + name: str = DEFAULT_MODULE_NAME, + capture_output: bool = False, +) -> RunRecord: + """Collect and run the given in-memory sources in a nested config; + return the structured results. + + With ``capture_output``, the terminal plugin is loaded and what it + renders is captured into :attr:`RunRecord.output` (and + :attr:`RunRecord.stdout`) instead of reaching the real stdout. + """ + ensemble = Ensemble( + *sources, + rootpath=rootpath, + spec=spec, + name=name, + capture_output=capture_output, + ) + with ensemble: + record = ensemble.run() + # Session teardown happens on the way out of the block, and it both + # emits reports and warnings of its own and writes the failure sections + # and summary line - so the complete picture only exists once the block + # has been left. + return ensemble.final_record(record) + + +def collect_tests( + *sources: Source, + rootpath: pathlib.Path | None = None, + spec: ConfigSpec | None = None, + name: str = DEFAULT_MODULE_NAME, +) -> list[Item]: + """Collect (only) the given in-memory sources; return the items. + + The nested config and session are torn down before returning; the + items remain usable for structural assertions (names, nodeids, marks). + + Raises :class:`~_pytest.nodes.Collector.CollectError` if collection + failed: an empty list is otherwise indistinguishable from "collection + blew up", which would quietly turn a "collects nothing" assertion into + one that holds for the wrong reason. Use :class:`Ensemble` directly, or + :func:`run_tests`, to inspect collection failures instead. + """ + with Ensemble(*sources, rootpath=rootpath, spec=spec, name=name) as ensemble: + items = ensemble.collect() + if errors := ensemble.collect_errors: + raise Collector.CollectError( + "collection failed:\n" + + "\n".join(f"{r.nodeid}: {r.longrepr}" for r in errors) + ) + return items diff --git a/src/_pytest/ensemble/collection.py b/src/_pytest/ensemble/collection.py new file mode 100644 index 00000000000..31787005c12 --- /dev/null +++ b/src/_pytest/ensemble/collection.py @@ -0,0 +1,217 @@ +"""In-memory collection for ensembles (EXPERIMENTAL).""" + +from __future__ import annotations + +from collections.abc import Callable +from collections.abc import Iterator +import contextlib +import importlib.util +import pathlib +import sys +import types +from typing import Final + +from _pytest.config import Config +from _pytest.config import ExitCode +from _pytest.ensemble.results import ensure_recorder +from _pytest.main import Session +from _pytest.nodes import Collector +from _pytest.nodes import Item +from _pytest.python import Module +from _pytest.reports import CollectReport + + +#: A test source: a module-like object collected as a virtual module, or a +#: loose class/callable wrapped into a synthesized one. +Source = types.ModuleType | type | Callable[..., object] + +#: Default name for the synthesized module holding loose sources. +DEFAULT_MODULE_NAME: Final[str] = "test_ensemble" + + +def _real_path(obj: types.ModuleType) -> pathlib.Path | None: + """The module's own file, if it was imported from one that still exists.""" + filename = getattr(obj, "__file__", None) + if filename is None: + return None + path = pathlib.Path(filename) + return path if path.is_file() else None + + +class EnsembleModule(Module): + """A Module collector backed by an in-memory python object. + + For a synthesized module the ``path`` is rootdir-relative - giving + well-formed nodeids - but never touched on disk. A module that was + genuinely imported keeps its real ``__file__`` instead, so its items + report true locations and a rendered ``file:line`` names the source it + actually came from. + + Either way the standard import chokepoint is bypassed: ``_getobj`` + serves the object it was handed. + """ + + _preset_obj: types.ModuleType + + @classmethod + def from_parent( # type: ignore[override] + cls, + parent: Session, + *, + obj: types.ModuleType, + name: str | None = None, + path: pathlib.Path | None = None, + ) -> EnsembleModule: + """The public constructor.""" + if name is None: + name = obj.__name__ + if path is None: + path = _real_path(obj) or parent.config.rootpath / f"{name}.py" + node: EnsembleModule = super().from_parent(parent, path=path) + node._preset_obj = obj + return node + + def _getobj(self) -> types.ModuleType: + return self._preset_obj + + +class EnsembleCollection: + """Plugin serving a preset list of collectors as the session's + collection tree, instead of walking filesystem paths.""" + + name = "ensemble-collection" + + def __init__(self, session: Session) -> None: + self.session = session + self.collectors: list[Collector] = [] + + def pytest_make_collect_report(self, collector: Collector) -> CollectReport | None: + if collector is self.session: + return CollectReport( + collector.nodeid, "passed", None, list(self.collectors) + ) + return None + + +def ensemble_collection(session: Session) -> EnsembleCollection: + """Get the preset-collection plugin for the session, installing it if needed.""" + plugin: EnsembleCollection | None = session.config.pluginmanager.get_plugin( + EnsembleCollection.name + ) + if plugin is None: + plugin = EnsembleCollection(session) + session.config.pluginmanager.register(plugin, EnsembleCollection.name) + assert plugin.session is session + return plugin + + +@contextlib.contextmanager +def running_session(config: Config) -> Iterator[Session]: + """A started :class:`Session` for a configured ensemble config. + + ``pytest_sessionstart`` runs on enter (installing ``SetupState`` and the + ``FixtureManager``); ``pytest_sessionfinish`` runs on exit. + """ + session = Session.from_config(config) + ensure_recorder(config) + ensemble_collection(session) + config.hook.pytest_sessionstart(session=session) + exitstatus: int | ExitCode = ExitCode.OK + try: + yield session + except BaseException: + exitstatus = ExitCode.INTERNAL_ERROR + raise + finally: + config.hook.pytest_sessionfinish(session=session, exitstatus=exitstatus) + + +def build_module( + name: str, *members: object, **named_members: object +) -> types.ModuleType: + """Create an in-memory module with an explicit name, collecting the + given members together. + + Positional members are stored under their ``__name__``; keyword + members under the given keyword (e.g. ``pytestmark=...``). + """ + module = types.ModuleType(name) + for member in members: + member_name = getattr(member, "__name__", None) + if member_name is None: + raise ValueError( + f"member {member!r} has no __name__; pass it as a keyword instead" + ) + setattr(module, member_name, member) + for member_name, member in named_members.items(): + setattr(module, member_name, member) + return module + + +def module_from_path(path: pathlib.Path, name: str | None = None) -> types.ModuleType: + """Import *path* as a module without registering it in ``sys.modules``. + + For running an example script that lives on disk as itself, rather than + copying it somewhere first: the resulting module keeps its real + ``__file__``, so :class:`EnsembleModule` gives its items true paths and + line numbers. + + Staying out of ``sys.modules`` keeps the ensemble hermetic and lets the + same file be imported more than once in a session. Bytecode writing is + suppressed for the same reason: importing a script must not leave a + ``__pycache__`` beside it. + """ + if name is None: + name = path.stem + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: + raise ImportError(f"cannot import {path} as a module") + module = importlib.util.module_from_spec(spec) + written = sys.dont_write_bytecode + sys.dont_write_bytecode = True + try: + spec.loader.exec_module(module) + finally: + sys.dont_write_bytecode = written + return module + + +def collect_sources( + session: Session, + *sources: Source, + name: str = DEFAULT_MODULE_NAME, +) -> list[Item]: + """Collect test items from in-memory objects through the standard + collection protocol. + + Module-like sources each become a :class:`EnsembleModule`; loose + classes and callables are wrapped into one synthesized module named + ``name``. The regular ``pytest_collection`` hook then runs, so + ``-k``/``-m`` deselection, ``pytest_collection_modifyitems`` and + ``pytest_collection_finish`` all apply. + """ + config = session.config + collection = ensemble_collection(session) + loose: list[object] = [] + for source in sources: + if isinstance(source, types.ModuleType): + collection.collectors.append( + EnsembleModule.from_parent(session, obj=source) + ) + else: + loose.append(source) + if loose: + module = build_module(name, *loose) + if not config.getini("collect_imported_tests"): + # python.py drops objects whose __module__ differs from the + # containing module; synthesized namespaces always differ. + raise ValueError( + "collect_imported_tests=False would silently drop loose " + "ensemble sources; pass a real module object instead" + ) + collection.collectors.append( + EnsembleModule.from_parent(session, obj=module, name=name) + ) + + config.hook.pytest_collection(session=session) + return session.items diff --git a/src/_pytest/ensemble/config.py b/src/_pytest/ensemble/config.py new file mode 100644 index 00000000000..69901251493 --- /dev/null +++ b/src/_pytest/ensemble/config.py @@ -0,0 +1,223 @@ +"""Nested config construction for ensembles (EXPERIMENTAL).""" + +from __future__ import annotations + +from collections.abc import Iterator +from collections.abc import Mapping +import contextlib +import copy +import dataclasses +import pathlib +from typing import Final +from typing import TextIO +import warnings + +from _pytest.config import Config +from _pytest.config import essential_plugins +from _pytest.config import PytestPluginManager +from _pytest.config.findpaths import ConfigValue +from _pytest.config.findpaths import parse_override_ini +from _pytest.stash import StashKey +from _pytest.terminal import terminal_file_key + + +#: Warnings raised while the ensemble config was being configured or +#: unconfigured. They are recorded rather than allowed out, so that an +#: ensemble cannot make its *host* fail - this suite runs with +#: ``filterwarnings = error``. +config_warnings_key = StashKey[list[warnings.WarningMessage]]() + + +#: Plugins loaded into an ensemble config by default: the essential core +#: plus the plugins that give tests their usual semantics, deliberately +#: excluding everything that renders output, captures io, or installs +#: process-global state (terminal, capture, cacheprovider, assertion, +#: debugging, faulthandler, logging, threadexception, unraisableexception, ...). +DEFAULT_PLUGINS: Final[tuple[str, ...]] = ( + *essential_plugins, # mark, main, runner, fixtures, helpconfig + "python", + "skipping", + "warnings", + "reports", + "unittest", + "monkeypatch", + "recwarn", + "tmpdir", + # Not for the rewriting - that is installed from Config._preparse, which + # an ensemble never runs - but for the failure *explanation*. Without + # this plugin ``assertion.util._reprcompare`` stays bound to whatever the + # host installed, so an ensemble silently renders its assertions with the + # host's verbosity, ini values and pytest_assertrepr_compare hooks. + "assertion", +) + + +@dataclasses.dataclass(frozen=True, kw_only=True) +class ConfigSpec: + """Declarative description of a nested pytest configuration. + + A spec is plain data: build it in a fixture, parametrize it, or derive + variants via :meth:`replace`/:meth:`with_plugins`. State is only + acquired when the spec is passed to :func:`configured`. + """ + + #: Root directory anchoring nodeids and synthetic module paths. + #: Must be an existing directory; it is never read from or written to. + rootpath: pathlib.Path | None = None + + #: Command line arguments, taken verbatim (never coerced to paths). + args: tuple[str, ...] = () + + #: Ini configuration values, authoritative (no config file is read). + #: Values are plain ``str``/``list[str]`` (ini mode) or preconstructed + #: :class:`ConfigValue` instances. + inicfg: Mapping[str, object] = dataclasses.field(default_factory=dict) + + #: Built-in plugin names to load. + plugins: tuple[str, ...] = DEFAULT_PLUGINS + + #: Additional plugins: importable module names or plugin objects. + #: Plugin objects are the ensemble replacement for conftest files. + extra_plugins: tuple[str | object, ...] = () + + #: Invocation directory; defaults to ``rootpath``, never the implicit cwd. + invocation_dir: pathlib.Path | None = None + + #: Not supported yet; ensemble configs never load conftest files. + load_conftests: bool = False + + #: Stream the terminal plugin writes to, when it is loaded at all. An + #: ensemble must never be given the stdout of whatever is running it, + #: so this is bound at construction rather than redirected around it. + output: TextIO | None = None + + def replace(self, **kw: object) -> ConfigSpec: + return dataclasses.replace(self, **kw) # type: ignore[arg-type] + + def with_plugins(self, *plugins: str | object) -> ConfigSpec: + """Return a spec with the given plugins added. + + Built-in plugin names extend :attr:`plugins`; anything else is + appended to :attr:`extra_plugins`. + """ + from _pytest.config import builtin_plugins + + names = tuple(p for p in plugins if isinstance(p, str) and p in builtin_plugins) + extras = tuple( + p for p in plugins if not (isinstance(p, str) and p in builtin_plugins) + ) + return self.replace( + plugins=self.plugins + names, + extra_plugins=self.extra_plugins + extras, + ) + + def without_plugins(self, *names: str) -> ConfigSpec: + """Return a spec with the given built-in plugin names removed.""" + return self.replace(plugins=tuple(p for p in self.plugins if p not in names)) + + +def _own(value: object) -> ConfigValue: + """Wrap a spec's ini value in a ConfigValue the config may own. + + Mutable values are copied: ``Config.addinivalue_line`` appends to the + cached list, and the cache would otherwise hold the caller's own object, + so a reused (frozen!) spec would grow every time it was configured. + """ + if isinstance(value, ConfigValue): + if isinstance(value.value, list): + value = dataclasses.replace(value, value=list(value.value)) + return value + if isinstance(value, list): + value = list(value) + return ConfigValue(value, origin="file", mode="ini") + + +@contextlib.contextmanager +def configured(spec: ConfigSpec) -> Iterator[Config]: + """Build a parsed *and* configured :class:`Config` from a spec. + + The config is constructed from the spec's explicit values through the + same parse phases a command line invocation uses, but without rootdir + discovery, config file reading, conftest loading, plugin autoloading, + or environment variable consultation. + + On exit, ``pytest_unconfigure`` and the config cleanup stack run. + """ + if spec.rootpath is None: + raise ValueError("ConfigSpec.rootpath is required to build a config") + if spec.load_conftests: + raise NotImplementedError( + "loading conftest files is not supported in ensemble configs yet" + ) + if not spec.rootpath.is_dir(): + raise ValueError(f"ConfigSpec.rootpath is not a directory: {spec.rootpath}") + missing = [name for name in essential_plugins if name not in spec.plugins] + if missing: + raise ValueError( + f"ConfigSpec.plugins must include the essential plugins, missing: {missing}" + ) + + invocation_dir = ( + spec.invocation_dir if spec.invocation_dir is not None else spec.rootpath + ) + pluginmanager = PytestPluginManager() + config = Config( + pluginmanager, + invocation_params=Config.InvocationParams( + args=spec.args, + plugins=spec.extra_plugins or None, + dir=invocation_dir, + ), + ) + # Before anything that can fail: the teardown below reads this, and a + # UsageError raised while parsing args would otherwise surface as a + # KeyError with the real error buried in __context__. + config.stash[config_warnings_key] = [] + try: + for name in spec.plugins: + pluginmanager.import_plugin(name) + for plugin in spec.extra_plugins: + if isinstance(plugin, str): + pluginmanager.import_plugin(plugin) + else: + pluginmanager.register(plugin) + + config.hook.pytest_addhooks.call_historic( + kwargs=dict(pluginmanager=pluginmanager) + ) + inicfg = {name: _own(value) for name, value in spec.inicfg.items()} + config._apply_rootdir( + rootpath=spec.rootpath, + inipath=None, + inicfg=inicfg, + ignored_config_files=(), + ) + config._register_core_ini_options() + + # Mirror the addopts/override-ini handling of Config.parse: without + # it every OverrideIniAction option (--strict-markers, --strict-config, + # ...) and every -o name=value would be parsed into the namespace and + # then silently dropped, so a spec asking for them would get a config + # that quietly ignores them. + args = config._validate_args(config.getini("addopts"), "via addopts config") + args += spec.args + config.known_args_namespace = config._parser.parse_known_args( + args, namespace=copy.copy(config.option) + ) + if overrides := parse_override_ini(config.known_args_namespace.override_ini): + config._inicfg.update(overrides) + config._inicache.clear() + + config._finalize_parse(args, decide_args=False) + if spec.output is not None: + # Must be stashed before configure: the terminal reporter binds + # its stream when it is constructed, and must never bind ours. + config.stash[terminal_file_key] = spec.output + with warnings.catch_warnings(record=True) as caught: + config._do_configure() + config.stash[config_warnings_key].extend(caught) + yield config + finally: + with warnings.catch_warnings(record=True) as caught: + config._ensure_unconfigure() + config.stash[config_warnings_key].extend(caught) diff --git a/src/_pytest/ensemble/results.py b/src/_pytest/ensemble/results.py new file mode 100644 index 00000000000..cf474106532 --- /dev/null +++ b/src/_pytest/ensemble/results.py @@ -0,0 +1,268 @@ +"""Structured results for ensemble runs (EXPERIMENTAL).""" + +from __future__ import annotations + +from collections import Counter +from collections.abc import Sequence +import dataclasses +from typing import TYPE_CHECKING +import warnings as warnings_module + +from _pytest.config import Config +from _pytest.ensemble.config import config_warnings_key +from _pytest.main import Session +from _pytest.nodes import Item +from _pytest.reports import CollectReport +from _pytest.reports import TestReport + + +if TYPE_CHECKING: + from _pytest.pytester import LineMatcher + + +class RunRecorder: + """Plugin recording reports/warnings/deselections on an ensemble config.""" + + name = "ensemble-recorder" + + def __init__(self) -> None: + self.test_reports: list[TestReport] = [] + self.collect_reports: list[CollectReport] = [] + self.warnings: list[warnings_module.WarningMessage] = [] + self.deselected: int = 0 + + def pytest_runtest_logreport(self, report: TestReport) -> None: + self.test_reports.append(report) + + def pytest_collectreport(self, report: CollectReport) -> None: + self.collect_reports.append(report) + + def pytest_warning_recorded( + self, warning_message: warnings_module.WarningMessage + ) -> None: + self.warnings.append(warning_message) + + def pytest_deselected(self, items: Sequence[Item]) -> None: + self.deselected += len(items) + + +def ensure_recorder(config: Config) -> RunRecorder: + """Get the recorder registered on the config, installing it if needed.""" + recorder: RunRecorder | None = config.pluginmanager.get_plugin(RunRecorder.name) + if recorder is None: + recorder = RunRecorder() + config.pluginmanager.register(recorder, RunRecorder.name) + return recorder + + +@dataclasses.dataclass(frozen=True) +class ItemRecord: + """The reports of one test item's run, by phase.""" + + nodeid: str + setup: TestReport | None + call: TestReport | None + teardown: TestReport | None + #: Aggregate category as reported by ``pytest_report_teststatus`` + #: ("passed", "failed", "skipped", "error", "xfailed", "xpassed", ...). + outcome: str + + @property + def reports(self) -> list[TestReport]: + return [r for r in (self.setup, self.call, self.teardown) if r is not None] + + @property + def passed(self) -> bool: + return self.outcome == "passed" + + @property + def failed(self) -> bool: + return self.outcome in ("failed", "error") + + @property + def skipped(self) -> bool: + return self.outcome == "skipped" + + +@dataclasses.dataclass(frozen=True) +class RunRecord: + """Typed, structured results of an ensemble run. + + Everything except :attr:`output` is derived from real report objects — + nothing is scraped from rendered output. + """ + + reports: list[TestReport] + collect_reports: list[CollectReport] + warnings: list[warnings_module.WarningMessage] + deselected: int + by_test: dict[str, ItemRecord] + _by_name: dict[str, ItemRecord] + _counts: dict[str, int] + #: What the terminal plugin rendered, when the ensemble was asked to + #: capture output; empty otherwise. + output: str = "" + #: Why the run stopped before every item had been run - the + #: ``shouldfail``/``shouldstop`` reason - or None if it ran to the end. + stopped: str | None = None + + @property + def stdout(self) -> LineMatcher: + """The rendered output as a matcher, for ``fnmatch_lines`` and friends.""" + # Imported lazily: pytester is a heavyweight module, and the intended + # direction of travel is pytester building on this package, not the + # other way round. + from _pytest.pytester import LineMatcher + + return LineMatcher(self.output.splitlines()) + + @classmethod + def from_recorder(cls, recorder: RunRecorder, *, config: Config) -> RunRecord: + counts: Counter[str] = Counter() + phases: dict[str, dict[str, TestReport]] = {} + categories: dict[str, list[str]] = {} + for report in recorder.test_reports: + phases.setdefault(report.nodeid, {})[report.when] = report + status = config.hook.pytest_report_teststatus(report=report, config=config) + if status is not None: + category = status[0] + else: + # The catch-all status impl lives in the terminal plugin, + # which ensemble configs exclude; mirror its categorization. + category = report.outcome + if category: + counts[category] += 1 + categories.setdefault(report.nodeid, []).append(category) + for collect_report in recorder.collect_reports: + if collect_report.failed: + counts["error"] += 1 + + by_test: dict[str, ItemRecord] = {} + for nodeid, by_when in phases.items(): + cats = categories.get(nodeid, []) + if "error" in cats: + outcome = "error" + elif cats: + outcome = cats[-1] + else: + outcome = "" + by_test[nodeid] = ItemRecord( + nodeid=nodeid, + setup=by_when.get("setup"), + call=by_when.get("call"), + teardown=by_when.get("teardown"), + outcome=outcome, + ) + + name_map: dict[str, ItemRecord | None] = {} + for nodeid, record in by_test.items(): + name = nodeid.rpartition("::")[2] + # Ambiguous bare names are poisoned rather than guessed. + name_map[name] = None if name in name_map else record + by_name = {name: rec for name, rec in name_map.items() if rec is not None} + + # Warnings raised while configuring come before the recorder exists; + # configured() holds on to them so they are not simply lost. + config_warnings = config.stash.get(config_warnings_key, []) + + return cls( + reports=list(recorder.test_reports), + collect_reports=list(recorder.collect_reports), + warnings=[*config_warnings, *recorder.warnings], + deselected=recorder.deselected, + by_test=by_test, + _by_name=by_name, + _counts=dict(counts), + ) + + @property + def collect_errors(self) -> list[CollectReport]: + """The collect reports that failed, if any.""" + return [report for report in self.collect_reports if report.failed] + + def __getitem__(self, name: str) -> ItemRecord: + if name in self.by_test: + return self.by_test[name] + if name in self._by_name: + return self._by_name[name] + raise KeyError( + f"no unambiguous test named {name!r}; known: {sorted(self.by_test)}" + ) + + def outcomes(self) -> dict[str, int]: + """Outcome category counts, keyed like the terminal summary + ("passed", "failed", "errors", "skipped", "xfailed", "xpassed").""" + counts = dict(self._counts) + if "error" in counts: + counts["errors"] = counts.pop("error") + return counts + + def assert_outcomes( + self, + *, + passed: int = 0, + skipped: int = 0, + failed: int = 0, + errors: int = 0, + xpassed: int = 0, + xfailed: int = 0, + warnings: int | None = None, + deselected: int | None = None, + ) -> None: + """Assert the run produced exactly the given outcome counts. + + Signature-compatible with :meth:`RunResult.assert_outcomes + <_pytest.pytester.RunResult.assert_outcomes>`. + """ + __tracebackhide__ = True + from _pytest.pytester_assertions import assert_outcomes + + outcomes = self.outcomes() + outcomes["warnings"] = len(self.warnings) + outcomes["deselected"] = self.deselected + assert_outcomes( + outcomes, + passed=passed, + skipped=skipped, + failed=failed, + errors=errors, + xpassed=xpassed, + xfailed=xfailed, + warnings=warnings, + deselected=deselected, + ) + + +def run_items(session: Session, items: Sequence[Item] | None = None) -> RunRecord: + """Run the session's items and return the structured results. + + With no explicit *items*, this goes through ``pytest_runtestloop``, the + hook a real run uses - so plugins wrapping it see what they expect, and + the terminal reporter gets to write its deferred final progress fill. + That hook signals early exit by raising ``Failed``/``Interrupted`` for + ``wrap_session`` to catch; an ensemble has no such wrapper, so the + reason is caught here and recorded on :attr:`RunRecord.stopped`. + + Given an explicit subset, the protocol hook is driven directly - the + loop hook has no way to express "just these" - while still honouring + ``shouldfail``/``shouldstop`` between items. + """ + recorder = ensure_recorder(session.config) + stopped: str | None = None + if items is None: + try: + session.config.hook.pytest_runtestloop(session=session) + except (Session.Failed, Session.Interrupted) as exc: + stopped = str(exc) + else: + for i, item in enumerate(items): + nextitem = items[i + 1] if i + 1 < len(items) else None + item.ihook.pytest_runtest_protocol(item=item, nextitem=nextitem) + if session.shouldfail: + stopped = str(session.shouldfail) + break + if session.shouldstop: + stopped = str(session.shouldstop) + break + record = RunRecord.from_recorder(recorder, config=session.config) + return dataclasses.replace(record, stopped=stopped) diff --git a/src/_pytest/faulthandler.py b/src/_pytest/faulthandler.py index 080cf583813..7fea9813818 100644 --- a/src/_pytest/faulthandler.py +++ b/src/_pytest/faulthandler.py @@ -15,6 +15,11 @@ fault_handler_stderr_fd_key = StashKey[int]() +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def pytest_addoption(parser: Parser) -> None: help_timeout = ( "Dump the traceback of all threads if a test takes " diff --git a/src/_pytest/fixtures.py b/src/_pytest/fixtures.py index 44c304905e7..e19279a6109 100644 --- a/src/_pytest/fixtures.py +++ b/src/_pytest/fixtures.py @@ -2286,6 +2286,13 @@ def parsefactories( holderobj = cast(object, node_or_obj.obj) # type: ignore[attr-defined] effective_node = node_or_obj + # An object that says it defines no fixtures is taken at its word. + # Scanning one means reading every attribute it has and validating + # each, which is wasted work for the many plugins - reporters, + # loggers, managers - that hold no fixtures at all. + if safe_getattr(holderobj, "__pytest_no_fixtures__", False): + return + # Avoid accessing `@property` (and other descriptors) when iterating fixtures. holderobj_tp: type | types.ModuleType if not safe_isclass(holderobj) and not isinstance(holderobj, types.ModuleType): diff --git a/src/_pytest/helpconfig.py b/src/_pytest/helpconfig.py index 1bceb05558c..b053c770e6e 100644 --- a/src/_pytest/helpconfig.py +++ b/src/_pytest/helpconfig.py @@ -19,6 +19,11 @@ import pytest +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + class HelpAction(argparse.Action): """An argparse Action that will raise a PrintHelp exception in order to skip the rest of the argument parsing when --help is passed. diff --git a/src/_pytest/legacypath.py b/src/_pytest/legacypath.py index e1fc38d19ed..19315db8214 100644 --- a/src/_pytest/legacypath.py +++ b/src/_pytest/legacypath.py @@ -38,6 +38,11 @@ import pexpect +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + @final class Testdir: """ diff --git a/src/_pytest/main.py b/src/_pytest/main.py index 1b337e20c7e..3fa44c85204 100644 --- a/src/_pytest/main.py +++ b/src/_pytest/main.py @@ -54,6 +54,11 @@ from _pytest.fixtures import FixtureManager +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def pytest_addoption(parser: Parser) -> None: group = parser.getgroup("general") group._addoption( # private to use reserved lower-case short option diff --git a/src/_pytest/mark/__init__.py b/src/_pytest/mark/__init__.py index 73354506df3..c1d2ec8e976 100644 --- a/src/_pytest/mark/__init__.py +++ b/src/_pytest/mark/__init__.py @@ -47,6 +47,11 @@ old_mark_config_key = StashKey[Config | None]() +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def param( *values: object, marks: MarkDecorator | Collection[MarkDecorator | Mark] = (), diff --git a/src/_pytest/pastebin.py b/src/_pytest/pastebin.py index e6a1430220a..068bc2a6f0c 100644 --- a/src/_pytest/pastebin.py +++ b/src/_pytest/pastebin.py @@ -19,6 +19,11 @@ pastebinfile_key = StashKey[IO[bytes]]() +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def pytest_addoption(parser: Parser) -> None: group = parser.getgroup("terminal reporting") group.addoption( diff --git a/src/_pytest/python.py b/src/_pytest/python.py index be0ea5b4d05..afdd016e050 100644 --- a/src/_pytest/python.py +++ b/src/_pytest/python.py @@ -93,6 +93,11 @@ ) +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def pytest_addoption(parser: Parser) -> None: parser.addini( "python_files", @@ -766,10 +771,22 @@ def _get_first_non_fixture_func(obj: object, names: Iterable[str]) -> object | N class Class(PyCollector): """Collector for test methods (and nested classes) in a Python class.""" + #: Explicitly provided class object, taking precedence over looking up + #: ``name`` on the parent's object (experimental). + _given_obj: type | None = None + @classmethod def from_parent(cls, parent, *, name, obj=None, **kw) -> Self: # type: ignore[override] """The public constructor.""" - return super().from_parent(name=name, parent=parent, **kw) + node: Self = super().from_parent(name=name, parent=parent, **kw) + if obj is not None: + node._given_obj = obj + return node + + def _getobj(self): + if self._given_obj is not None: + return self._given_obj + return super()._getobj() def newinstance(self): return self.obj() diff --git a/src/_pytest/reports.py b/src/_pytest/reports.py index 72fe10e96c7..ea75f5e5fad 100644 --- a/src/_pytest/reports.py +++ b/src/_pytest/reports.py @@ -46,6 +46,11 @@ from _pytest.runner import CallInfo +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def getworkerinfoline(node): try: return node._workerinfocache diff --git a/src/_pytest/runner.py b/src/_pytest/runner.py index cb723d134b9..eb622938e68 100644 --- a/src/_pytest/runner.py +++ b/src/_pytest/runner.py @@ -48,6 +48,11 @@ # pytest plugin hooks. +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def pytest_addoption(parser: Parser) -> None: group = parser.getgroup("terminal reporting", "Reporting", after="general") group.addoption( diff --git a/src/_pytest/setuponly.py b/src/_pytest/setuponly.py index 7e6b46bcdb4..087206aceb2 100644 --- a/src/_pytest/setuponly.py +++ b/src/_pytest/setuponly.py @@ -4,7 +4,6 @@ from _pytest._io.saferepr import saferepr from _pytest.config import Config -from _pytest.config import ExitCode from _pytest.config.argparsing import Parser from _pytest.fixtures import FixtureDef from _pytest.fixtures import SubRequest @@ -12,6 +11,11 @@ import pytest +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def pytest_addoption(parser: Parser) -> None: group = parser.getgroup("debugconfig") group.addoption( @@ -92,7 +96,7 @@ def _show_fixture_action( @pytest.hookimpl(tryfirst=True) -def pytest_cmdline_main(config: Config) -> int | ExitCode | None: +def pytest_configure(config: Config) -> None: + # See the note in setupplan.py: configure time, not cmdline time. if config.option.setuponly: config.option.setupshow = True - return None diff --git a/src/_pytest/setupplan.py b/src/_pytest/setupplan.py index 4e124cce243..d47acb02e40 100644 --- a/src/_pytest/setupplan.py +++ b/src/_pytest/setupplan.py @@ -1,13 +1,17 @@ from __future__ import annotations from _pytest.config import Config -from _pytest.config import ExitCode from _pytest.config.argparsing import Parser from _pytest.fixtures import FixtureDef from _pytest.fixtures import SubRequest import pytest +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def pytest_addoption(parser: Parser) -> None: group = parser.getgroup("debugconfig") group.addoption( @@ -32,8 +36,10 @@ def pytest_fixture_setup( @pytest.hookimpl(tryfirst=True) -def pytest_cmdline_main(config: Config) -> int | ExitCode | None: +def pytest_configure(config: Config) -> None: + # Normalizing at configure time rather than in pytest_cmdline_main means + # it also applies to programmatically constructed configs, which are + # configured but never go through the command line entry point. if config.option.setupplan: config.option.setuponly = True config.option.setupshow = True - return None diff --git a/src/_pytest/skipping.py b/src/_pytest/skipping.py index aa97e3b7dc6..90c3cafe992 100644 --- a/src/_pytest/skipping.py +++ b/src/_pytest/skipping.py @@ -26,6 +26,11 @@ from _pytest.stash import StashKey +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def pytest_addoption(parser: Parser) -> None: group = parser.getgroup("general") group.addoption( diff --git a/src/_pytest/stepwise.py b/src/_pytest/stepwise.py index 8901540eb59..ce803753419 100644 --- a/src/_pytest/stepwise.py +++ b/src/_pytest/stepwise.py @@ -20,6 +20,11 @@ STEPWISE_CACHE_DIR = "cache/stepwise" +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def pytest_addoption(parser: Parser) -> None: group = parser.getgroup("general") group.addoption( diff --git a/src/_pytest/terminal.py b/src/_pytest/terminal.py index 852153b9215..e54ba7aa3ee 100644 --- a/src/_pytest/terminal.py +++ b/src/_pytest/terminal.py @@ -52,6 +52,7 @@ from _pytest.reports import BaseReport from _pytest.reports import CollectReport from _pytest.reports import TestReport +from _pytest.stash import StashKey if TYPE_CHECKING: @@ -81,6 +82,11 @@ ] +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + class MoreQuietAction(argparse.Action): """A modified copy of the argparse count action which counts down and updates the legacy quiet attribute at the same time. @@ -293,10 +299,17 @@ def pytest_addoption(parser: Parser) -> None: ) +#: Stream the terminal reporter should write to, if it must not be the +#: process stdout. Set on the config's stash *before* it is configured; a +#: nested run has to be given its own sink rather than inheriting the +#: stdout of whatever is running it. +terminal_file_key = StashKey[TextIO]() + + def pytest_configure(config: Config) -> None: # Eagerly validate the value; it is only read lazily during reporting. config.getini("console_output_style") - reporter = TerminalReporter(config, sys.stdout) + reporter = TerminalReporter(config, config.stash.get(terminal_file_key, sys.stdout)) config.pluginmanager.register(reporter, "terminalreporter") if config.option.debug or config.option.traceconfig: @@ -384,6 +397,8 @@ def get_location(self, config: Config) -> str | None: @final class TerminalReporter: + __pytest_no_fixtures__ = True + def __init__(self, config: Config, file: TextIO | None = None) -> None: import _pytest.config @@ -419,9 +434,13 @@ def _determine_show_progress_info( ) -> Literal["progress", "count", "times", False]: """Return whether we should display progress information based on the current config.""" # do not show progress if we are not capturing output (#3038) unless explicitly - # overridden by progress-even-when-capture-no + # overridden by progress-even-when-capture-no. + # Writing to a stream of our own counts as captured: the concern is + # progress interleaving with test output, and nothing else is writing + # there. if ( self.config.getoption("capture", "no") == "no" + and self.config.stash.get(terminal_file_key, None) is None and self.config.getini("console_output_style") != "progress-even-when-capture-no" ): diff --git a/src/_pytest/threadexception.py b/src/_pytest/threadexception.py index eb57783be26..11541e067cf 100644 --- a/src/_pytest/threadexception.py +++ b/src/_pytest/threadexception.py @@ -24,6 +24,11 @@ from exceptiongroup import ExceptionGroup +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + class ThreadExceptionMeta(NamedTuple): msg: str cause_msg: str diff --git a/src/_pytest/unittest.py b/src/_pytest/unittest.py index d5286af1470..685244f8dcd 100644 --- a/src/_pytest/unittest.py +++ b/src/_pytest/unittest.py @@ -56,6 +56,11 @@ ) +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def pytest_pycollect_makeitem( collector: Module | Class, name: str, obj: object ) -> UnitTestCase | None: @@ -379,13 +384,9 @@ def addDuration(self, testcase: unittest.TestCase, elapsed: float) -> None: pass def runtest(self) -> None: - from _pytest.debugging import maybe_wrap_pytest_function_for_tracing - testcase = self.instance assert testcase is not None - maybe_wrap_pytest_function_for_tracing(self) - # Let the unittest framework handle async functions. if is_async_function(self.obj): testcase(result=self) @@ -399,7 +400,7 @@ def runtest(self) -> None: # We need to consider if the test itself is skipped, or the whole class. assert isinstance(self.parent, UnitTestCase) skipped = _is_skipped(self.obj) or _is_skipped(self.parent.obj) - if self.config.getoption("usepdb") and not skipped: + if self.config.getoption("usepdb", False) and not skipped: self._explicit_tearDown = testcase.tearDown setattr(testcase, "tearDown", lambda *args: None) diff --git a/src/_pytest/unraisableexception.py b/src/_pytest/unraisableexception.py index 6c092fb6bd3..73c68aa1e2c 100644 --- a/src/_pytest/unraisableexception.py +++ b/src/_pytest/unraisableexception.py @@ -28,6 +28,11 @@ gc_collect_iterations_key = StashKey[int]() +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + def gc_collect_harder(iterations: int) -> None: for _ in range(iterations): gc.collect() diff --git a/src/_pytest/warnings.py b/src/_pytest/warnings.py index 2c721d6e3c9..4ca50af9c37 100644 --- a/src/_pytest/warnings.py +++ b/src/_pytest/warnings.py @@ -15,6 +15,11 @@ import pytest +#: This plugin defines no fixtures, so the fixture manager need not read +#: every attribute it has looking for them. +__pytest_no_fixtures__ = True + + @contextmanager def catch_warnings_for_item( config: Config, diff --git a/testing/python/collect.py b/testing/python/collect.py index c9023f98595..dc37031ede8 100644 --- a/testing/python/collect.py +++ b/testing/python/collect.py @@ -2,13 +2,15 @@ from __future__ import annotations import os +from pathlib import Path import sys import textwrap from typing import Any import _pytest._code from _pytest.config import ExitCode -from _pytest.main import Session +from _pytest.ensemble import collect_tests +from _pytest.ensemble import run_tests from _pytest.monkeypatch import MonkeyPatch from _pytest.nodes import Collector from _pytest.pytester import Pytester @@ -180,54 +182,63 @@ def __new__(self): ] ) - def test_class_subclassobject(self, pytester: Pytester) -> None: - pytester.getmodulecol( + def test_class_subclassobject(self, tmp_path: Path) -> None: + class test: + pass + + assert collect_tests(test, rootpath=tmp_path) == [] + + def test_class_from_parent_without_obj_resolves_by_name( + self, pytester: Pytester + ) -> None: + """Without an explicit obj, the class is looked up on the parent's object.""" + modcol = pytester.getmodulecol( + """ + class TestGroup: + def test_method(self): + pass """ - class test(object): - pass - """ ) - result = pytester.runpytest() - result.stdout.fnmatch_lines(["*collected 0*"]) + cls = pytest.Class.from_parent(modcol, name="TestGroup") + assert cls.obj is modcol.obj.TestGroup - def test_static_method(self, pytester: Pytester) -> None: + def test_static_method(self, tmp_path: Path) -> None: """Support for collecting staticmethod tests (#2528, #2699)""" - pytester.getmodulecol( - """ - import pytest - class Test(object): - @staticmethod - def test_something(): - pass - @pytest.fixture - def fix(self): - return 1 + class Test: + @staticmethod + def test_something(): + pass - @staticmethod - def test_fix(fix): - assert fix == 1 - """ - ) - result = pytester.runpytest() - result.stdout.fnmatch_lines(["*collected 2 items*", "*2 passed in*"]) + @pytest.fixture + def fix(self): + return 1 - def test_setup_teardown_class_as_classmethod(self, pytester: Pytester) -> None: - pytester.makepyfile( - test_mod1=""" - class TestClassMethod(object): - @classmethod - def setup_class(cls): - pass - def test_1(self): - pass - @classmethod - def teardown_class(cls): - pass - """ - ) - result = pytester.runpytest() - result.stdout.fnmatch_lines(["*1 passed*"]) + @staticmethod + def test_fix(fix): + assert fix == 1 + + record = run_tests(Test, rootpath=tmp_path) + record.assert_outcomes(passed=2) + + def test_setup_teardown_class_as_classmethod(self, tmp_path: Path) -> None: + events = [] + + class TestClassMethod: + @classmethod + def setup_class(cls): + events.append("setup") + + def test_1(self): + pass + + @classmethod + def teardown_class(cls): + events.append("teardown") + + record = run_tests(TestClassMethod, rootpath=tmp_path) + record.assert_outcomes(passed=1) + assert events == ["setup", "teardown"] def test_issue1035_obj_has_getattr(self, pytester: Pytester) -> None: modcol = pytester.getmodulecol( @@ -384,32 +395,32 @@ def __call__(self, tmp_path): ) @staticmethod - def make_function(pytester: Pytester, **kwargs: Any) -> Any: - from _pytest.fixtures import FixtureManager - - config = pytester.parseconfigure() - session = Session.from_config(config) - session._fixturemanager = FixtureManager(session) + def make_function(tmp_path: Path, **kwargs: Any) -> Any: + from _pytest.ensemble import ConfigSpec + from _pytest.ensemble import configured + from _pytest.ensemble import running_session - return pytest.Function.from_parent(parent=session, **kwargs) + with configured(ConfigSpec(rootpath=tmp_path)) as config: + with running_session(config) as session: + return pytest.Function.from_parent(parent=session, **kwargs) - def test_function_equality(self, pytester: Pytester) -> None: + def test_function_equality(self, tmp_path: Path) -> None: def func1(): pass def func2(): pass - f1 = self.make_function(pytester, name="name", callobj=func1) + f1 = self.make_function(tmp_path, name="name", callobj=func1) assert f1 == f1 f2 = self.make_function( - pytester, name="name", callobj=func2, originalname="foobar" + tmp_path, name="name", callobj=func2, originalname="foobar" ) assert f1 != f2 - def test_repr_produces_actual_test_id(self, pytester: Pytester) -> None: + def test_repr_produces_actual_test_id(self, tmp_path: Path) -> None: f = self.make_function( - pytester, name=r"test[\xe5]", callobj=self.test_repr_produces_actual_test_id + tmp_path, name=r"test[\xe5]", callobj=self.test_repr_produces_actual_test_id ) assert repr(f) == r"" diff --git a/testing/python/fixtures.py b/testing/python/fixtures.py index 95fe5c389b8..f68fe3be39f 100644 --- a/testing/python/fixtures.py +++ b/testing/python/fixtures.py @@ -9,6 +9,7 @@ from _pytest.compat import getfuncargnames from _pytest.config import ExitCode +from _pytest.ensemble import run_tests from _pytest.fixtures import deduplicate_names from _pytest.fixtures import ParamValueKey from _pytest.fixtures import TopRequest @@ -1182,50 +1183,38 @@ def test_request_fixturenames_dynamic_fixture(self, pytester: Pytester) -> None: result = pytester.runpytest("-vv") result.stdout.fnmatch_lines(["*1 passed*"]) - def test_setupdecorator_and_xunit(self, pytester: Pytester) -> None: - pytester.makepyfile( - """ - import pytest - - values = [] + def test_setupdecorator_and_xunit(self, tmp_path: Path) -> None: + values: list[str] = [] - @pytest.fixture(scope='module', autouse=True) - def setup_module(): - values.append("module") + @pytest.fixture(scope="module", autouse=True) + def setup_module(): + values.append("module") - @pytest.fixture(autouse=True) - def setup_function(): - values.append("function") + @pytest.fixture(autouse=True) + def setup_function(): + values.append("function") - def test_func(): - pass + def test_func(): + pass - class TestClass: - @pytest.fixture(scope="class", autouse=True) - @classmethod - def setup_class(cls): - values.append("class") + class TestClass: + @pytest.fixture(scope="class", autouse=True) + @classmethod + def setup_class(cls): + values.append("class") - @pytest.fixture(autouse=True) - def setup_method(self): - values.append("method") + @pytest.fixture(autouse=True) + def setup_method(self): + values.append("method") - def test_method(self): - pass + def test_method(self): + pass - def test_all(): - assert values == [ - "module", - "function", - "class", - "function", - "method", - "function", - ] - """ + record = run_tests( + setup_module, setup_function, test_func, TestClass, rootpath=tmp_path ) - reprec = pytester.inline_run("-v") - reprec.assertoutcome(passed=3) + record.assert_outcomes(passed=2) + assert values == ["module", "function", "class", "function", "method"] def test_fixtures_sub_subdir_normalize_sep(self, pytester: Pytester) -> None: # this tests that normalization of nodeids takes place diff --git a/testing/python/metafunc.py b/testing/python/metafunc.py index 566071d47a1..89577901141 100644 --- a/testing/python/metafunc.py +++ b/testing/python/metafunc.py @@ -5,6 +5,7 @@ from collections.abc import Sequence import dataclasses import itertools +from pathlib import Path import re import sys import textwrap @@ -19,6 +20,8 @@ from _pytest import python from _pytest.compat import getfuncargnames from _pytest.compat import NOTSET +from _pytest.ensemble import build_module +from _pytest.ensemble import run_tests from _pytest.outcomes import fail from _pytest.outcomes import Failed from _pytest.pytester import Pytester @@ -1302,29 +1305,23 @@ def test_method(self, metafunc, pytestconfig): result = pytester.runpytest(p, "-v") result.assert_outcomes(passed=2) - def test_two_functions(self, pytester: Pytester) -> None: - p = pytester.makepyfile( - """ - def pytest_generate_tests(metafunc): - metafunc.parametrize('arg1', [10, 20], ids=['0', '1']) + def test_two_functions(self, tmp_path: Path) -> None: + def pytest_generate_tests(metafunc): + metafunc.parametrize("arg1", [10, 20], ids=["0", "1"]) - def test_func1(arg1): - assert arg1 == 10 + def test_func1(arg1): + assert arg1 == 10 - def test_func2(arg1): - assert arg1 in (10, 20) - """ - ) - result = pytester.runpytest("-v", p) - result.stdout.fnmatch_lines( - [ - "*test_func1*0*PASS*", - "*test_func1*1*FAIL*", - "*test_func2*PASS*", - "*test_func2*PASS*", - "*1 failed, 3 passed*", - ] + def test_func2(arg1): + assert arg1 in (10, 20) + + module = build_module( + "test_two_functions", pytest_generate_tests, test_func1, test_func2 ) + record = run_tests(module, rootpath=tmp_path) + record.assert_outcomes(passed=3, failed=1) + assert record["test_two_functions.py::test_func1[0]"].passed + assert record["test_two_functions.py::test_func1[1]"].failed def test_noself_in_method(self, pytester: Pytester) -> None: p = pytester.makepyfile( diff --git a/testing/test_debugging.py b/testing/test_debugging.py index e2b1d2b8527..e03e52e5aac 100644 --- a/testing/test_debugging.py +++ b/testing/test_debugging.py @@ -1220,6 +1220,18 @@ def test_func_kw(myparam, request, func="func_kw"): TestPDB.flush(child) +def test_trace_ignores_non_function_items(pytester: Pytester) -> None: + """--trace only wraps python function items; others run untouched.""" + pytester.maketxtfile( + test_doc=""" + >>> 1 + 1 + 2 + """ + ) + result = pytester.runpytest("--trace", "--doctest-glob=*.txt") + result.assert_outcomes(passed=1) + + def test_trace_after_runpytest(pytester: Pytester) -> None: """Test that debugging's pytest_configure is reentrant.""" p1 = pytester.makepyfile( diff --git a/testing/test_ensemble.py b/testing/test_ensemble.py new file mode 100644 index 00000000000..552386541de --- /dev/null +++ b/testing/test_ensemble.py @@ -0,0 +1,935 @@ +"""Tests for the experimental _pytest.ensemble API.""" + +from __future__ import annotations + +from collections.abc import Generator +from contextlib import ExitStack +import os +from pathlib import Path +import sys +import types +import unittest +import warnings + +from _pytest._io import TerminalWriter +from _pytest.config import Config +from _pytest.config import ExitCode +from _pytest.config.exceptions import UsageError +from _pytest.config.findpaths import ConfigValue +from _pytest.ensemble import build_module +from _pytest.ensemble import collect_tests +from _pytest.ensemble import ConfigSpec +from _pytest.ensemble import configured +from _pytest.ensemble import Ensemble +from _pytest.ensemble import EnsembleModule +from _pytest.ensemble import module_from_path +from _pytest.ensemble import run_tests +from _pytest.ensemble import running_session +from _pytest.nodes import Collector +from _pytest.pytester import Pytester +import pytest + + +class TestConfigSpec: + def test_rootpath_required(self) -> None: + with ( + pytest.raises(ValueError, match="rootpath is required"), + ExitStack() as stack, + ): + stack.enter_context(configured(ConfigSpec())) + + def test_rootpath_must_be_dir(self, tmp_path: Path) -> None: + with pytest.raises(ValueError, match="not a directory"), ExitStack() as stack: + stack.enter_context(configured(ConfigSpec(rootpath=tmp_path / "missing"))) + + def test_essential_plugins_validated(self, tmp_path: Path) -> None: + with ( + pytest.raises(ValueError, match=r"essential plugins.*runner"), + ExitStack() as stack, + ): + stack.enter_context( + configured(ConfigSpec(rootpath=tmp_path, plugins=("python", "mark"))) + ) + + def test_load_conftests_unsupported(self, tmp_path: Path) -> None: + with pytest.raises(NotImplementedError, match="conftest"), ExitStack() as stack: + stack.enter_context( + configured(ConfigSpec(rootpath=tmp_path, load_conftests=True)) + ) + + def test_configured_basics(self, tmp_path: Path) -> None: + spec = ConfigSpec(rootpath=tmp_path, args=("-k", "nothing")) + with configured(spec) as config: + assert config.rootpath == tmp_path + assert config.inipath is None + assert config.args == [] + assert config.args_source is Config.ArgsSource.SPEC + assert config.getoption("keyword") == "nothing" + assert config.invocation_params.dir == tmp_path + # paired teardown ran + assert not config._configured + + def test_inicfg_is_authoritative(self, tmp_path: Path) -> None: + spec = ConfigSpec(rootpath=tmp_path, inicfg={"usefixtures": ["myfix"]}) + with configured(spec) as config: + assert config.getini("usefixtures") == ["myfix"] + + def test_excluded_plugins_absent(self, tmp_path: Path) -> None: + with configured(ConfigSpec(rootpath=tmp_path)) as config: + assert config.pluginmanager.get_plugin("capturemanager") is None + assert config.pluginmanager.get_plugin("terminalreporter") is None + assert not config.pluginmanager.hasplugin("capture") + assert not config.pluginmanager.hasplugin("terminal") + assert not config.pluginmanager.hasplugin("cacheprovider") + + def test_spec_derivation_helpers(self) -> None: + spec = ConfigSpec() + derived = spec.with_plugins("capture").without_plugins("unittest") + assert "capture" in derived.plugins + assert "unittest" not in derived.plugins + # frozen: original unchanged + assert "capture" not in spec.plugins + + @pytest.mark.parametrize( + "spec_kwargs", + [ + pytest.param({"args": ("--strict-markers",)}, id="override-ini-action"), + pytest.param({"args": ("-o", "strict_markers=true")}, id="dash-o"), + pytest.param( + {"inicfg": {"addopts": "--strict-markers"}}, id="addopts-in-inicfg" + ), + ], + ) + def test_ini_overrides_are_applied( + self, tmp_path: Path, spec_kwargs: dict[str, object] + ) -> None: + """Options that override ini values must not be silently dropped.""" + with configured(ConfigSpec(rootpath=tmp_path, **spec_kwargs)) as config: # type: ignore[arg-type] + assert config.getini("strict_markers") is True + + def test_ini_overrides_are_not_invented(self, tmp_path: Path) -> None: + with configured(ConfigSpec(rootpath=tmp_path)) as config: + assert config.getini("strict_markers") is None + + def test_unregistered_marker_is_strict(self, tmp_path: Path) -> None: + """The end the overrides serve: --strict-markers actually bites. + + Enforcement is asserted through ``-m`` expression validation rather + than a ``@pytest.mark.unregistered`` decorator, because decorators in + the enclosing test body are resolved against the *host* config at + decoration time, long before the ensemble config exists. + """ + + def test_fn() -> None: ... + + spec = ConfigSpec( + rootpath=tmp_path, args=("--strict-markers", "-m", "nowhere_registered") + ) + with pytest.raises(UsageError, match="Unknown marker"): + run_tests(test_fn, spec=spec) + + def test_unregistered_marker_allowed_without_strict(self, tmp_path: Path) -> None: + def test_fn() -> None: ... + + spec = ConfigSpec(rootpath=tmp_path, args=("-m", "nowhere_registered")) + run_tests(test_fn, spec=spec).assert_outcomes(deselected=1) + + def test_setup_plan_normalizes_options(self, tmp_path: Path) -> None: + """--setup-plan implies --setup-only/--setup-show. + + The implication used to live in ``pytest_cmdline_main``, which an + ensemble never reaches, so the flag was accepted and then ignored. + """ + spec = ConfigSpec(rootpath=tmp_path, args=("--setup-plan",)).with_plugins( + "setuponly", "setupplan" + ) + with configured(spec) as config: + assert config.getoption("setuponly") is True + assert config.getoption("setupshow") is True + + def test_setup_plan_skips_the_call_phase(self, tmp_path: Path) -> None: + ran = [] + + def test_fn() -> None: + ran.append(1) + + spec = ConfigSpec(rootpath=tmp_path, args=("--setup-plan",)).with_plugins( + "setuponly", "setupplan" + ) + record = run_tests(test_fn, spec=spec) + assert ran == [] + assert record["test_fn"].call is None + + @pytest.mark.parametrize("wrap", [False, True], ids=["plain-list", "ConfigValue"]) + def test_spec_survives_reuse(self, tmp_path: Path, wrap: bool) -> None: + """A frozen spec must not grow when it is configured repeatedly. + + ``addinivalue_line`` appends to the cached list, so handing the + caller's own list to the config made every reuse accumulate. + """ + value: object = ["mine: a marker"] + if wrap: + value = ConfigValue(value, origin="file", mode="ini") + spec = ConfigSpec(rootpath=tmp_path, inicfg={"markers": value}) + + seen = [] + for _ in range(3): + with configured(spec) as config: + seen.append(len(config.getini("markers"))) + stored = spec.inicfg["markers"] + raw = stored.value if isinstance(stored, ConfigValue) else stored + assert len(raw) == 1 # type: ignore[arg-type] + assert len(set(seen)) == 1, f"config saw growing marker lists: {seen}" + + def test_assertion_explanation_is_the_ensemble_s(self, tmp_path: Path) -> None: + """The failure explanation must be configured by the ensemble. + + ``assertion.util._reprcompare`` is process-global; without the + assertion plugin it stays bound to whatever the host installed, so + an explanation is still produced and the test goes green while + silently reflecting the host's configuration. + """ + seen: list[tuple[str, object, object]] = [] + + class Comparer: + def pytest_assertrepr_compare( + self, op: str, left: object, right: object + ) -> list[str]: + seen.append((op, left, right)) + return ["ensemble-owned explanation"] + + def test_fails() -> None: + # deliberately false, so the comparison hook has something to explain + assert 1 == 2 # type: ignore[comparison-overlap] + + spec = ConfigSpec(rootpath=tmp_path, extra_plugins=(Comparer(),)) + record = run_tests(test_fails, spec=spec) + record.assert_outcomes(failed=1) + assert seen == [("==", 1, 2)] + assert "ensemble-owned explanation" in record["test_fails"].call.longreprtext # type: ignore[union-attr] + + def test_extra_plugin_by_name(self, tmp_path: Path) -> None: + """String entries in extra_plugins are imported, objects registered.""" + spec = ConfigSpec(rootpath=tmp_path, extra_plugins=("_pytest.setuponly",)) + with configured(spec) as config: + assert config.pluginmanager.get_plugin("_pytest.setuponly") is not None + + +class TestTerminalLessConfig: + """A config without the terminal plugin still has to render sometimes.""" + + def test_get_terminal_writer_falls_back(self, tmp_path: Path) -> None: + with configured(ConfigSpec(rootpath=tmp_path)) as config: + assert config.pluginmanager.get_plugin("terminalreporter") is None + assert isinstance(config.get_terminal_writer(), TerminalWriter) + + def test_pdb_on_failure_does_not_crash( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """--pdb used to die reaching into a terminalreporter that is absent.""" + entered: list[object] = [] + + def fake_post_mortem(tb_or_exc: object) -> None: + entered.append(tb_or_exc) + + monkeypatch.setattr("_pytest.debugging.post_mortem", fake_post_mortem) + + def test_fails() -> None: + raise AssertionError("boom") + + spec = ConfigSpec(rootpath=tmp_path, args=("--pdb",)).with_plugins("debugging") + record = run_tests(test_fails, spec=spec) + record.assert_outcomes(failed=1) + assert len(entered) == 1 + + +class TestCapturedOutput: + def test_rendered_report_is_captured(self, tmp_path: Path) -> None: + def test_ok() -> None: ... + + def test_bad() -> None: + left = 1 + assert left == 2 + + record = run_tests(test_ok, test_bad, rootpath=tmp_path, capture_output=True) + record.assert_outcomes(passed=1, failed=1) + record.stdout.fnmatch_lines( + [ + "*test session starts*", + "*FAILURES*", + "*1 failed, 1 passed*", + ] + ) + # the structured view still agrees with the rendered one + assert record["test_bad"].failed + + def test_output_never_reaches_the_outer_stdout( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """An ensemble must not be handed the stdout of whatever runs it.""" + + class Tripwire: + def write(self, text: str) -> int: + raise AssertionError(f"ensemble wrote to the outer stdout: {text!r}") + + def flush(self) -> None: ... + + def isatty(self) -> bool: + return False + + def test_ok() -> None: ... + + monkeypatch.setattr(sys, "stdout", Tripwire()) + record = run_tests(test_ok, rootpath=tmp_path, capture_output=True) + assert "1 passed" in record.output + + def test_not_captured_by_default(self, tmp_path: Path) -> None: + def test_ok() -> None: ... + + record = run_tests(test_ok, rootpath=tmp_path) + assert record.output == "" + assert record.stdout.lines == [] + + +def test_no_fixtures_optout_is_truthful() -> None: + """Every plugin claiming to hold no fixtures must actually hold none. + + The opt-out makes the fixture manager skip a plugin entirely, so a + fixture added to one of these later would silently never be collected. + This is the guard against that. + """ + import inspect + + from _pytest.config import get_config + from _pytest.fixtures import FixtureFunctionDefinition + + config = get_config([]) + config.pluginmanager.import_plugin("python") + + lying = [] + for plugin in config.pluginmanager.get_plugins(): + if plugin is None or not getattr(plugin, "__pytest_no_fixtures__", False): + continue + holder = plugin if inspect.ismodule(plugin) else type(plugin) + for name in dir(holder): + if type(getattr(holder, name, None)) is FixtureFunctionDefinition: + lying.append( + f"{getattr(plugin, '__name__', type(plugin).__name__)}.{name}" + ) + assert lying == [], f"declared __pytest_no_fixtures__ but define fixtures: {lying}" + + +class TestRunLoop: + def test_goes_through_pytest_runtestloop(self, tmp_path: Path) -> None: + """Plugins wrapping the loop hook must see an ensemble run.""" + seen: list[str] = [] + + class LoopWatcher: + @pytest.hookimpl(wrapper=True) + def pytest_runtestloop( + self, session: object + ) -> Generator[None, object, object]: + seen.append("loop") + return (yield) + + def test_ok() -> None: ... + + spec = ConfigSpec(rootpath=tmp_path, extra_plugins=(LoopWatcher(),)) + run_tests(test_ok, spec=spec).assert_outcomes(passed=1) + assert seen == ["loop"] + + def test_progress_column_is_rendered(self, tmp_path: Path) -> None: + """The terminal's deferred final fill needs the loop hook, and the + progress column needs somewhere safe to write.""" + + def test_a() -> None: ... + + def test_b() -> None: ... + + record = run_tests(test_a, test_b, rootpath=tmp_path, capture_output=True) + record.stdout.fnmatch_lines(["*test_ensemble.py ..*[[]100%[]]*"]) + + +class TestRealSources: + """Sources that exist on disk keep their own identity.""" + + def test_imported_module_keeps_its_path(self, tmp_path: Path) -> None: + example = ( + Path(__file__).parent + / "example_scripts/fixtures/fill_fixtures/test_funcarg_basic.py" + ) + module = module_from_path(example) + items = collect_tests(module, rootpath=example.parent) + assert items[0].path == example + assert items[0].location[0] == "test_funcarg_basic.py" + run_tests(module, rootpath=example.parent).assert_outcomes(passed=1) + + def test_module_from_path_stays_out_of_sys_modules(self) -> None: + example = ( + Path(__file__).parent + / "example_scripts/fixtures/fill_fixtures/test_funcarg_basic.py" + ) + before = set(sys.modules) + module_from_path(example) + assert set(sys.modules) == before + + def test_synthesized_module_still_gets_a_synthetic_path( + self, tmp_path: Path + ) -> None: + def test_fn() -> None: ... + + items = collect_tests(test_fn, rootpath=tmp_path) + assert items[0].path == tmp_path / "test_ensemble.py" + + +class TestCollection: + def test_collect_loose_functions(self, tmp_path: Path) -> None: + def test_one() -> None: ... + + def test_two() -> None: ... + + items = collect_tests(test_one, test_two, rootpath=tmp_path) + assert [item.nodeid for item in items] == [ + "test_ensemble.py::test_one", + "test_ensemble.py::test_two", + ] + + def test_collect_class(self, tmp_path: Path) -> None: + class TestGroup: + def test_method(self) -> None: ... + + @staticmethod + def test_static() -> None: ... + + items = collect_tests(TestGroup, rootpath=tmp_path) + assert [item.name for item in items] == ["test_method", "test_static"] + assert items[0].nodeid == "test_ensemble.py::TestGroup::test_method" + + def test_collect_module_object(self, tmp_path: Path) -> None: + def test_in_module() -> None: ... + + module = build_module("my_virtual", test_in_module) + items = collect_tests(module, rootpath=tmp_path) + assert [item.nodeid for item in items] == ["my_virtual.py::test_in_module"] + + def test_module_pytestmark_applies(self, tmp_path: Path) -> None: + def test_marked() -> None: ... + + module = build_module( + "marked_mod", + test_marked, + pytestmark=[pytest.mark.skip(reason="module-wide")], + ) + record = run_tests(module, rootpath=tmp_path) + record.assert_outcomes(skipped=1) + + def test_parametrize(self, tmp_path: Path) -> None: + @pytest.mark.parametrize("x", [1, 2, 3]) + def test_param(x: int) -> None: + assert x < 3 + + items = collect_tests(test_param, rootpath=tmp_path) + assert [item.name for item in items] == [ + "test_param[1]", + "test_param[2]", + "test_param[3]", + ] + record = run_tests(test_param, rootpath=tmp_path) + record.assert_outcomes(passed=2, failed=1) + + def test_keyword_deselection(self, tmp_path: Path) -> None: + def test_alpha() -> None: ... + + def test_beta() -> None: ... + + spec = ConfigSpec(rootpath=tmp_path, args=("-k", "alpha")) + record = run_tests(test_alpha, test_beta, spec=spec) + record.assert_outcomes(passed=1, deselected=1) + + def test_lowercase_class_not_collected(self, tmp_path: Path) -> None: + class test: + pass + + assert collect_tests(test, rootpath=tmp_path) == [] + + def test_unittest_testcase(self, tmp_path: Path) -> None: + class MyCase(unittest.TestCase): + def test_method(self) -> None: + self.assertEqual(1, 1) + + record = run_tests(MyCase, rootpath=tmp_path) + record.assert_outcomes(passed=1) + + def test_two_module_sources(self, tmp_path: Path) -> None: + def test_a() -> None: ... + + def test_b() -> None: ... + + mod_a = build_module("mod_a", test_a) + mod_b = build_module("mod_b", test_b) + items = collect_tests(mod_a, mod_b, rootpath=tmp_path) + assert [item.nodeid for item in items] == [ + "mod_a.py::test_a", + "mod_b.py::test_b", + ] + + def test_build_module_requires_named_members(self) -> None: + with pytest.raises(ValueError, match="has no __name__"): + build_module("mod", 42) + + def test_collect_imported_tests_false_rejects_loose(self, tmp_path: Path) -> None: + """Loose sources always live in a synthesized namespace, which + collect_imported_tests=False would silently drop.""" + + def test_loose() -> None: ... + + spec = ConfigSpec(rootpath=tmp_path, inicfg={"collect_imported_tests": "false"}) + with pytest.raises(ValueError, match="collect_imported_tests"): + collect_tests(test_loose, spec=spec) + + +class TestRunning: + def test_outcome_categories(self, tmp_path: Path) -> None: + def test_passes() -> None: ... + + @pytest.mark.skipif("True", reason="nope") + def test_skips() -> None: ... + + def test_fails() -> None: + left = 1 + assert left == 2 + + @pytest.mark.xfail(reason="known") + def test_xfails() -> None: + raise AssertionError("boom") + + record = run_tests( + test_passes, test_skips, test_fails, test_xfails, rootpath=tmp_path + ) + record.assert_outcomes(passed=1, skipped=1, failed=1, xfailed=1) + assert record["test_passes"].passed + assert record["test_fails"].failed + assert record["test_skips"].skipped + assert "nope" in record["test_skips"].setup.longreprtext # type: ignore[union-attr] + assert [r.when for r in record["test_passes"].reports] == [ + "setup", + "call", + "teardown", + ] + + def test_setup_error_is_error(self, tmp_path: Path) -> None: + @pytest.fixture + def broken() -> None: + raise RuntimeError("bad setup") + + def test_uses_broken(broken: None) -> None: ... + + record = run_tests(broken, test_uses_broken, rootpath=tmp_path) + record.assert_outcomes(errors=1) + assert record["test_uses_broken"].outcome == "error" + + def test_warning_recorded(self, tmp_path: Path) -> None: + def test_warns() -> None: + warnings.warn(UserWarning("boo")) + + # The host suite runs with filterwarnings=error, which is process + # state the nested run inherits; the ensemble's own ini filters + # take precedence over it. + spec = ConfigSpec(rootpath=tmp_path, inicfg={"filterwarnings": ["always"]}) + record = run_tests(test_warns, spec=spec) + record.assert_outcomes(passed=1, warnings=1) + assert "boo" in str(record.warnings[0].message) + + def test_getitem_ambiguity(self, tmp_path: Path) -> None: + def test_same() -> None: ... + + mod_a = build_module("dup_a", test_same) + mod_b = build_module("dup_b", test_same) + record = run_tests(mod_a, mod_b, rootpath=tmp_path) + record.assert_outcomes(passed=2) + assert record["dup_a.py::test_same"].passed + with pytest.raises(KeyError, match="no unambiguous test"): + record["test_same"] + + def test_stepwise_ensemble(self, tmp_path: Path) -> None: + def test_one() -> None: ... + + with Ensemble(test_one, rootpath=tmp_path) as ensemble: + items = ensemble.collect() + assert len(items) == 1 + record = ensemble.run() + record.assert_outcomes(passed=1) + + def test_sequential_ensembles(self, tmp_path: Path) -> None: + def test_first() -> None: ... + + def test_second() -> None: + assert False + + run_tests(test_first, rootpath=tmp_path).assert_outcomes(passed=1) + run_tests(test_second, rootpath=tmp_path).assert_outcomes(failed=1) + + def test_run_subset_of_collected_items(self, tmp_path: Path) -> None: + def test_a() -> None: ... + + def test_b() -> None: ... + + with Ensemble(test_a, test_b, rootpath=tmp_path) as ensemble: + items = ensemble.collect() + record = ensemble.run(items[:1]) + record.assert_outcomes(passed=1) + assert list(record.by_test) == ["test_ensemble.py::test_a"] + + def test_argument_errors_are_not_masked(self, tmp_path: Path) -> None: + """A bad argument must report itself, not a KeyError from teardown. + + The teardown reads the config-warnings stash unconditionally, so + initialising it late meant an error raised while parsing arguments + surfaced as ``KeyError`` with the real cause only in __context__. + """ + spec = ConfigSpec(rootpath=tmp_path, args=("--not-an-option",)) + with pytest.raises(UsageError, match="unrecognized arguments"): + with configured(spec): + pass + + def test_configure_warnings_do_not_escape(self, tmp_path: Path) -> None: + """An ensemble must not warn into whatever is running it. + + A host suite running with ``filterwarnings = error`` would fail a + test for a warning that is not its own. + """ + + class WarnsAtConfigure: + def pytest_configure(self, config: object) -> None: + warnings.warn(UserWarning("from-the-ensemble")) + + def test_ok() -> None: ... + + spec = ConfigSpec( + rootpath=tmp_path, + extra_plugins=(WarnsAtConfigure(),), + inicfg={"filterwarnings": ["always"]}, + ) + with warnings.catch_warnings(record=True) as escaped: + warnings.simplefilter("always") + record = run_tests(test_ok, spec=spec) + + assert [str(w.message) for w in escaped] == [] + assert "from-the-ensemble" in [str(w.message) for w in record.warnings] + + def test_session_teardown_is_in_the_record(self, tmp_path: Path) -> None: + """A record built during the run predates pytest_sessionfinish. + + Plugins emit warnings from there, so a record that stopped at + run() made assert_outcomes(warnings=...) quietly wrong. + """ + + class WarnsLate: + def pytest_sessionfinish(self, session: object, exitstatus: object) -> None: + warnings.warn(UserWarning("late")) + + def test_ok() -> None: ... + + spec = ConfigSpec( + rootpath=tmp_path, + extra_plugins=(WarnsLate(),), + inicfg={"filterwarnings": ["always"]}, + ) + record = run_tests(test_ok, spec=spec) + record.assert_outcomes(passed=1, warnings=1) + assert [str(w.message) for w in record.warnings] == ["late"] + + def test_maxfail_stops_the_run(self, tmp_path: Path) -> None: + """--maxfail/-x must not be silently inert. + + The early exit lives in pytest_runtestloop, which an ensemble does + not go through, so without this the whole run proceeded and a test + about stopping early would assert the opposite of its subject. + """ + + def test_a() -> None: + raise AssertionError + + def test_b() -> None: + raise AssertionError + + def test_c() -> None: ... + + spec = ConfigSpec(rootpath=tmp_path, args=("--maxfail=1",)) + record = run_tests(test_a, test_b, test_c, spec=spec) + record.assert_outcomes(failed=1) + assert list(record.by_test) == ["test_ensemble.py::test_a"] + assert record.stopped == "stopping after 1 failures" + + def test_runs_to_the_end_without_maxfail(self, tmp_path: Path) -> None: + def test_a() -> None: + raise AssertionError + + def test_b() -> None: ... + + record = run_tests(test_a, test_b, rootpath=tmp_path) + record.assert_outcomes(failed=1, passed=1) + assert record.stopped is None + + def test_collection_error_counts_as_error(self, tmp_path: Path) -> None: + @pytest.mark.parametrize("absent", [1]) + def test_bad() -> None: ... + + record = run_tests(test_bad, rootpath=tmp_path) + record.assert_outcomes(errors=1) + assert [r.nodeid for r in record.collect_errors] == ["test_ensemble.py"] + + def test_collect_tests_raises_on_collection_failure(self, tmp_path: Path) -> None: + """An empty item list must not stand in for a collection failure.""" + + @pytest.mark.parametrize("absent", [1]) + def test_bad() -> None: ... + + with pytest.raises(Collector.CollectError, match="uses no argument"): + collect_tests(test_bad, rootpath=tmp_path) + + def test_collect_errors_reachable_stepwise(self, tmp_path: Path) -> None: + """Ensemble.collect() stays permissive, but says what went wrong.""" + + @pytest.mark.parametrize("absent", [1]) + def test_bad() -> None: ... + + with Ensemble(test_bad, rootpath=tmp_path) as ensemble: + assert ensemble.collect() == [] + assert [r.nodeid for r in ensemble.collect_errors] == ["test_ensemble.py"] + + def test_no_collect_errors_when_nothing_matches(self, tmp_path: Path) -> None: + """Genuinely collecting nothing is not an error.""" + + class NotATest: + pass + + assert collect_tests(NotATest, rootpath=tmp_path) == [] + + def test_outcome_empty_when_no_call_phase(self, tmp_path: Path) -> None: + """--setup-only produces setup/teardown reports whose status + category is empty, so the item has no aggregate outcome.""" + + def test_noop() -> None: ... + + spec = ConfigSpec(rootpath=tmp_path, args=("--setup-only",)).with_plugins( + "setuponly" + ) + record = run_tests(test_noop, spec=spec) + assert record["test_noop"].outcome == "" + + +class TestEnsembleLifecycle: + def test_rootpath_fills_in_spec(self, tmp_path: Path) -> None: + """An explicit rootpath supplies a spec that does not carry one.""" + spec = ConfigSpec(args=("-k", "nothing")) + with Ensemble(rootpath=tmp_path, spec=spec) as ensemble: + assert ensemble.config.rootpath == tmp_path + + def test_not_reentrant(self, tmp_path: Path) -> None: + ensemble = Ensemble(rootpath=tmp_path) + with ensemble: + with pytest.raises(RuntimeError, match="not reentrant"), ExitStack() as s: + s.enter_context(ensemble) + + def test_failure_to_start_session_unconfigures(self, tmp_path: Path) -> None: + """If the session fails to start, the already-configured config is + still torn down.""" + + class BoomPlugin: + def pytest_sessionstart(self, session: object) -> None: + raise RuntimeError("boom") + + plugin = BoomPlugin() + spec = ConfigSpec(rootpath=tmp_path, extra_plugins=(plugin,)) + ensemble = Ensemble(spec=spec) + with pytest.raises(RuntimeError, match="boom"), ExitStack() as stack: + stack.enter_context(ensemble) + + def test_collect_is_idempotent(self, tmp_path: Path) -> None: + """A second collect() must not register the same sources again.""" + + def test_a() -> None: ... + + with Ensemble(test_a, rootpath=tmp_path) as ensemble: + first = ensemble.collect() + second = ensemble.collect() + assert [i.nodeid for i in first] == [i.nodeid for i in second] + ensemble.run().assert_outcomes(passed=1) + + def test_collect_accepts_extra_sources(self, tmp_path: Path) -> None: + """Later rounds add to the tree without colliding with earlier ones.""" + + def test_a() -> None: ... + + def test_b() -> None: ... + + with Ensemble(test_a, rootpath=tmp_path) as ensemble: + ensemble.collect() + ensemble.collect(test_b) + assert [i.nodeid for i in ensemble.session.items] == [ + "test_ensemble.py::test_a", + "test_ensemble_1.py::test_b", + ] + ensemble.run().assert_outcomes(passed=2) + + def test_exception_in_body_reaches_session_teardown(self, tmp_path: Path) -> None: + """A failure inside the ensemble body is forwarded to the session + teardown, not swallowed by closing the stack blind.""" + seen: list[int] = [] + + class Recorder: + def pytest_sessionfinish(self, exitstatus: int) -> None: + seen.append(exitstatus) + + spec = ConfigSpec(rootpath=tmp_path, extra_plugins=(Recorder(),)) + with pytest.raises(RuntimeError, match="inner"): + with Ensemble(spec=spec): + raise RuntimeError("inner") + assert seen == [ExitCode.INTERNAL_ERROR] + + +class TestFixtures: + def test_function_fixture(self, tmp_path: Path) -> None: + @pytest.fixture + def value() -> int: + return 41 + + def test_uses_value(value: int) -> None: + assert value == 41 + + record = run_tests(value, test_uses_value, rootpath=tmp_path) + record.assert_outcomes(passed=1) + + def test_class_scoped_fixture_and_teardown_order(self, tmp_path: Path) -> None: + events: list[str] = [] + + class TestGroup: + @pytest.fixture(scope="class") + def resource(self) -> Generator[str]: + events.append("setup") + yield "res" + events.append("teardown") + + def test_one(self, resource: str) -> None: + events.append("one") + + def test_two(self, resource: str) -> None: + events.append("two") + + record = run_tests(TestGroup, rootpath=tmp_path) + record.assert_outcomes(passed=2) + assert events == ["setup", "one", "two", "teardown"] + + def test_module_scoped_fixture(self, tmp_path: Path) -> None: + events: list[str] = [] + + @pytest.fixture(scope="module") + def modres() -> Generator[int]: + events.append("setup") + yield 1 + events.append("teardown") + + def test_a(modres: int) -> None: + events.append("a") + + def test_b(modres: int) -> None: + events.append("b") + + record = run_tests(modres, test_a, test_b, rootpath=tmp_path) + record.assert_outcomes(passed=2) + assert events == ["setup", "a", "b", "teardown"] + + def test_request_module_is_synthesized_module(self, tmp_path: Path) -> None: + seen: list[types.ModuleType] = [] + + def test_introspect(request: pytest.FixtureRequest) -> None: + seen.append(request.module) + + run_tests(test_introspect, rootpath=tmp_path).assert_outcomes(passed=1) + (module,) = seen + assert isinstance(module, types.ModuleType) + assert module.__name__ == "test_ensemble" + + def test_fixture_from_extra_plugin(self, tmp_path: Path) -> None: + class FixturePlugin: + @pytest.fixture + def injected(self) -> str: + return "from-plugin" + + def test_uses_injected(injected: str) -> None: + assert injected == "from-plugin" + + spec = ConfigSpec(rootpath=tmp_path, extra_plugins=(FixturePlugin(),)) + run_tests(test_uses_injected, spec=spec).assert_outcomes(passed=1) + + def test_monkeypatch_fixture_available(self, tmp_path: Path) -> None: + def test_uses_monkeypatch(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("ENSEMBLE_PROBE", "1") + assert os.environ["ENSEMBLE_PROBE"] == "1" + + run_tests(test_uses_monkeypatch, rootpath=tmp_path).assert_outcomes(passed=1) + assert "ENSEMBLE_PROBE" not in os.environ + + +class TestNodeConstruction: + def test_class_from_parent_keeps_obj(self, tmp_path: Path) -> None: + """Class.from_parent(obj=...) takes precedence over name lookup.""" + + class Hidden: + def test_method(self) -> None: ... + + empty = build_module("holder") + with configured(ConfigSpec(rootpath=tmp_path)) as config: + with running_session(config) as session: + module = EnsembleModule.from_parent(session, obj=empty, name="holder") + cls = pytest.Class.from_parent( + module, name="NotAnAttribute", obj=Hidden + ) + assert cls.obj is Hidden + assert [item.name for item in cls.collect()] == ["test_method"] + + +class TestHermeticity: + def test_no_process_state_leaked(self, tmp_path: Path) -> None: + def test_noop() -> None: ... + + # Warm-up: let lazy imports happen before snapshotting. + run_tests(test_noop, rootpath=tmp_path).assert_outcomes(passed=1) + + cwd = os.getcwd() + sys_path = list(sys.path) + modules = set(sys.modules) + environ = dict(os.environ) + + run_tests(test_noop, rootpath=tmp_path).assert_outcomes(passed=1) + + assert os.getcwd() == cwd + assert sys.path == sys_path + assert set(sys.modules) == modules + assert dict(os.environ) == environ + + def test_no_files_created(self, tmp_path: Path) -> None: + def test_noop() -> None: ... + + run_tests(test_noop, rootpath=tmp_path).assert_outcomes(passed=1) + assert list(tmp_path.iterdir()) == [] + + +class TestPytesterInterplay: + def test_ensemble_inside_full_pytest_run(self, pytester: Pytester) -> None: + """The ensemble API works inside a captured, terminal-full pytest run.""" + pytester.makepyfile( + """ + from _pytest.ensemble import run_tests + + def test_host(tmp_path): + def test_inner(): + assert 1 + 1 == 2 + + record = run_tests(test_inner, rootpath=tmp_path) + record.assert_outcomes(passed=1) + """ + ) + result = pytester.runpytest_inprocess() + result.assert_outcomes(passed=1) diff --git a/testing/test_reports.py b/testing/test_reports.py index 23c5968bb52..38eef33194c 100644 --- a/testing/test_reports.py +++ b/testing/test_reports.py @@ -2,31 +2,61 @@ from __future__ import annotations from collections.abc import Sequence +from concurrent.futures import ProcessPoolExecutor +from pathlib import Path +from _pytest import timing from _pytest._code.code import ExceptionChainRepr from _pytest._code.code import ExceptionRepr from _pytest.approx import approx from _pytest.config import Config +from _pytest.ensemble import ConfigSpec +from _pytest.ensemble import run_tests from _pytest.pytester import Pytester from _pytest.reports import CollectReport from _pytest.reports import TestReport import pytest +def raise_value_error() -> None: + """Raise in a worker process, for test_chained_exceptions_no_reprcrash. + + ``ProcessPoolExecutor`` pickles the callable it is handed by reference, + so it has to live at module level rather than inside the test. + """ + raise ValueError("value error") + + +class RaiseCollectError: + """Plugin failing collection of the ensemble's module. + + The originals collected a file whose whole source was ``qwe abc``, a + syntax error; an ensemble's sources are real python objects, so there is + no unparsable source to hand it and the collection error is raised from + a hook instead. Both end in a ``CollectError``, which is what makes the + report's ``longrepr`` a plain string rather than a structured repr - the + serialization path these tests are about. + """ + + def pytest_generate_tests(self, metafunc: pytest.Metafunc) -> None: + raise pytest.Collector.CollectError("qwe abc") + + class TestReportSerialization: - def test_xdist_longrepr_to_str_issue_241(self, pytester: Pytester) -> None: + def test_xdist_longrepr_to_str_issue_241(self, tmp_path: Path) -> None: """Regarding issue pytest-xdist#241. This test came originally from test_remote.py in xdist (ca03269). """ - pytester.makepyfile( - """ - def test_a(): assert False - def test_b(): pass - """ - ) - reprec = pytester.inline_run() - reports = reprec.getreports("pytest_runtest_logreport") + + def test_a(): + assert False + + def test_b(): + pass + + record = run_tests(test_a, test_b, rootpath=tmp_path) + reports = record.reports assert len(reports) == 6 test_a_call = reports[1] assert test_a_call.when == "call" @@ -37,18 +67,17 @@ def test_b(): pass assert test_b_call.outcome == "passed" assert test_b_call._to_json()["longrepr"] is None - def test_xdist_report_longrepr_reprcrash_130(self, pytester: Pytester) -> None: + def test_xdist_report_longrepr_reprcrash_130(self, tmp_path: Path) -> None: """Regarding issue pytest-xdist#130 This test came originally from test_remote.py in xdist (ca03269). """ - reprec = pytester.inline_runsource( - """ - def test_fail(): - assert False, 'Expected Message' - """ - ) - reports = reprec.getreports("pytest_runtest_logreport") + + def test_fail(): + assert False, "Expected Message" + + record = run_tests(test_fail, rootpath=tmp_path) + reports = record.reports assert len(reports) == 3 rep = reports[1] added_section = ("Failure Metadata", "metadata metadata", "*") @@ -76,22 +105,27 @@ def test_fail(): # Missing section attribute PR171 assert added_section in a.longrepr.sections - def test_reprentries_serialization_170(self, pytester: Pytester) -> None: + def test_reprentries_serialization_170(self, tmp_path: Path) -> None: """Regarding issue pytest-xdist#170 This test came originally from test_remote.py in xdist (ca03269). """ from _pytest._code.code import ReprEntry - reprec = pytester.inline_runsource( - """ - def test_repr_entry(): - x = 0 - assert x - """, - "--showlocals", + def test_repr_entry(): + x = 0 + assert x + + # --showlocals is a terminal option, so the terminal plugin has to be + # loaded for it to exist at all; capture_output does that and binds it + # to a buffer of its own. + record = run_tests( + test_repr_entry, + rootpath=tmp_path, + spec=ConfigSpec(args=("--showlocals",)), + capture_output=True, ) - reports = reprec.getreports("pytest_runtest_logreport") + reports = record.reports assert len(reports) == 3 rep = reports[1] assert isinstance(rep.longrepr, ExceptionRepr) @@ -101,6 +135,7 @@ def test_repr_entry(): rep_entries = rep.longrepr.reprtraceback.reprentries a_entries = a.longrepr.reprtraceback.reprentries + assert rep_entries for a_entry, rep_entry in zip(a_entries, rep_entries, strict=True): assert isinstance(rep_entry, ReprEntry) assert rep_entry.reprfileloc is not None @@ -120,22 +155,25 @@ def test_repr_entry(): assert rep_entry.reprlocals.lines == a_entry.reprlocals.lines assert rep_entry.style == a_entry.style - def test_reprentries_serialization_196(self, pytester: Pytester) -> None: + def test_reprentries_serialization_196(self, tmp_path: Path) -> None: """Regarding issue pytest-xdist#196 This test came originally from test_remote.py in xdist (ca03269). """ from _pytest._code.code import ReprEntryNative - reprec = pytester.inline_runsource( - """ - def test_repr_entry_native(): - x = 0 - assert x - """, - "--tb=native", + def test_repr_entry_native(): + x = 0 + assert x + + # --tb is a terminal option; see test_reprentries_serialization_170. + record = run_tests( + test_repr_entry_native, + rootpath=tmp_path, + spec=ConfigSpec(args=("--tb=native",)), + capture_output=True, ) - reports = reprec.getreports("pytest_runtest_logreport") + reports = record.reports assert len(reports) == 3 rep = reports[1] assert isinstance(rep.longrepr, ExceptionRepr) @@ -145,29 +183,50 @@ def test_repr_entry_native(): rep_entries = rep.longrepr.reprtraceback.reprentries a_entries = a.longrepr.reprtraceback.reprentries + assert rep_entries for rep_entry, a_entry in zip(rep_entries, a_entries, strict=True): assert isinstance(rep_entry, ReprEntryNative) assert rep_entry.lines == a_entry.lines - def test_itemreport_outcomes(self, pytester: Pytester) -> None: + def test_itemreport_outcomes(self, tmp_path: Path) -> None: # This test came originally from test_remote.py in xdist (ca03269). - reprec = pytester.inline_runsource( - """ - import pytest - def test_pass(): pass - def test_fail(): 0/0 - @pytest.mark.skipif("True") - def test_skip(): pass - def test_skip_imperative(): - pytest.skip("hello") - @pytest.mark.xfail("True") - def test_xfail(): 0/0 - def test_xfail_imperative(): - pytest.xfail("hello") - """ + def test_pass(): + pass + + def test_fail(): + 0 / 0 # noqa: B018 + + @pytest.mark.skipif("True") + def test_skip(): + pass + + def test_skip_imperative(): + pytest.skip("hello") + + @pytest.mark.xfail("True") + def test_xfail(): + 0 / 0 # noqa: B018 + + def test_xfail_imperative(): + pytest.xfail("hello") + + record = run_tests( + test_pass, + test_fail, + test_skip, + test_skip_imperative, + test_xfail, + test_xfail_imperative, + rootpath=tmp_path, ) - reports = reprec.getreports("pytest_runtest_logreport") + reports = record.reports assert len(reports) == 17 # with setup/teardown "passed" reports + assert record.outcomes() == { + "passed": 1, + "failed": 1, + "skipped": 2, + "xfailed": 2, + } for rep in reports: d = rep._to_json() newrep = TestReport._from_json(d) @@ -183,10 +242,17 @@ def test_xfail_imperative(): if rep.failed: assert newrep.longreprtext == rep.longreprtext - def test_collectreport_passed(self, pytester: Pytester) -> None: + def test_collectreport_passed(self, tmp_path: Path) -> None: """This test came originally from test_remote.py in xdist (ca03269).""" - reprec = pytester.inline_runsource("def test_func(): pass") - reports = reprec.getreports("pytest_collectreport") + + def test_func(): + pass + + record = run_tests(test_func, rootpath=tmp_path) + reports = record.collect_reports + # session and module; a real run also collects a Directory in between, + # which an ensemble has no equivalent of. + assert len(reports) == 2 for rep in reports: d = rep._to_json() newrep = CollectReport._from_json(d) @@ -194,11 +260,20 @@ def test_collectreport_passed(self, pytester: Pytester) -> None: assert newrep.failed == rep.failed assert newrep.skipped == rep.skipped - def test_collectreport_fail(self, pytester: Pytester) -> None: + def test_collectreport_fail(self, tmp_path: Path) -> None: """This test came originally from test_remote.py in xdist (ca03269).""" - reprec = pytester.inline_runsource("qwe abc") - reports = reprec.getreports("pytest_collectreport") + + def test_func(): + pass + + record = run_tests( + test_func, + rootpath=tmp_path, + spec=ConfigSpec(extra_plugins=(RaiseCollectError(),)), + ) + reports = record.collect_reports assert reports + assert record.collect_errors for rep in reports: d = rep._to_json() newrep = CollectReport._from_json(d) @@ -208,11 +283,20 @@ def test_collectreport_fail(self, pytester: Pytester) -> None: if rep.failed: assert newrep.longrepr == str(rep.longrepr) - def test_extended_report_deserialization(self, pytester: Pytester) -> None: + def test_extended_report_deserialization(self, tmp_path: Path) -> None: """This test came originally from test_remote.py in xdist (ca03269).""" - reprec = pytester.inline_runsource("qwe abc") - reports = reprec.getreports("pytest_collectreport") + + def test_func(): + pass + + record = run_tests( + test_func, + rootpath=tmp_path, + spec=ConfigSpec(extra_plugins=(RaiseCollectError(),)), + ) + reports = record.collect_reports assert reports + assert record.collect_errors for rep in reports: rep.extra = True # type: ignore[attr-defined] d = rep._to_json() @@ -224,14 +308,11 @@ def test_extended_report_deserialization(self, pytester: Pytester) -> None: if rep.failed: assert newrep.longrepr == str(rep.longrepr) - def test_paths_support(self, pytester: Pytester) -> None: + def test_paths_support(self, tmp_path: Path) -> None: """Report attributes which are path-like should become strings.""" - pytester.makepyfile( - """ - def test_a(): - assert False - """ - ) + + def test_a(): + assert False class MyPathLike: def __init__(self, path: str) -> None: @@ -240,26 +321,24 @@ def __init__(self, path: str) -> None: def __fspath__(self) -> str: return self.path - reprec = pytester.inline_run() - reports = reprec.getreports("pytest_runtest_logreport") + record = run_tests(test_a, rootpath=tmp_path) + reports = record.reports assert len(reports) == 3 test_a_call = reports[1] - test_a_call.path1 = MyPathLike(str(pytester.path)) # type: ignore[attr-defined] - test_a_call.path2 = pytester.path # type: ignore[attr-defined] + test_a_call.path1 = MyPathLike(str(tmp_path)) # type: ignore[attr-defined] + test_a_call.path2 = tmp_path # type: ignore[attr-defined] data = test_a_call._to_json() - assert data["path1"] == str(pytester.path) - assert data["path2"] == str(pytester.path) + assert data["path1"] == str(tmp_path) + assert data["path2"] == str(tmp_path) - def test_deserialization_failure(self, pytester: Pytester) -> None: + def test_deserialization_failure(self, tmp_path: Path) -> None: """Check handling of failure during deserialization of report types.""" - pytester.makepyfile( - """ - def test_a(): - assert False - """ - ) - reprec = pytester.inline_run() - reports = reprec.getreports("pytest_runtest_logreport") + + def test_a(): + assert False + + record = run_tests(test_a, rootpath=tmp_path) + reports = record.reports assert len(reports) == 3 test_a_call = reports[1] data = test_a_call._to_json() @@ -273,39 +352,45 @@ def test_a(): TestReport._from_json(data) @pytest.mark.parametrize("report_class", [TestReport, CollectReport]) - def test_chained_exceptions( - self, pytester: Pytester, tw_mock, report_class - ) -> None: + def test_chained_exceptions(self, tmp_path: Path, tw_mock, report_class) -> None: """Check serialization/deserialization of report objects containing chained exceptions (#5786)""" - pytester.makepyfile( - f""" - def foo(): - raise ValueError('value error') - def test_a(): - try: - foo() - except ValueError as e: - raise RuntimeError('runtime error') from e - if {report_class is CollectReport}: + + def foo(): + raise ValueError("value error") + + def test_a(): + try: + foo() + except ValueError as e: + raise RuntimeError("runtime error") from e + + class RaiseWhileCollecting: + """The original raised the chained exception at module import + time, by calling ``test_a()`` at module level; an ensemble source + is a module that has already been imported, so the same call is + made from a collection hook instead - which likewise turns the + module's CollectReport into a failed one carrying the chain. + """ + + def pytest_generate_tests(self, metafunc: pytest.Metafunc) -> None: test_a() - """ - ) - reprec = pytester.inline_run() + spec = ConfigSpec(rootpath=tmp_path) + if report_class is CollectReport: + spec = spec.replace(extra_plugins=(RaiseWhileCollecting(),)) + record = run_tests(test_a, spec=spec) + + report: TestReport | CollectReport if report_class is TestReport: - reports: Sequence[TestReport] | Sequence[CollectReport] = reprec.getreports( - "pytest_runtest_logreport" - ) + reports: Sequence[TestReport] | Sequence[CollectReport] = record.reports # we have 3 reports: setup/call/teardown assert len(reports) == 3 # get the call report report = reports[1] else: assert report_class is CollectReport - # three collection reports: session, test file, directory - reports = reprec.getreports("pytest_collectreport") - assert len(reports) == 3 - report = reports[1] + # only the module's collection fails; the session's does not + (report,) = record.collect_errors def check_longrepr(longrepr: ExceptionChainRepr) -> None: """Check the attributes of the given longrepr object according to the test file. @@ -320,8 +405,10 @@ def check_longrepr(longrepr: ExceptionChainRepr) -> None: tb1, _fileloc1, desc1 = entry1 tb2, _fileloc2, desc2 = entry2 - assert "ValueError('value error')" in str(tb1) - assert "RuntimeError('runtime error')" in str(tb2) + # the sources are this file's own now, so they are quoted the way + # this file is formatted + assert 'ValueError("value error")' in str(tb1) + assert 'RuntimeError("runtime error")' in str(tb2) assert ( desc1 @@ -346,30 +433,23 @@ def check_longrepr(longrepr: ExceptionChainRepr) -> None: # elsewhere and we do check the contents of the longrepr object after loading it. loaded_report.longrepr.toterminal(tw_mock) - def test_chained_exceptions_no_reprcrash(self, pytester: Pytester, tw_mock) -> None: + def test_chained_exceptions_no_reprcrash(self, tmp_path: Path, tw_mock) -> None: """Regression test for tracebacks without a reprcrash (#5971) This happens notably on exceptions raised by multiprocess.pool: the exception transfer from subprocess to main process creates an artificial exception, which ExceptionInfo can't obtain the ReprFileLocation from. """ - pytester.makepyfile( - """ - from concurrent.futures import ProcessPoolExecutor - - def func(): - raise ValueError('value error') - - def test_a(): - with ProcessPoolExecutor() as p: - p.submit(func).result() - """ - ) - pytester.syspathinsert() - reprec = pytester.inline_run() + def test_a(): + with ProcessPoolExecutor() as p: + # the pool pickles the callable by reference, so it is + # ``raise_value_error`` at this module's level rather than a + # function defined in here + p.submit(raise_value_error).result() - reports = reprec.getreports("pytest_runtest_logreport") + record = run_tests(test_a, rootpath=tmp_path) + reports = record.reports def check_longrepr(longrepr: object) -> None: assert isinstance(longrepr, ExceptionChainRepr) @@ -403,6 +483,8 @@ def check_longrepr(longrepr: object) -> None: assert isinstance(loaded_report.longrepr, ExceptionChainRepr) loaded_report.longrepr.toterminal(tw_mock) + # ensemble: a conftest in a subdirectory, failing to import, observed + # through a subprocess run. def test_report_prevent_ConftestImportFailure_hiding_exception( self, pytester: Pytester ) -> None: @@ -414,20 +496,18 @@ def test_report_prevent_ConftestImportFailure_hiding_exception( result.stdout.fnmatch_lines(["E *Error: No module named 'unknown'"]) result.stdout.no_fnmatch_line("ERROR - *ConftestImportFailure*") - def test_report_timestamps_match_duration(self, pytester: Pytester, mock_timing): - reprec = pytester.inline_runsource( - """ - import pytest - from _pytest import timing - @pytest.fixture - def fixture_(): - timing.sleep(5) - yield - timing.sleep(5) - def test_1(fixture_): timing.sleep(10) - """ - ) - reports = reprec.getreports("pytest_runtest_logreport") + def test_report_timestamps_match_duration(self, tmp_path: Path, mock_timing): + @pytest.fixture + def fixture_(): + timing.sleep(5) + yield + timing.sleep(5) + + def test_1(fixture_): + timing.sleep(10) + + record = run_tests(fixture_, test_1, rootpath=tmp_path) + reports = record.reports assert len(reports) == 3 for report in reports: data = report._to_json() @@ -440,7 +520,7 @@ def test_1(fixture_): timing.sleep(10) ) def test_exception_group_with_only_skips( self, - pytester: Pytester, + tmp_path: Path, first_skip_reason: str, second_skip_reason: str, skip_reason_output: str, @@ -450,80 +530,108 @@ def test_exception_group_with_only_skips( it is reported as a single skipped test, not as an error. This is a regression test for issue #13537. """ - pytester.makepyfile( - test_it=f""" - import pytest - @pytest.fixture - def fixA(): - yield - pytest.skip(reason="{first_skip_reason}") - @pytest.fixture - def fixB(): - yield - pytest.skip(reason="{second_skip_reason}") - def test_skip(fixA, fixB): - assert True - """ + + @pytest.fixture + def fixA(): + yield + pytest.skip(reason=first_skip_reason) + + @pytest.fixture + def fixB(): + yield + pytest.skip(reason=second_skip_reason) + + def test_skip(fixA, fixB): + assert True + + record = run_tests( + fixA, + fixB, + test_skip, + rootpath=tmp_path, + spec=ConfigSpec(args=("-v",)), + capture_output=True, ) - result = pytester.runpytest("-v") - result.assert_outcomes(passed=1, skipped=1) - out = result.stdout.str() + record.assert_outcomes(passed=1, skipped=1) + # the merged reason is on the report itself, not only in the rendering + teardown = record["test_skip"].teardown + assert teardown is not None + assert teardown.skipped + assert teardown.longrepr is not None + assert isinstance(teardown.longrepr, tuple) + assert skip_reason_output == f"({teardown.longrepr[2]})" + out = record.output assert skip_reason_output in out assert "ERROR at teardown" not in out @pytest.mark.parametrize( "use_item_location, skip_file_location", - [(True, "test_it.py"), (False, "runner.py")], + # the item lives in this very file, since its source is written here: + # the original's "test_it.py" is this file's name now + [(True, "test_reports.py"), (False, "runner.py")], ) def test_exception_group_skips_use_item_location( - self, pytester: Pytester, use_item_location: bool, skip_file_location: str + self, tmp_path: Path, use_item_location: bool, skip_file_location: str ): """ Regression for #13537: If any skip inside an ExceptionGroup has _use_item_location=True, the report location should point to the test item, not the fixture teardown. """ - pytester.makepyfile( - test_it=f""" - import pytest - @pytest.fixture - def fix_item1(): - yield - exc = pytest.skip.Exception("A") - exc._use_item_location = True - raise exc - @pytest.fixture - def fix_item2(): - yield - exc = pytest.skip.Exception("B") - exc._use_item_location = {use_item_location} - raise exc - def test_both(fix_item1, fix_item2): - assert True - """ - ) - result = pytester.runpytest("-rs") - result.assert_outcomes(passed=1, skipped=1) - out = result.stdout.str() - # Both reasons should appear - assert "A" and "B" in out + @pytest.fixture + def fix_item1(): + yield + exc = pytest.skip.Exception("A") + exc._use_item_location = True + raise exc + + @pytest.fixture + def fix_item2(): + yield + exc = pytest.skip.Exception("B") + exc._use_item_location = use_item_location + raise exc + + def test_both(fix_item1, fix_item2): + assert True + + record = run_tests( + fix_item1, + fix_item2, + test_both, + rootpath=tmp_path, + spec=ConfigSpec(args=("-rs",)), + capture_output=True, + ) + record.assert_outcomes(passed=1, skipped=1) + + teardown = record["test_both"].teardown + assert teardown is not None + assert isinstance(teardown.longrepr, tuple) + path, _lineno, reason = teardown.longrepr + # Both reasons should be reported (the original only got as far as + # asserting this of the rendering, and did so with a bug) + assert reason == "A; B" # Crucially, the skip should be attributed to the test item, not teardown + assert str(path).endswith(skip_file_location) + out = record.output + assert "A; B" in out assert skip_file_location in out class TestHooks: """Test that the hooks are working correctly for plugins""" - def test_test_report(self, pytester: Pytester, pytestconfig: Config) -> None: - pytester.makepyfile( - """ - def test_a(): assert False - def test_b(): pass - """ - ) - reprec = pytester.inline_run() - reports = reprec.getreports("pytest_runtest_logreport") + def test_test_report(self, tmp_path: Path, pytestconfig: Config) -> None: + def test_a(): + assert False + + def test_b(): + pass + + record = run_tests(test_a, test_b, rootpath=tmp_path) + reports = record.reports assert len(reports) == 6 for rep in reports: data = pytestconfig.hook.pytest_report_to_serializable( @@ -537,16 +645,18 @@ def test_b(): pass assert new_rep.when == rep.when assert new_rep.outcome == rep.outcome - def test_collect_report(self, pytester: Pytester, pytestconfig: Config) -> None: - pytester.makepyfile( - """ - def test_a(): assert False - def test_b(): pass - """ - ) - reprec = pytester.inline_run() - reports = reprec.getreports("pytest_collectreport") - assert len(reports) == 3 + def test_collect_report(self, tmp_path: Path, pytestconfig: Config) -> None: + def test_a(): + assert False + + def test_b(): + pass + + record = run_tests(test_a, test_b, rootpath=tmp_path) + reports = record.collect_reports + # session and module; a real run also collects a Directory in between, + # which an ensemble has no equivalent of. + assert len(reports) == 2 for rep in reports: data = pytestconfig.hook.pytest_report_to_serializable( config=pytestconfig, report=rep @@ -563,15 +673,17 @@ def test_b(): pass "hook_name", ["pytest_runtest_logreport", "pytest_collectreport"] ) def test_invalid_report_types( - self, pytester: Pytester, pytestconfig: Config, hook_name: str + self, tmp_path: Path, pytestconfig: Config, hook_name: str ) -> None: - pytester.makepyfile( - """ - def test_a(): pass - """ + def test_a(): + pass + + record = run_tests(test_a, rootpath=tmp_path) + reports: Sequence[TestReport] | Sequence[CollectReport] = ( + record.reports + if hook_name == "pytest_runtest_logreport" + else record.collect_reports ) - reprec = pytester.inline_run() - reports = reprec.getreports(hook_name) assert reports rep = reports[0] data = pytestconfig.hook.pytest_report_to_serializable( diff --git a/testing/test_warnings.py b/testing/test_warnings.py index 017781c2355..d27776d561b 100644 --- a/testing/test_warnings.py +++ b/testing/test_warnings.py @@ -857,15 +857,16 @@ def test_issue4445_initial_conftest(self, pytester: Pytester, capwarn) -> None: ) pytester.parseconfig("--help") - # with stacklevel=2 the warning should originate from config._preparse and is - # thrown by an erroneous conftest.py + # with stacklevel=2 the warning should originate from the conftest + # loading phase of config.parse and is thrown by an erroneous + # conftest.py assert len(capwarn.captured) == 1 warning, location = capwarn.captured.pop() file, _, func = location assert "could not load initial conftests" in str(warning.message) assert f"config{os.sep}__init__.py" in file - assert func == "parse" + assert func == "_load_initial_conftests_phase" @pytest.mark.filterwarnings("default") def test_conftest_warning_captured(self, pytester: Pytester) -> None: