p53209761
/
test_ok2
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Switch to new-style pluggy hook wrappers 

Fix #11122.
This commit is contained in:
Ran Benita 2023-06-12 22:30:06 +03:00
parent 7008385253
commit b41acaea12
34 changed files with 334 additions and 275 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
changelog/11122.improvement.rst Normal file
View File

@ -0,0 +1,6 @@
``pluggy>=1.2.0`` is now required.
pytest now uses "new-style" hook wrappers internally, available since pluggy 1.2.0.
See `pluggy's 1.2.0 changelog <https://pluggy.readthedocs.io/en/latest/changelog.html#pluggy-1-2-0-2023-06-21>`_ and the :ref:`updated docs <hookwrapper>` for details.
Plugins which want to use new-style wrappers can do so if they require this version of pytest or later.

14
doc/en/example/simple.rst
View File

@ -808,11 +808,10 @@ case we just write some information out to a ``failures`` file:
import pytest
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
@pytest.hookimpl(wrapper=True, tryfirst=True)
def pytest_runtest_makereport(item, call):
# execute all other hooks to obtain the report object
outcome = yield
rep = outcome.get_result()
rep = yield
# we only look at actual failing test calls, not setup/teardown
if rep.when == "call" and rep.failed:
@ -826,6 +825,8 @@ case we just write some information out to a ``failures`` file:
f.write(rep.nodeid + extra + "\n")
return rep
if you then have failing tests:
@ -899,16 +900,17 @@ here is a little example implemented via a local plugin:
phase_report_key = StashKey[Dict[str, CollectReport]]()
@pytest.hookimpl(tryfirst=True, hookwrapper=True)
@pytest.hookimpl(wrapper=True, tryfirst=True)
def pytest_runtest_makereport(item, call):
# execute all other hooks to obtain the report object
outcome = yield
rep = outcome.get_result()
rep = yield
# store test results for each phase of a call, which can
# be "setup", "call", "teardown"
item.stash.setdefault(phase_report_key, {})[rep.when] = rep
return rep
@pytest.fixture
def something(request):

57
doc/en/how-to/writing_hook_functions.rst
View File

@ -56,7 +56,7 @@ The remaining hook functions will not be called in this case.
.. _`hookwrapper`:
hookwrapper: executing around other hooks
hook wrappers: executing around other hooks
-------------------------------------------------
.. currentmodule:: _pytest.core
@ -69,10 +69,8 @@ which yields exactly once. When pytest invokes hooks it first executes
hook wrappers and passes the same arguments as to the regular hooks.
At the yield point of the hook wrapper pytest will execute the next hook
implementations and return their result to the yield point in the form of
a :py:class:`Result <pluggy._Result>` instance which encapsulates a result or
exception info. The yield point itself will thus typically not raise
exceptions (unless there are bugs).
implementations and return their result to the yield point, or will
propagate an exception if they raised.
Here is an example definition of a hook wrapper:
@ -81,26 +79,35 @@ Here is an example definition of a hook wrapper:
import pytest
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_pyfunc_call(pyfuncitem):
do_something_before_next_hook_executes()
outcome = yield
# outcome.excinfo may be None or a (cls, val, tb) tuple
# If the outcome is an exception, will raise the exception.
res = yield
res = outcome.get_result() # will raise if outcome was exception
new_res = post_process_result(res)
post_process_result(res)
# Override the return value to the plugin system.
return new_res
outcome.force_result(new_res) # to override the return value to the plugin system
The hook wrapper needs to return a result for the hook, or raise an exception.
Note that hook wrappers don't return results themselves, they merely
perform tracing or other side effects around the actual hook implementations.
If the result of the underlying hook is a mutable object, they may modify
that result but it's probably better to avoid it.
In many cases, the wrapper only needs to perform tracing or other side effects
around the actual hook implementations, in which case it can return the result
value of the ``yield``. The simplest (though useless) hook wrapper is
``return (yield)``.
In other cases, the wrapper wants the adjust or adapt the result, in which case
it can return a new value. If the result of the underlying hook is a mutable
object, the wrapper may modify that result, but it's probably better to avoid it.
If the hook implementation failed with an exception, the wrapper can handle that
exception using a ``try-catch-finally`` around the ``yield``, by propagating it,
supressing it, or raising a different exception entirely.
For more information, consult the
:ref:`pluggy documentation about hookwrappers <pluggy:hookwrappers>`.
:ref:`pluggy documentation about hook wrappers <pluggy:hookwrappers>`.
.. _plugin-hookorder:
@ -130,11 +137,14 @@ after others, i.e. the position in the ``N``-sized list of functions:
# Plugin 3
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_collection_modifyitems(items):
# will execute even before the tryfirst one above!
outcome = yield
# will execute after all non-hookwrappers executed
try:
return (yield)
finally:
# will execute after all non-wrappers executed
...
Here is the order of execution:
@ -149,12 +159,11 @@ Here is the order of execution:
Plugin1).
4. Plugin3's pytest_collection_modifyitems then executing the code after the yield
point. The yield receives a :py:class:`Result <pluggy._Result>` instance which encapsulates
the result from calling the non-wrappers. Wrappers shall not modify the result.
point. The yield receives the result from calling the non-wrappers, or raises
an exception if the non-wrappers raised.
It's possible to use ``tryfirst`` and ``trylast`` also in conjunction with
``hookwrapper=True`` in which case it will influence the ordering of hookwrappers
among each other.
It's possible to use ``tryfirst`` and ``trylast`` also on hook wrappers
in which case it will influence the ordering of hook wrappers among each other.
Declaring new hooks

2
doc/en/requirements.txt
View File

@ -1,5 +1,5 @@
pallets-sphinx-themes
pluggy>=1.0
pluggy>=1.2.0
pygments-pytest>=2.3.0
sphinx-removed-in>=0.2.0
sphinx>=5,<6

2
setup.cfg
View File

@ -46,7 +46,7 @@ py_modules = py
install_requires =
iniconfig
packaging
pluggy>=0.12,<2.0
pluggy>=1.2.0,<2.0
colorama;sys_platform=="win32"
exceptiongroup>=1.0.0rc8;python_version<"3.11"
tomli>=1.0.0;python_version<"3.11"

13
src/_pytest/assertion/__init__.py
View File

@ -112,8 +112,8 @@ def pytest_collection(session: "Session") -> None:
assertstate.hook.set_session(session)
@hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_protocol(item: Item) -> Generator[None, None, None]:
@hookimpl(wrapper=True, tryfirst=True)
def pytest_runtest_protocol(item: Item) -> Generator[None, object, object]:
"""Setup the pytest_assertrepr_compare and pytest_assertion_pass hooks.
The rewrite module will use util._reprcompare if it exists to use custom
@ -162,10 +162,11 @@ def pytest_runtest_protocol(item: Item) -> Generator[None, None, None]:
util._assertion_pass = call_assertion_pass_hook
yield
util._reprcompare, util._assertion_pass = saved_assert_hooks
util._config = None
try:
return (yield)
finally:
util._reprcompare, util._assertion_pass = saved_assert_hooks
util._config = None
def pytest_sessionfinish(session: "Session") -> None:

33
src/_pytest/cacheprovider.py
View File

@ -217,12 +217,12 @@ class LFPluginCollWrapper:
self.lfplugin = lfplugin
self._collected_at_least_one_failure = False
@hookimpl(hookwrapper=True)
def pytest_make_collect_report(self, collector: nodes.Collector):
@hookimpl(wrapper=True)
def pytest_make_collect_report(
self, collector: nodes.Collector
) -> Generator[None, CollectReport, CollectReport]:
res = yield
if isinstance(collector, (Session, Package)):
out = yield
res: CollectReport = out.get_result()
# Sort any lf-paths to the beginning.
lf_paths = self.lfplugin._last_failed_paths
@ -240,19 +240,16 @@ class LFPluginCollWrapper:
key=sort_key,
reverse=True,
)
return
elif isinstance(collector, File):
if collector.path in self.lfplugin._last_failed_paths:
out = yield
res = out.get_result()
result = res.result
lastfailed = self.lfplugin.lastfailed
# Only filter with known failures.
if not self._collected_at_least_one_failure:
if not any(x.nodeid in lastfailed for x in result):
return
return res
self.lfplugin.config.pluginmanager.register(
LFPluginCollSkipfiles(self.lfplugin), "lfplugin-collskip"
)
@ -268,8 +265,8 @@ class LFPluginCollWrapper:
# Keep all sub-collectors.
or isinstance(x, nodes.Collector)
]
return
yield
return res
class LFPluginCollSkipfiles:
@ -342,14 +339,14 @@ class LFPlugin:
else:
self.lastfailed[report.nodeid] = True
@hookimpl(hookwrapper=True, tryfirst=True)
@hookimpl(wrapper=True, tryfirst=True)
def pytest_collection_modifyitems(
self, config: Config, items: List[nodes.Item]
) -> Generator[None, None, None]:
yield
res = yield
if not self.active:
return
return res
if self.lastfailed:
previously_failed = []
@ -394,6 +391,8 @@ class LFPlugin:
else:
self._report_status += "not deselecting items."
return res
def pytest_sessionfinish(self, session: Session) -> None:
config = self.config
if config.getoption("cacheshow") or hasattr(config, "workerinput"):
@ -414,11 +413,11 @@ class NFPlugin:
assert config.cache is not None
self.cached_nodeids = set(config.cache.get("cache/nodeids", []))
@hookimpl(hookwrapper=True, tryfirst=True)
@hookimpl(wrapper=True, tryfirst=True)
def pytest_collection_modifyitems(
self, items: List[nodes.Item]
) -> Generator[None, None, None]:
yield
res = yield
if self.active:
new_items: Dict[str, nodes.Item] = {}
@ -436,6 +435,8 @@ class NFPlugin:
else:
self.cached_nodeids.update(item.nodeid for item in items)
return res
def _get_increasing_order(self, items: Iterable[nodes.Item]) -> List[nodes.Item]:
return sorted(items, key=lambda item: item.path.stat().st_mtime, reverse=True) # type: ignore[no-any-return]

49
src/_pytest/capture.py
View File

@ -36,6 +36,7 @@ from _pytest.fixtures import SubRequest
from _pytest.nodes import Collector
from _pytest.nodes import File
from _pytest.nodes import Item
from _pytest.reports import CollectReport
_CaptureMethod = Literal["fd", "sys", "no", "tee-sys"]
@ -130,8 +131,8 @@ def _windowsconsoleio_workaround(stream: TextIO) -> None:
sys.stderr = _reopen_stdio(sys.stderr, "wb")
@hookimpl(hookwrapper=True)
def pytest_load_initial_conftests(early_config: Config):
@hookimpl(wrapper=True)
def pytest_load_initial_conftests(early_config: Config) -> Generator[None, None, None]:
ns = early_config.known_args_namespace
if ns.capture == "fd":
_windowsconsoleio_workaround(sys.stdout)
@ -145,12 +146,16 @@ def pytest_load_initial_conftests(early_config: Config):
# Finally trigger conftest loading but while capturing (issue #93).
capman.start_global_capturing()
outcome = yield
capman.suspend_global_capture()
if outcome.excinfo is not None:
try:
try:
yield
finally:
capman.suspend_global_capture()
except BaseException:
out, err = capman.read_global_capture()
sys.stdout.write(out)
sys.stderr.write(err)
raise
# IO Helpers.
@ -841,41 +846,45 @@ class CaptureManager:
self.deactivate_fixture()
self.suspend_global_capture(in_=False)
out, err = self.read_global_capture()
item.add_report_section(when, "stdout", out)
item.add_report_section(when, "stderr", err)
out, err = self.read_global_capture()
item.add_report_section(when, "stdout", out)
item.add_report_section(when, "stderr", err)
# Hooks
@hookimpl(hookwrapper=True)
def pytest_make_collect_report(self, collector: Collector):
@hookimpl(wrapper=True)
def pytest_make_collect_report(
self, collector: Collector
) -> Generator[None, CollectReport, CollectReport]:
if isinstance(collector, File):
self.resume_global_capture()
outcome = yield
self.suspend_global_capture()
try:
rep = yield
finally:
self.suspend_global_capture()
out, err = self.read_global_capture()
rep = outcome.get_result()
if out:
rep.sections.append(("Captured stdout", out))
if err:
rep.sections.append(("Captured stderr", err))
else:
yield
rep = yield
return rep
@hookimpl(hookwrapper=True)
@hookimpl(wrapper=True)
def pytest_runtest_setup(self, item: Item) -> Generator[None, None, None]:
with self.item_capture("setup", item):
yield
return (yield)
@hookimpl(hookwrapper=True)
@hookimpl(wrapper=True)
def pytest_runtest_call(self, item: Item) -> Generator[None, None, None]:
with self.item_capture("call", item):
yield
return (yield)
@hookimpl(hookwrapper=True)
@hookimpl(wrapper=True)
def pytest_runtest_teardown(self, item: Item) -> Generator[None, None, None]:
with self.item_capture("teardown", item):
yield
return (yield)
@hookimpl(tryfirst=True)
def pytest_keyboard_interrupt(self) -> None:

12
src/_pytest/config/__init__.py
View File

@ -1341,12 +1341,14 @@ class Config:
else:
raise
@hookimpl(hookwrapper=True)
def pytest_collection(self) -> Generator[None, None, None]:
@hookimpl(wrapper=True)
def pytest_collection(self) -> Generator[None, object, object]:
# Validate invalid ini keys after collection is done so we take in account
# options added by late-loading conftest files.
yield
self._validate_config_options()
try:
return (yield)
finally:
self._validate_config_options()
def _checkversion(self) -> None:
import pytest
@ -1448,7 +1450,7 @@ class Config:
"""Issue and handle a warning during the "configure" stage.
During ``pytest_configure`` we can't capture warnings using the ``catch_warnings_for_item``
function because it is not possible to have hookwrappers around ``pytest_configure``.
function because it is not possible to have hook wrappers around ``pytest_configure``.
This function is mainly intended for plugins that need to issue warnings during
``pytest_configure`` (or similar stages).

6
src/_pytest/debugging.py
View File

@ -304,10 +304,10 @@ class PdbInvoke:
class PdbTrace:
@hookimpl(hookwrapper=True)
def pytest_pyfunc_call(self, pyfuncitem) -> Generator[None, None, None]:
@hookimpl(wrapper=True)
def pytest_pyfunc_call(self, pyfuncitem) -> Generator[None, object, object]:
wrap_pytest_function_for_tracing(pyfuncitem)
yield
return (yield)
def wrap_pytest_function_for_tracing(pyfuncitem):

8
src/_pytest/faulthandler.py
View File

@ -62,8 +62,8 @@ def get_timeout_config_value(config: Config) -> float:
return float(config.getini("faulthandler_timeout") or 0.0)
@pytest.hookimpl(hookwrapper=True, trylast=True)
def pytest_runtest_protocol(item: Item) -> Generator[None, None, None]:
@pytest.hookimpl(wrapper=True, trylast=True)
def pytest_runtest_protocol(item: Item) -> Generator[None, object, object]:
timeout = get_timeout_config_value(item.config)
if timeout > 0:
import faulthandler
@ -71,11 +71,11 @@ def pytest_runtest_protocol(item: Item) -> Generator[None, None, None]:
stderr = item.config.stash[fault_handler_stderr_fd_key]
faulthandler.dump_traceback_later(timeout, file=stderr)
try:
yield
return (yield)
finally:
faulthandler.cancel_dump_traceback_later()
else:
yield
return (yield)
@pytest.hookimpl(tryfirst=True)

10
src/_pytest/helpconfig.py
View File

@ -2,6 +2,7 @@
import os
import sys
from argparse import Action
from typing import Generator
from typing import List
from typing import Optional
from typing import Union
@ -97,10 +98,9 @@ def pytest_addoption(parser: Parser) -> None:
)
@pytest.hookimpl(hookwrapper=True)
def pytest_cmdline_parse():
outcome = yield
config: Config = outcome.get_result()
@pytest.hookimpl(wrapper=True)
def pytest_cmdline_parse() -> Generator[None, Config, Config]:
config = yield
if config.option.debug:
# --debug | --debug <file.log> was provided.
@ -128,6 +128,8 @@ def pytest_cmdline_parse():
config.add_cleanup(unset_tracing)
return config
def showversion(config: Config) -> None:
if config.option.version > 1:

8
src/_pytest/hookspec.py
View File

@ -60,7 +60,7 @@ def pytest_addhooks(pluginmanager: "PytestPluginManager") -> None:
:param pytest.PytestPluginManager pluginmanager: The pytest plugin manager.
.. note::
This hook is incompatible with ``hookwrapper=True``.
This hook is incompatible with hook wrappers.
"""
@ -74,7 +74,7 @@ def pytest_plugin_registered(
:param pytest.PytestPluginManager manager: pytest plugin manager.
.. note::
This hook is incompatible with ``hookwrapper=True``.
This hook is incompatible with hook wrappers.
"""
@ -113,7 +113,7 @@ def pytest_addoption(parser: "Parser", pluginmanager: "PytestPluginManager") ->
attribute or can be retrieved as the ``pytestconfig`` fixture.
.. note::
This hook is incompatible with ``hookwrapper=True``.
This hook is incompatible with hook wrappers.
"""
@ -128,7 +128,7 @@ def pytest_configure(config: "Config") -> None:
imported.
.. note::
This hook is incompatible with ``hookwrapper=True``.
This hook is incompatible with hook wrappers.
:param pytest.Config config: The pytest config object.
"""

44
src/_pytest/logging.py
View File

@ -738,27 +738,26 @@ class LoggingPlugin:
return True
@hookimpl(hookwrapper=True, tryfirst=True)
@hookimpl(wrapper=True, tryfirst=True)
def pytest_sessionstart(self) -> Generator[None, None, None]:
self.log_cli_handler.set_when("sessionstart")
with catching_logs(self.log_cli_handler, level=self.log_cli_level):
with catching_logs(self.log_file_handler, level=self.log_file_level):
yield
return (yield)
@hookimpl(hookwrapper=True, tryfirst=True)
@hookimpl(wrapper=True, tryfirst=True)
def pytest_collection(self) -> Generator[None, None, None]:
self.log_cli_handler.set_when("collection")
with catching_logs(self.log_cli_handler, level=self.log_cli_level):
with catching_logs(self.log_file_handler, level=self.log_file_level):
yield
return (yield)
@hookimpl(hookwrapper=True)
def pytest_runtestloop(self, session: Session) -> Generator[None, None, None]:
@hookimpl(wrapper=True)
def pytest_runtestloop(self, session: Session) -> Generator[None, object, object]:
if session.config.option.collectonly:
yield
return
return (yield)
if self._log_cli_enabled() and self._config.getoption("verbose") < 1:
# The verbose flag is needed to avoid messy test progress output.
@ -766,7 +765,7 @@ class LoggingPlugin:
with catching_logs(self.log_cli_handler, level=self.log_cli_level):
with catching_logs(self.log_file_handler, level=self.log_file_level):
yield # Run all the tests.
return (yield) # Run all the tests.
@hookimpl
def pytest_runtest_logstart(self) -> None:
@ -791,12 +790,13 @@ class LoggingPlugin:
item.stash[caplog_records_key][when] = caplog_handler.records
item.stash[caplog_handler_key] = caplog_handler
yield
try:
yield
finally:
log = report_handler.stream.getvalue().strip()
item.add_report_section(when, "log", log)
log = report_handler.stream.getvalue().strip()
item.add_report_section(when, "log", log)
@hookimpl(hookwrapper=True)
@hookimpl(wrapper=True)
def pytest_runtest_setup(self, item: nodes.Item) -> Generator[None, None, None]:
self.log_cli_handler.set_when("setup")
@ -804,31 +804,33 @@ class LoggingPlugin:
item.stash[caplog_records_key] = empty
yield from self._runtest_for(item, "setup")
@hookimpl(hookwrapper=True)
@hookimpl(wrapper=True)
def pytest_runtest_call(self, item: nodes.Item) -> Generator[None, None, None]:
self.log_cli_handler.set_when("call")
yield from self._runtest_for(item, "call")
@hookimpl(hookwrapper=True)
@hookimpl(wrapper=True)
def pytest_runtest_teardown(self, item: nodes.Item) -> Generator[None, None, None]:
self.log_cli_handler.set_when("teardown")
yield from self._runtest_for(item, "teardown")
del item.stash[caplog_records_key]
del item.stash[caplog_handler_key]
try:
yield from self._runtest_for(item, "teardown")
finally:
del item.stash[caplog_records_key]
del item.stash[caplog_handler_key]
@hookimpl
def pytest_runtest_logfinish(self) -> None:
self.log_cli_handler.set_when("finish")
@hookimpl(hookwrapper=True, tryfirst=True)
@hookimpl(wrapper=True, tryfirst=True)
def pytest_sessionfinish(self) -> Generator[None, None, None]:
self.log_cli_handler.set_when("sessionfinish")
with catching_logs(self.log_cli_handler, level=self.log_cli_level):
with catching_logs(self.log_file_handler, level=self.log_file_level):
yield
return (yield)
@hookimpl
def pytest_unconfigure(self) -> None:

44
src/_pytest/pytester.py
View File

@ -160,29 +160,31 @@ class LsofFdLeakChecker:
else:
return True
@hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_protocol(self, item: Item) -> Generator[None, None, None]:
@hookimpl(wrapper=True, tryfirst=True)
def pytest_runtest_protocol(self, item: Item) -> Generator[None, object, object]:
lines1 = self.get_open_files()
yield
if hasattr(sys, "pypy_version_info"):
gc.collect()
lines2 = self.get_open_files()
try:
return (yield)
finally:
if hasattr(sys, "pypy_version_info"):
gc.collect()
lines2 = self.get_open_files()
new_fds = {t[0] for t in lines2} - {t[0] for t in lines1}
leaked_files = [t for t in lines2 if t[0] in new_fds]
if leaked_files:
error = [
"***** %s FD leakage detected" % len(leaked_files),
*(str(f) for f in leaked_files),
"*** Before:",
*(str(f) for f in lines1),
"*** After:",
*(str(f) for f in lines2),
"***** %s FD leakage detected" % len(leaked_files),
"*** function %s:%s: %s " % item.location,
"See issue #2366",
]
item.warn(PytestWarning("\n".join(error)))
new_fds = {t[0] for t in lines2} - {t[0] for t in lines1}
leaked_files = [t for t in lines2 if t[0] in new_fds]
if leaked_files:
error = [
"***** %s FD leakage detected" % len(leaked_files),
*(str(f) for f in leaked_files),
"*** Before:",
*(str(f) for f in lines1),
"*** After:",
*(str(f) for f in lines2),
"***** %s FD leakage detected" % len(leaked_files),
"*** function %s:%s: %s " % item.location,
"See issue #2366",
]
item.warn(PytestWarning("\n".join(error)))
# used at least by pytest-xdist plugin

3
src/_pytest/reports.py
View File

@ -249,6 +249,9 @@ class TestReport(BaseReport):
"""
__test__ = False
# Defined by skipping plugin.
# xfail reason if xfailed, otherwise not defined. Use hasattr to distinguish.
wasxfail: str
def __init__(
self,

32
src/_pytest/setuponly.py
View File

@ -28,24 +28,26 @@ def pytest_addoption(parser: Parser) -> None:
)
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_fixture_setup(
fixturedef: FixtureDef[object], request: SubRequest
) -> Generator[None, None, None]:
yield
if request.config.option.setupshow:
if hasattr(request, "param"):
# Save the fixture parameter so ._show_fixture_action() can
# display it now and during the teardown (in .finish()).
if fixturedef.ids:
if callable(fixturedef.ids):
param = fixturedef.ids(request.param)
) -> Generator[None, object, object]:
try:
return (yield)
finally:
if request.config.option.setupshow:
if hasattr(request, "param"):
# Save the fixture parameter so ._show_fixture_action() can
# display it now and during the teardown (in .finish()).
if fixturedef.ids:
if callable(fixturedef.ids):
param = fixturedef.ids(request.param)
else:
param = fixturedef.ids[request.param_index]
else:
param = fixturedef.ids[request.param_index]
else:
param = request.param
fixturedef.cached_param = param # type: ignore[attr-defined]
_show_fixture_action(fixturedef, "SETUP")
param = request.param
fixturedef.cached_param = param # type: ignore[attr-defined]
_show_fixture_action(fixturedef, "SETUP")
def pytest_fixture_post_finalizer(fixturedef: FixtureDef[object]) -> None:

26
src/_pytest/skipping.py
View File

@ -19,6 +19,7 @@ from _pytest.outcomes import fail
from _pytest.outcomes import skip
from _pytest.outcomes import xfail
from _pytest.reports import BaseReport
from _pytest.reports import TestReport
from _pytest.runner import CallInfo
from _pytest.stash import StashKey
@ -243,7 +244,7 @@ def pytest_runtest_setup(item: Item) -> None:
xfail("[NOTRUN] " + xfailed.reason)
@hookimpl(hookwrapper=True)
@hookimpl(wrapper=True)
def pytest_runtest_call(item: Item) -> Generator[None, None, None]:
xfailed = item.stash.get(xfailed_key, None)
if xfailed is None:
@ -252,18 +253,20 @@ def pytest_runtest_call(item: Item) -> Generator[None, None, None]:
if xfailed and not item.config.option.runxfail and not xfailed.run:
xfail("[NOTRUN] " + xfailed.reason)
yield
# The test run may have added an xfail mark dynamically.
xfailed = item.stash.get(xfailed_key, None)
if xfailed is None:
item.stash[xfailed_key] = xfailed = evaluate_xfail_marks(item)
try:
return (yield)
finally:
# The test run may have added an xfail mark dynamically.
xfailed = item.stash.get(xfailed_key, None)
if xfailed is None:
item.stash[xfailed_key] = xfailed = evaluate_xfail_marks(item)
@hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item: Item, call: CallInfo[None]):
outcome = yield
rep = outcome.get_result()
@hookimpl(wrapper=True)
def pytest_runtest_makereport(
item: Item, call: CallInfo[None]
) -> Generator[None, TestReport, TestReport]:
rep = yield
xfailed = item.stash.get(xfailed_key, None)
if item.config.option.runxfail:
pass # don't interfere
@ -286,6 +289,7 @@ def pytest_runtest_makereport(item: Item, call: CallInfo[None]):
else:
rep.outcome = "passed"
rep.wasxfail = xfailed.reason
return rep
def pytest_report_teststatus(report: BaseReport) -> Optional[Tuple[str, str, str]]:

23
src/_pytest/terminal.py
View File

@ -15,7 +15,6 @@ from functools import partial
from pathlib import Path
from typing import Any
from typing import Callable
from typing import cast
from typing import ClassVar
from typing import Dict
from typing import final
@ -849,12 +848,11 @@ class TerminalReporter:
for line in doc.splitlines():
self._tw.line("{}{}".format(indent + " ", line))
@hookimpl(hookwrapper=True)
@hookimpl(wrapper=True)
def pytest_sessionfinish(
self, session: "Session", exitstatus: Union[int, ExitCode]
):
outcome = yield
outcome.get_result()
) -> Generator[None, None, None]:
result = yield
self._tw.line("")
summary_exit_codes = (
ExitCode.OK,
@ -875,17 +873,20 @@ class TerminalReporter:
elif session.shouldstop:
self.write_sep("!", str(session.shouldstop), red=True)
self.summary_stats()
return result
@hookimpl(hookwrapper=True)
@hookimpl(wrapper=True)
def pytest_terminal_summary(self) -> Generator[None, None, None]:
self.summary_errors()
self.summary_failures()
self.summary_warnings()
self.summary_passes()
yield
self.short_test_summary()
# Display any extra warnings from teardown here (if any).
self.summary_warnings()
try:
return (yield)
finally:
self.short_test_summary()
# Display any extra warnings from teardown here (if any).
self.summary_warnings()
def pytest_keyboard_interrupt(self, excinfo: ExceptionInfo[BaseException]) -> None:
self._keyboardinterrupt_memo = excinfo.getrepr(funcargs=True)
@ -1466,7 +1467,7 @@ def _get_raw_skip_reason(report: TestReport) -> str:
The string is just the part given by the user.
"""
if hasattr(report, "wasxfail"):
reason = cast(str, report.wasxfail)
reason = report.wasxfail
if reason.startswith("reason: "):
reason = reason[len("reason: ") :]
return reason

32
src/_pytest/threadexception.py
View File

@ -59,30 +59,34 @@ class catch_threading_exception:
def thread_exception_runtest_hook() -> Generator[None, None, None]:
with catch_threading_exception() as cm:
yield
if cm.args:
thread_name = "<unknown>" if cm.args.thread is None else cm.args.thread.name
msg = f"Exception in thread {thread_name}\n\n"
msg += "".join(
traceback.format_exception(
cm.args.exc_type,
cm.args.exc_value,
cm.args.exc_traceback,
try:
yield
finally:
if cm.args:
thread_name = (
"<unknown>" if cm.args.thread is None else cm.args.thread.name
)
)
warnings.warn(pytest.PytestUnhandledThreadExceptionWarning(msg))
msg = f"Exception in thread {thread_name}\n\n"
msg += "".join(
traceback.format_exception(
cm.args.exc_type,
cm.args.exc_value,
cm.args.exc_traceback,
)
)
warnings.warn(pytest.PytestUnhandledThreadExceptionWarning(msg))
@pytest.hookimpl(hookwrapper=True, trylast=True)
@pytest.hookimpl(wrapper=True, trylast=True)
def pytest_runtest_setup() -> Generator[None, None, None]:
yield from thread_exception_runtest_hook()
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
@pytest.hookimpl(wrapper=True, tryfirst=True)
def pytest_runtest_call() -> Generator[None, None, None]:
yield from thread_exception_runtest_hook()
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
@pytest.hookimpl(wrapper=True, tryfirst=True)
def pytest_runtest_teardown() -> Generator[None, None, None]:
yield from thread_exception_runtest_hook()

16
src/_pytest/tmpdir.py
View File

@ -28,7 +28,7 @@ from _pytest.fixtures import fixture
from _pytest.fixtures import FixtureRequest
from _pytest.monkeypatch import MonkeyPatch
from _pytest.nodes import Item
from _pytest.reports import CollectReport
from _pytest.reports import TestReport
from _pytest.stash import StashKey
tmppath_result_key = StashKey[Dict[str, bool]]()
@ -309,10 +309,12 @@ def pytest_sessionfinish(session, exitstatus: Union[int, ExitCode]):
cleanup_dead_symlinks(basetemp)
@hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item: Item, call):
outcome = yield
result: CollectReport = outcome.get_result()
@hookimpl(wrapper=True, tryfirst=True)
def pytest_runtest_makereport(
item: Item, call
) -> Generator[None, TestReport, TestReport]:
rep = yield
assert rep.when is not None
empty: Dict[str, bool] = {}
item.stash.setdefault(tmppath_result_key, empty)[result.when] = result.passed
item.stash.setdefault(tmppath_result_key, empty)[rep.when] = rep.passed
return rep

13
src/_pytest/unittest.py
View File

@ -376,8 +376,8 @@ def pytest_runtest_makereport(item: Item, call: CallInfo[None]) -> None:
# Twisted trial support.
@hookimpl(hookwrapper=True)
def pytest_runtest_protocol(item: Item) -> Generator[None, None, None]:
@hookimpl(wrapper=True)
def pytest_runtest_protocol(item: Item) -> Generator[None, object, object]:
if isinstance(item, TestCaseFunction) and "twisted.trial.unittest" in sys.modules:
ut: Any = sys.modules["twisted.python.failure"]
Failure__init__ = ut.Failure.__init__
@ -400,10 +400,13 @@ def pytest_runtest_protocol(item: Item) -> Generator[None, None, None]:
Failure__init__(self, exc_value, exc_type, exc_tb)
ut.Failure.__init__ = excstore
yield
ut.Failure.__init__ = Failure__init__
try:
res = yield
finally:
ut.Failure.__init__ = Failure__init__
else:
yield
res = yield
return res
def check_testcase_implements_trial_reporter(done: List[int] = []) -> None:

36
src/_pytest/unraisableexception.py
View File

@ -61,33 +61,35 @@ class catch_unraisable_exception:
def unraisable_exception_runtest_hook() -> Generator[None, None, None]:
with catch_unraisable_exception() as cm:
yield
if cm.unraisable:
if cm.unraisable.err_msg is not None:
err_msg = cm.unraisable.err_msg
else:
err_msg = "Exception ignored in"
msg = f"{err_msg}: {cm.unraisable.object!r}\n\n"
msg += "".join(
traceback.format_exception(
cm.unraisable.exc_type,
cm.unraisable.exc_value,
cm.unraisable.exc_traceback,
try:
yield
finally:
if cm.unraisable:
if cm.unraisable.err_msg is not None:
err_msg = cm.unraisable.err_msg
else:
err_msg = "Exception ignored in"
msg = f"{err_msg}: {cm.unraisable.object!r}\n\n"
msg += "".join(
traceback.format_exception(
cm.unraisable.exc_type,
cm.unraisable.exc_value,
cm.unraisable.exc_traceback,
)
)
)
warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))
warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
@pytest.hookimpl(wrapper=True, tryfirst=True)
def pytest_runtest_setup() -> Generator[None, None, None]:
yield from unraisable_exception_runtest_hook()
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
@pytest.hookimpl(wrapper=True, tryfirst=True)
def pytest_runtest_call() -> Generator[None, None, None]:
yield from unraisable_exception_runtest_hook()
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
@pytest.hookimpl(wrapper=True, tryfirst=True)
def pytest_runtest_teardown() -> Generator[None, None, None]:
yield from unraisable_exception_runtest_hook()

45
src/_pytest/warnings.py
View File

@ -60,17 +60,18 @@ def catch_warnings_for_item(
for arg in mark.args:
warnings.filterwarnings(*parse_warning_filter(arg, escape=False))
yield
for warning_message in log:
ihook.pytest_warning_recorded.call_historic(
kwargs=dict(
warning_message=warning_message,
nodeid=nodeid,
when=when,
location=None,
try:
yield
finally:
for warning_message in log:
ihook.pytest_warning_recorded.call_historic(
kwargs=dict(
warning_message=warning_message,
nodeid=nodeid,
when=when,
location=None,
)
)
)
def warning_record_to_str(warning_message: warnings.WarningMessage) -> str:
@ -103,24 +104,24 @@ def warning_record_to_str(warning_message: warnings.WarningMessage) -> str:
return msg
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_runtest_protocol(item: Item) -> Generator[None, None, None]:
@pytest.hookimpl(wrapper=True, tryfirst=True)
def pytest_runtest_protocol(item: Item) -> Generator[None, object, object]:
with catch_warnings_for_item(
config=item.config, ihook=item.ihook, when="runtest", item=item
):
yield
return (yield)
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_collection(session: Session) -> Generator[None, None, None]:
@pytest.hookimpl(wrapper=True, tryfirst=True)
def pytest_collection(session: Session) -> Generator[None, object, object]:
config = session.config
with catch_warnings_for_item(
config=config, ihook=config.hook, when="collect", item=None
):
yield
return (yield)
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_terminal_summary(
terminalreporter: TerminalReporter,
) -> Generator[None, None, None]:
@ -128,23 +129,23 @@ def pytest_terminal_summary(
with catch_warnings_for_item(
config=config, ihook=config.hook, when="config", item=None
):
yield
return (yield)
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_sessionfinish(session: Session) -> Generator[None, None, None]:
config = session.config
with catch_warnings_for_item(
config=config, ihook=config.hook, when="config", item=None
):
yield
return (yield)
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_load_initial_conftests(
early_config: "Config",
) -> Generator[None, None, None]:
with catch_warnings_for_item(
config=early_config, ihook=early_config.hook, when="config", item=None
):
yield
return (yield)

9
testing/conftest.py
View File

@ -1,6 +1,7 @@
import dataclasses
import re
import sys
from typing import Generator
from typing import List
import pytest
@ -21,11 +22,11 @@ if sys.gettrace():
sys.settrace(orig_trace)
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
def pytest_collection_modifyitems(items):
@pytest.hookimpl(wrapper=True, tryfirst=True)
def pytest_collection_modifyitems(items) -> Generator[None, None, None]:
"""Prefer faster tests.
Use a hookwrapper to do this in the beginning, so e.g. --ff still works
Use a hook wrapper to do this in the beginning, so e.g. --ff still works
correctly.
"""
fast_items = []
@ -62,7 +63,7 @@ def pytest_collection_modifyitems(items):
items[:] = fast_items + neutral_items + slow_items + slowest_items
yield
return (yield)
@pytest.fixture

4
testing/logging/test_reporting.py
View File

@ -1040,13 +1040,13 @@ def test_log_set_path(pytester: Pytester) -> None:
"""
import os
import pytest
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
@pytest.hookimpl(wrapper=True, tryfirst=True)
def pytest_runtest_setup(item):
config = item.config
logging_plugin = config.pluginmanager.get_plugin("logging-plugin")
report_file = os.path.join({}, item._request.node.name)
logging_plugin.set_log_path(report_file)
yield
return (yield)
""".format(
repr(report_dir_base)
)

19
testing/python/collect.py
View File

@ -827,11 +827,11 @@ class TestConftestCustomization:
textwrap.dedent(
"""\
import pytest
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_pycollect_makemodule():
outcome = yield
mod = outcome.get_result()
mod = yield
mod.obj.hello = "world"
return mod
"""
),
encoding="utf-8",
@ -855,14 +855,13 @@ class TestConftestCustomization:
textwrap.dedent(
"""\
import pytest
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_pycollect_makeitem():
outcome = yield
if outcome.excinfo is None:
result = outcome.get_result()
if result:
for func in result:
func._some123 = "world"
result = yield
if result:
for func in result:
func._some123 = "world"
return result
"""
),
encoding="utf-8",

7
testing/test_collection.py
View File

@ -334,12 +334,11 @@ class TestPrunetraceback:
pytester.makeconftest(
"""
import pytest
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_make_collect_report():
outcome = yield
rep = outcome.get_result()
rep = yield
rep.headerlines += ["header1"]
outcome.force_result(rep)
return rep
"""
)
result = pytester.runpytest(p)

2
testing/test_config.py
View File

@ -1317,7 +1317,7 @@ def test_load_initial_conftest_last_ordering(_config_for_test):
hookimpls = [
(
hookimpl.function.__module__,
"wrapper" if hookimpl.hookwrapper else "nonwrapper",
"wrapper" if (hookimpl.wrapper or hookimpl.hookwrapper) else "nonwrapper",
)
for hookimpl in hc.get_hookimpls()
]

6
testing/test_mark.py
View File

@ -806,12 +806,12 @@ class TestKeywordSelection:
pytester.makepyfile(
conftest="""
import pytest
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_pycollect_makeitem(name):
outcome = yield
item = yield
if name == "TestClass":
item = outcome.get_result()
item.extra_keyword_matches.add("xxx")
return item
"""
)
reprec = pytester.inline_run(p.parent, "-s", "-k", keyword)

12
testing/test_python_path.py
View File

@ -85,8 +85,8 @@ def test_clean_up(pytester: Pytester) -> None:
# This is tough to test behaviorally because the cleanup really runs last.
# So the test make several implementation assumptions:
# - Cleanup is done in pytest_unconfigure().
# - Not a hookwrapper.
# So we can add a hookwrapper ourselves to test what it does.
# - Not a hook wrapper.
# So we can add a hook wrapper ourselves to test what it does.
pytester.makefile(".ini", pytest="[pytest]\npythonpath=I_SHALL_BE_REMOVED\n")
pytester.makepyfile(test_foo="""def test_foo(): pass""")
@ -94,12 +94,14 @@ def test_clean_up(pytester: Pytester) -> None:
after: Optional[List[str]] = None
class Plugin:
@pytest.hookimpl(hookwrapper=True, tryfirst=True)
@pytest.hookimpl(wrapper=True, tryfirst=True)
def pytest_unconfigure(self) -> Generator[None, None, None]:
nonlocal before, after
before = sys.path.copy()
yield
after = sys.path.copy()
try:
return (yield)
finally:
after = sys.path.copy()
result = pytester.runpytest_inprocess(plugins=[Plugin()])
assert result.ret == 0

10
testing/test_runner.py
View File

@ -542,10 +542,10 @@ def test_runtest_in_module_ordering(pytester: Pytester) -> None:
@pytest.fixture
def mylist(self, request):
return request.function.mylist
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_runtest_call(self, item):
try:
(yield).get_result()
yield
except ValueError:
pass
def test_hello1(self, mylist):
@ -826,12 +826,12 @@ def test_unicode_in_longrepr(pytester: Pytester) -> None:
pytester.makeconftest(
"""\
import pytest
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_runtest_makereport():
outcome = yield
rep = outcome.get_result()
rep = yield
if rep.when == "call":
rep.longrepr = 'ä'
return rep
"""
)
pytester.makepyfile(

4
testing/test_terminal.py
View File

@ -725,12 +725,12 @@ class TestTerminalFunctional:
)
assert result.ret == 0
def test_deselected_with_hookwrapper(self, pytester: Pytester) -> None:
def test_deselected_with_hook_wrapper(self, pytester: Pytester) -> None:
pytester.makeconftest(
"""
import pytest
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_collection_modifyitems(config, items):
yield
deselected = items.pop()

2
testing/test_unittest.py
View File

@ -952,7 +952,7 @@ def test_issue333_result_clearing(pytester: Pytester) -> None:
pytester.makeconftest(
"""
import pytest
@pytest.hookimpl(hookwrapper=True)
@pytest.hookimpl(wrapper=True)
def pytest_runtest_call(item):
yield
assert 0