diff --git a/changelog.d/215.added.rst b/changelog.d/215.added.rst new file mode 100644 index 00000000..d9ca5d2d --- /dev/null +++ b/changelog.d/215.added.rst @@ -0,0 +1 @@ +Automatically deliver pytest-timeout signal failures cooperatively in asynchronous tests and fixtures on Python 3.11 and newer when pytest-timeout provides its expiry hook. diff --git a/docs/reference/index.rst b/docs/reference/index.rst index 5c3095f7..eb497b4f 100644 --- a/docs/reference/index.rst +++ b/docs/reference/index.rst @@ -9,6 +9,7 @@ Reference fixtures/index functions hooks + timeouts markers/index decorators/index changelog diff --git a/docs/reference/timeouts.rst b/docs/reference/timeouts.rst new file mode 100644 index 00000000..c5b675cc --- /dev/null +++ b/docs/reference/timeouts.rst @@ -0,0 +1,30 @@ +======== +Timeouts +======== + +On Python 3.11 and newer, pytest-asyncio automatically delivers pytest-timeout +signal failures by cancelling the active asynchronous test or fixture when +pytest-timeout provides the ``pytest_timeout_expired`` hook. Cooperative cleanup +can then run before pytest reports the original timeout. No configuration is +needed. Python 3.10 and older pytest-timeout versions retain their existing +signal behavior. Cooperative test execution requires a native ``async def`` +function, bound method, or plain ``functools.partial`` of one. Synchronous +coroutine creators, including functions marked with +``inspect.markcoroutinefunction()`` and subclasses of ``functools.partial``, +retain their existing call timing, context, and synchronous signal delivery. +pytest-timeout still controls the configured duration, covered test phases, +debugger detection, and timeout diagnostics. + +Cancellation waits for the event loop and coroutine to cooperate. A raised +signal exception can interrupt CPU-bound Python code, but cooperative +cancellation cannot interrupt code that never yields. It also cannot stop +an indefinitely blocking callback or a task that refuses cancellation. +Use pytest-timeout's ``thread`` method or an independent process watchdog when +the process must be terminated; these stop the entire process without normal +test teardown. A timeout during final event-loop shutdown stops that shutdown +and closes the loop; remaining resource cleanup may be incomplete. + +The integration does not take over runners managed by other async plugins or +synchronous tests that call ``asyncio.run()`` themselves. It wraps asynchronous +tests and fixtures in a timeout context, so ``asyncio.current_task().get_coro()`` +returns the wrapper coroutine rather than the original test coroutine. diff --git a/pyproject.toml b/pyproject.toml index f8dd729d..b510043e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,7 +51,7 @@ optional-dependencies.docs = [ "sphinx-tabs>=3.5", ] optional-dependencies.testing = [ - "coverage>=6.2", + "coverage>=7.10.3", "hypothesis>=5.7.1", ] urls."Bug Tracker" = "https://github.com/pytest-dev/pytest-asyncio/issues" @@ -70,7 +70,7 @@ dev = [ { include-group = "typing" }, ] test = [ - "coverage>=6.2", + "coverage>=7.10.3", "hypothesis>=5.7.1", ] docs = [ @@ -160,13 +160,22 @@ filterwarnings = [ ] [tool.coverage.run] -source = [ +source_pkgs = [ "pytest_asyncio", ] +patch = [ + "subprocess", +] branch = true data_file = "coverage/coverage" parallel = true +[tool.coverage.paths] +source = [ + "pytest_asyncio", + "*/site-packages/pytest_asyncio", +] + [tool.coverage.report] show_missing = true diff --git a/pytest_asyncio/_timeout.py b/pytest_asyncio/_timeout.py new file mode 100644 index 00000000..e6021b4a --- /dev/null +++ b/pytest_asyncio/_timeout.py @@ -0,0 +1,130 @@ +"""Cooperative delivery of pytest-timeout's signal failures.""" + +from __future__ import annotations + +import asyncio +import contextvars +import sys +import threading +from collections.abc import Callable, Coroutine +from dataclasses import dataclass +from typing import Any, TypeVar + +import pytest + +__tracebackhide__ = True +_T = TypeVar("_T") + + +@dataclass +class _Delivery: + config: pytest.Config + loop: asyncio.AbstractEventLoop + closing: bool = False + exception: BaseException | None = None + timeout: asyncio.Timeout | None = None + + def run(self, operation: Callable[[], _T]) -> _T: + previous = self.config.stash.get(_CURRENT_DELIVERY, None) + try: + try: + self.config.stash[_CURRENT_DELIVERY] = self + result = operation() + finally: + # Once the runner returns, a new signal can fail synchronously. + # Stop claiming it before deciding which outcome to propagate. + self.config.stash[_CURRENT_DELIVERY] = previous + except (KeyboardInterrupt, SystemExit, pytest.exit.Exception): + raise + except asyncio.CancelledError as exc: + if self.exception is None: + raise + # asyncio.Timeout converts only its own cancellation to TimeoutError. + # Preserve cancellation requested by another caller. + raise exc from self.exception + except BaseException as exc: + if self.exception is None or exc is self.exception: + raise + raise self.exception from exc + if self.exception is not None: + raise self.exception + return result + + def interrupt(self) -> None: + if self.config.stash.get(_CURRENT_DELIVERY, None) is not self: + return + if self.closing: + # A completed shutdown phase can consume stop(). Keep stopping + # until Runner.close() returns; never stop a reusable invocation. + self.loop.stop() + self.loop.call_soon(self.interrupt) + elif self.timeout is not None: + self.timeout.reschedule(self.loop.time()) + + +# SIGALRM only reaches the main thread; worker runners keep their native behavior. +_CURRENT_DELIVERY = pytest.StashKey[_Delivery | None]() + + +def _supports_cooperative_timeouts(config: pytest.Config) -> bool: + # Test modules can load pytest-timeout after pytest_configure has run. + return ( + sys.version_info >= (3, 11) + and threading.current_thread() is threading.main_thread() + and config.hook.pytest_timeout_expired.has_spec() + ) + + +@pytest.hookimpl(tryfirst=True, optionalhook=True) +def pytest_timeout_expired(item: pytest.Item, exception: BaseException) -> bool | None: + if threading.current_thread() is not threading.main_thread(): + return None + invocation = item.config.stash.get(_CURRENT_DELIVERY, None) + if invocation is None: + return None + if invocation.exception is None: + invocation.exception = exception + # Raising here can interrupt asyncio before it schedules a task's next + # step. Return to the interrupted code and cancel at a safe loop turn. + # Late callbacks check ownership instead of relying on Handle.cancel(): + # SIGINT can interrupt scheduling before the handle is returned. + if not invocation.loop.is_closed(): + invocation.loop.call_soon_threadsafe(invocation.interrupt) + return True + + +def run( + runner: asyncio.Runner, + coro_factory: Callable[[], Coroutine[Any, Any, _T]], + *, + context: contextvars.Context, + config: pytest.Config, +) -> _T: + """Run a native coroutine factory with cooperative timeout delivery.""" + if not _supports_cooperative_timeouts(config): + return runner.run(coro_factory(), context=context) + + invocation = _Delivery(config, runner.get_loop()) + + async def invoke() -> _T: + if invocation.exception is not None: + raise invocation.exception + try: + async with asyncio.timeout(None) as timeout: + invocation.timeout = timeout + # Create the user coroutine only once its task owns execution. + # Runner and task factories retain ownership of invoke(). + return await coro_factory() + finally: + # The signal may have queued delivery just as the coroutine exits. + # Do not reschedule a timeout whose context has already exited. + invocation.timeout = None + + return invocation.run(lambda: runner.run(invoke(), context=context)) + + +def close(runner: asyncio.Runner, *, config: pytest.Config) -> None: + if not _supports_cooperative_timeouts(config): + runner.close() + return + _Delivery(config, runner.get_loop(), closing=True).run(runner.close) diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 38b75e41..85ac9cc4 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -54,6 +54,12 @@ PytestPluginManager, ) +from ._timeout import ( + close as _close_with_timeout, + pytest_timeout_expired as pytest_timeout_expired, + run as _run_with_timeout, +) + if sys.version_info >= (3, 11): from asyncio import Runner else: @@ -405,7 +411,9 @@ async def setup(): return res context = contextvars.copy_context() - result = runner.run(setup(), context=context) + result = _run_with_timeout( + runner, setup, context=context, config=request.config + ) reset_contextvars = _apply_contextvar_changes(context) @@ -422,7 +430,9 @@ async def async_finalizer() -> None: msg += "Yield only once." raise ValueError(msg) - runner.run(async_finalizer(), context=context) + _run_with_timeout( + runner, async_finalizer, context=context, config=request.config + ) if reset_contextvars is not None: reset_contextvars() @@ -453,7 +463,9 @@ async def setup(): return res context = contextvars.copy_context() - result = runner.run(setup(), context=context) + result = _run_with_timeout( + runner, setup, context=context, config=request.config + ) # Copy the context vars modified by the setup task into the current # context, and (if needed) add a finalizer to reset them. @@ -563,7 +575,7 @@ def runtest(self) -> None: runner = self._request.getfixturevalue(runner_fixture_id) context = contextvars.copy_context() synchronized_obj = _synchronize_coroutine( - getattr(*self._synchronization_target_attr), runner, context + getattr(*self._synchronization_target_attr), runner, context, self.config ) with MonkeyPatch.context() as c: c.setattr(*self._synchronization_target_attr, synchronized_obj) @@ -890,10 +902,22 @@ def pytest_pyfunc_call(pyfuncitem: Function) -> object | None: return None +def _is_native_coroutine_function(func: object) -> bool: + # Partial subclasses can override __call__; binding them can discard it. + if type(func) is functools.partial: + return _is_native_coroutine_function(func.func) + if inspect.ismethod(func): + return _is_native_coroutine_function(func.__func__) + return inspect.isfunction(func) and bool( + func.__code__.co_flags & inspect.CO_COROUTINE + ) + + def _synchronize_coroutine( func: Callable[..., CoroutineType], runner: asyncio.Runner, context: contextvars.Context, + config: Config, ): """ Return a sync wrapper around a coroutine executing it in the @@ -902,8 +926,17 @@ def _synchronize_coroutine( @functools.wraps(func) def inner(*args, **kwargs): - coro = func(*args, **kwargs) - runner.run(coro, context=context) + if not _is_native_coroutine_function(func): + # Synchronous creators must run in the caller's context, even when + # inspect.markcoroutinefunction() marks them as coroutine functions. + runner.run(func(*args, **kwargs), context=context) + return + _run_with_timeout( + runner, + functools.partial(func, *args, **kwargs), + context=context, + config=config, + ) return inner @@ -1042,15 +1075,15 @@ def _scoped_runner( _set_event_loop(runner.get_loop()) try: yield runner - except Exception as e: - runner.__exit__(type(e), e, e.__traceback__) + except Exception: + _close_with_timeout(runner, config=request.config) else: with warnings.catch_warnings(): warnings.filterwarnings( "ignore", ".*BaseEventLoop.shutdown_asyncgens.*", RuntimeWarning ) try: - runner.__exit__(None, None, None) + _close_with_timeout(runner, config=request.config) except RuntimeError: warnings.warn( _RUNNER_TEARDOWN_WARNING % traceback.format_exc(), diff --git a/tests/test_timeout.py b/tests/test_timeout.py new file mode 100644 index 00000000..1d575f56 --- /dev/null +++ b/tests/test_timeout.py @@ -0,0 +1,329 @@ +from __future__ import annotations + +import asyncio +import contextvars +import functools +import inspect +import signal +import sys +from collections.abc import Callable +from textwrap import dedent +from types import CoroutineType +from typing import Literal + +import pytest +from pytest import Pytester + +from pytest_asyncio._timeout import pytest_timeout_expired, run +from pytest_asyncio.plugin import _synchronize_coroutine + +pytestmark = pytest.mark.skipif( + not hasattr(signal, "SIGALRM"), reason="requires the signal timeout method" +) + + +@pytest.fixture +def timeout_plugin(request: pytest.FixtureRequest): + if not request.config.hook.pytest_timeout_expired.has_spec(): + pytest.skip("requires pytest-timeout's signal-expiry hook") + + +@pytest.fixture +def cooperative_timeout(timeout_plugin: None): + if sys.version_info < (3, 11): + pytest.skip("cooperative timeouts require Python 3.11 or newer") + + +@pytest.mark.parametrize("event", ["timeout", "interrupt"]) +def test_startup(event: str, request: pytest.FixtureRequest, cooperative_timeout: None): + expired = pytest.fail.Exception("original timeout") + entered = [] + tasks = [] + + async def body(): + entered.append(True) + return 42 + + def task_factory(loop, coro, **kwargs): + loop.set_task_factory(None) + + async def traced(): + return await coro + + task = loop.create_task(traced(), **kwargs) + tasks.append(task) + if event == "interrupt": + signal.raise_signal(signal.SIGINT) + pytest_timeout_expired(request.node, expired) + return task + + with asyncio.Runner() as runner: + runner.get_loop().set_task_factory(task_factory) + expected = KeyboardInterrupt if event == "interrupt" else type(expired) + with pytest.raises(expected) as caught: + run(runner, body, context=contextvars.copy_context(), config=request.config) + if event == "interrupt": + runner.run(asyncio.sleep(0)) + assert tasks[0].result() == 42 + else: + assert caught.value is expired + assert not entered + + +@pytest.mark.parametrize("kind", ["synchronous", "partial_subclass"]) +def test_creator_context( + kind: str, request: pytest.FixtureRequest, cooperative_timeout: None +): + value = contextvars.ContextVar("value", default="caller") + events = [] + + async def body(argument): + events.append(("body", value.get(), argument)) + value.set("updated") + + def creator(argument): + events.append(("creator", value.get())) + return body(argument) + + if hasattr(inspect, "markcoroutinefunction"): + inspect.markcoroutinefunction(creator) + + class SyncPartial(functools.partial): + def __call__(self, *args, **kwargs): + events.append(("creator", value.get())) + return super().__call__(*args, **kwargs) + + functions: dict[str, Callable[..., CoroutineType]] = { + "synchronous": creator, + "partial_subclass": SyncPartial(body), + } + context = contextvars.copy_context() + context.run(value.set, "task") + with asyncio.Runner() as runner: + synchronized = _synchronize_coroutine( + functions[kind], runner, context, request.config + ) + synchronized(42) + assert events == [("creator", "caller"), ("body", "task", 42)] + assert value.get() == "caller" + assert context.get(value) == "updated" + + +@pytest.mark.parametrize( + ("cleanup", "expected"), + [ + (None, pytest.fail.Exception), + (pytest.xfail.Exception("cleanup xfail"), pytest.fail.Exception), + (ValueError("cleanup error"), pytest.fail.Exception), + ("interrupt", KeyboardInterrupt), + ( + pytest.exit.Exception("requested exit", returncode=4), + pytest.exit.Exception, + ), + (SystemExit(7), SystemExit), + ("cancel", asyncio.CancelledError), + ], +) +def test_cleanup( + cleanup: BaseException | Literal["interrupt", "cancel"] | None, + expected: type[BaseException], + request: pytest.FixtureRequest, + cooperative_timeout: None, +): + expired = pytest.fail.Exception("original timeout") + cancelled = [] + + async def body(): + loop = asyncio.get_running_loop() + loop.call_soon(pytest_timeout_expired, request.node, expired) + try: + # Allow expiry, delivery, and cancellation turns, but never hang if + # delivery breaks. The assertion below verifies cancellation ran. + for _ in range(5): + await asyncio.sleep(0) + except asyncio.CancelledError as cancellation: + cancelled.append(True) + if cleanup == "interrupt": + signal.raise_signal(signal.SIGINT) + elif cleanup == "cancel": + task = asyncio.current_task() + assert task is not None + task.cancel() + elif cleanup is not None: + raise cleanup from cancellation + await asyncio.sleep(0) + + with asyncio.Runner() as runner, pytest.raises(expected) as caught: + run(runner, body, context=contextvars.copy_context(), config=request.config) + assert cancelled + if cleanup == "cancel": + assert caught.value.__cause__ is expired + elif cleanup != "interrupt": + assert caught.value is ( + expired if expected is pytest.fail.Exception else cleanup + ) + + +def test_signal_timeout_preserves_shared_loop( + pytester: Pytester, cooperative_timeout: None, monkeypatch: pytest.MonkeyPatch +): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import asyncio + import functools + import signal + import time + import pytest + + pytest_plugins = "pytest_timeout" + + cleaned = [] + + async def application_wait(): + await asyncio.Future() + + @pytest.mark.timeout(0.1, method="signal", func_only=True) + @pytest.mark.asyncio(loop_scope="module") + async def timeout(trigger="timer"): + loop = asyncio.get_running_loop() + original = loop.call_soon + task = asyncio.current_task() + + def reschedule(callback, *args, context=None): + if getattr(callback, "__self__", None) is task: + loop.call_soon = original + signal.raise_signal(signal.SIGALRM) + return original(callback, *args, context=context) + + if trigger == "reschedule": + loop.call_soon = reschedule + else: + loop.call_soon(time.sleep, 0.2) + try: + await asyncio.sleep(0) + await application_wait() + finally: + loop.call_soon = original + cleaned.append(trigger) + + test_timer = functools.wraps(timeout)(functools.partial(timeout)) + + class TestTimeout: + @pytest.mark.timeout(10, method="signal", func_only=True) + @pytest.mark.asyncio(loop_scope="module") + async def test_timeout(self): + await timeout("reschedule") + + @pytest.mark.asyncio(loop_scope="module") + async def test_later(): + assert cleaned == ["timer", "reschedule"] + + @pytest.mark.timeout(10, method="signal", func_only=True) + def test_synchronous(): + with pytest.raises(pytest.fail.Exception, match="Timeout"): + signal.raise_signal(signal.SIGALRM) + """)) + monkeypatch.setenv("PYTEST_DISABLE_PLUGIN_AUTOLOAD", "1") + plugins = ["-p", "pytest_asyncio.plugin"] + if pytest.version_tuple < (9, 1): + # Older pytest cannot initialize pytest-timeout's options when loaded late. + plugins.extend(("-p", "pytest_timeout")) + result = pytester.runpytest_subprocess(*plugins, "--tb=short", timeout=10) + result.assert_outcomes(failed=2, passed=2) + result.stdout.fnmatch_lines(["E *Failed: Timeout*from pytest-timeout.*"] * 2) + result.stdout.fnmatch_lines(["*in application_wait*", "*CancelledError*"]) + assert "_timeout.py" not in result.stdout.str() + + +def test_timeout_during_async_cleanup(pytester: Pytester, cooperative_timeout: None): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import asyncio + import signal + import threading + from concurrent.futures import ThreadPoolExecutor + import pytest + import pytest_asyncio + + executor = None + worker = None + release = threading.Event() + + async def timeout(): + asyncio.get_running_loop().call_soon(signal.raise_signal, signal.SIGALRM) + await asyncio.Future() + + async def background(): + try: + await asyncio.Future() + finally: + await timeout() + + @pytest_asyncio.fixture + async def coroutine(phase): + if phase == "coroutine_setup": + await timeout() + + @pytest_asyncio.fixture + async def fixture(phase): + if phase == "generator_setup": + await timeout() + yield + if phase == "teardown": + await timeout() + + @pytest.mark.parametrize( + "phase", + [ + "coroutine_setup", "generator_setup", "teardown", + "shutdown", "shutdown_boundary", + ], + ) + @pytest.mark.timeout(10, method="signal") + @pytest.mark.asyncio + async def test_timeout(phase, coroutine, fixture): + global executor, worker + if phase == "shutdown": + asyncio.create_task(background()) + await asyncio.sleep(0) + if phase == "shutdown_boundary": + loop = asyncio.get_running_loop() + executor = ThreadPoolExecutor() + loop.set_default_executor(executor) + worker = executor.submit(release.wait, 5) + original = loop.shutdown_asyncgens + + async def shutdown_asyncgens(): + await original() + signal.raise_signal(signal.SIGALRM) + + loop.shutdown_asyncgens = shutdown_asyncgens + + def test_later(): + if executor is not None: + try: + assert not worker.done() + finally: + release.set() + executor.shutdown(wait=True) + """)) + result = pytester.runpytest_subprocess(timeout=10) + result.assert_outcomes(errors=5, passed=4) + result.stdout.fnmatch_lines(["E *Failed: Timeout*from pytest-timeout.*"] * 5) + + +@pytest.mark.skipif(sys.version_info >= (3, 11), reason="legacy Python 3.10 behavior") +def test_signal_timeout_on_python310(pytester: Pytester, timeout_plugin: None): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import signal + import pytest + + @pytest.mark.timeout(10, method="signal", func_only=True) + @pytest.mark.asyncio + async def test_signal(): + with pytest.raises(pytest.fail.Exception, match="Timeout"): + signal.raise_signal(signal.SIGALRM) + """)) + result = pytester.runpytest_subprocess(timeout=10) + result.assert_outcomes(passed=1) diff --git a/tox.ini b/tox.ini index 21ba3ead..eab431ae 100644 --- a/tox.ini +++ b/tox.ini @@ -2,7 +2,7 @@ minversion = 4.28.0 requires = tox-uv>=1.25 -envlist = build, py310, py311, py312, py313, py314, py310-lower-bounds, docs, pyright +envlist = build, py310, py311, py312, py313, py314, py314t, py310-lower-bounds, pytest-timeout-dev, docs, pyright isolated_build = true passenv = CI @@ -36,6 +36,18 @@ runner = uv-venv-runner deps = pytest @ git+https://github.com/pytest-dev/pytest.git +[testenv:pytest-timeout-dev] +description = Run tests against the proposed pytest-timeout expiry hook +runner = uv-venv-runner +# Replace this prerequisite with a released test dependency once available. +deps = + pytest-timeout @ git+https://github.com/tamird/pytest-timeout.git@2675aa33b2856118109fbd4f250dd60ca8522e7a + # Cover the pytest floor where cooperative delivery first becomes available. + pytest==8.4.0; python_version == "3.11" +commands = + python -c "import pytest_timeout; assert hasattr(pytest_timeout.TimeoutHooks, 'pytest_timeout_expired')" + make test + [testenv:docs] allowlist_externals = git @@ -84,9 +96,10 @@ commands = pyright pytest_asyncio/ tests/ [gh-actions] python = - 3.10: py310, py310-lower-bounds, build - 3.11: py311 + 3.10: py310, py310-lower-bounds, build, pytest-timeout-dev + 3.11: py311, pytest-timeout-dev 3.12: py312 3.13: py313, pyright - 3.14-dev: py314 + 3.14: py314, pytest-timeout-dev + 3.14t: py314t, pytest-timeout-dev pypy3: pypy3 diff --git a/uv.lock b/uv.lock index ab545d10..8569e7b2 100644 --- a/uv.lock +++ b/uv.lock @@ -1155,7 +1155,7 @@ typing = [ [package.metadata] requires-dist = [ { name = "backports-asyncio-runner", marker = "python_full_version < '3.11'", specifier = ">=1.1,<2" }, - { name = "coverage", marker = "extra == 'testing'", specifier = ">=6.2" }, + { name = "coverage", marker = "extra == 'testing'", specifier = ">=7.10.3" }, { name = "hypothesis", marker = "extra == 'testing'", specifier = ">=5.7.1" }, { name = "pytest", specifier = ">=8.4,<10" }, { name = "sphinx", marker = "extra == 'docs'", specifier = ">=5.3" }, @@ -1174,7 +1174,7 @@ build = [ dev = [ { name = "build" }, { name = "check-wheel-contents" }, - { name = "coverage", specifier = ">=6.2" }, + { name = "coverage", specifier = ">=7.10.3" }, { name = "hypothesis", specifier = ">=5.7.1" }, { name = "pre-commit" }, { name = "pyright", extras = ["nodejs"] }, @@ -1190,7 +1190,7 @@ docs = [ { name = "sphinx-tabs", specifier = ">=3.5" }, ] test = [ - { name = "coverage", specifier = ">=6.2" }, + { name = "coverage", specifier = ">=7.10.3" }, { name = "hypothesis", specifier = ">=5.7.1" }, ] typing = [