From 58aed6442f2238000b60767912875b535c530a82 Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Wed, 19 Aug 2026 19:38:35 -0400 Subject: [PATCH 01/12] Add cooperative signal timeouts A signal timeout can interrupt asyncio while it schedules a task's next step, leaving the task pending without a continuation [1]. Catching the resulting pytest failure and cancelling that task cannot reliably recover it. Add opt-in cooperative delivery through pytest-timeout's proposed expiry hook [2]. Queue cancellation without raising into the scheduler, then report the supplied failure at the owned runner boundary. Preserve external cancellation, process-control exceptions, and the interrupted await's traceback. Stop final runner shutdown separately when it times out; ordinary shared-runner invocations remain reusable. Keep timer selection, deadlines, debugger handling, and diagnostics in pytest-timeout. Existing signal behavior remains the default. Cooperative cancellation cannot stop blocking or cancellation-resistant code, so the thread method remains the hard process-stop option. Final runner shutdown may leave resource cleanup incomplete. Pin the producer prerequisite in a dedicated integration environment until its hook is released. Collect coverage from the isolated pytest processes and map installed-wheel paths back to the source tree. Require Coverage.py 7.10.3 for its subprocess-directory fixes [3]. [1]: https://github.com/pytest-dev/pytest-timeout/issues/113 [2]: https://github.com/pytest-dev/pytest-timeout/pull/204 [3]: https://coverage.readthedocs.io/en/7.10.3/changes.html --- changelog.d/215.added.rst | 1 + docs/reference/configuration.rst | 7 + docs/reference/index.rst | 1 + docs/reference/timeouts.rst | 28 +++ pyproject.toml | 15 +- pytest_asyncio/_timeout.py | 175 +++++++++++++++++ pytest_asyncio/plugin.py | 42 +++- tests/test_timeout.py | 320 +++++++++++++++++++++++++++++++ tox.ini | 16 +- uv.lock | 6 +- 10 files changed, 594 insertions(+), 17 deletions(-) create mode 100644 changelog.d/215.added.rst create mode 100644 docs/reference/timeouts.rst create mode 100644 pytest_asyncio/_timeout.py create mode 100644 tests/test_timeout.py diff --git a/changelog.d/215.added.rst b/changelog.d/215.added.rst new file mode 100644 index 00000000..c25324f7 --- /dev/null +++ b/changelog.d/215.added.rst @@ -0,0 +1 @@ +Add opt-in cooperative delivery of pytest-timeout signal failures in asynchronous tests and fixtures. diff --git a/docs/reference/configuration.rst b/docs/reference/configuration.rst index 5341f02a..7775c955 100644 --- a/docs/reference/configuration.rst +++ b/docs/reference/configuration.rst @@ -56,3 +56,10 @@ The value can also be set via the ``--asyncio-mode`` command-line option: If the asyncio mode is set in both the pytest configuration file and the command-line option, the command-line option takes precedence. If no asyncio mode is specified, the mode defaults to `strict`. + +asyncio_cooperative_timeouts +============================ + +Enable cooperative delivery of pytest-timeout signal failures. This is disabled +by default and can also be enabled with ``--asyncio-cooperative-timeouts``. +See :doc:`timeouts` for the required plugin support and cancellation limits. 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..d2d9cd18 --- /dev/null +++ b/docs/reference/timeouts.rst @@ -0,0 +1,28 @@ +======== +Timeouts +======== + +Enable ``--asyncio-cooperative-timeouts`` or set +``asyncio_cooperative_timeouts = true`` to deliver pytest-timeout signal +failures by cancelling the active asynchronous test or fixture. This requires +a pytest-timeout version providing ``pytest_timeout_expired``. Cooperative +cleanup can then run before pytest reports the original timeout. +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. It cannot +stop a callback that blocks indefinitely or a task that refuses cancellation. +Use pytest-timeout's ``thread`` method or an independent process watchdog when +the process must be terminated. A timeout during final event-loop shutdown +stops that shutdown and closes the loop; remaining resource cleanup may be +incomplete. + +Custom Python 3.10 task factories may return tasks without cancellation +counters. For those tasks, an escaping ``CancelledError`` is preserved and +the timeout is chained to it, because the runner cannot reliably distinguish +timeout cancellation from another cancellation request. + +Cooperative timeouts are disabled by default. Enabling them without a +compatible pytest-timeout plugin is a configuration error. The integration +does not take over runners managed by other async plugins or synchronous tests +that call ``asyncio.run()`` themselves. 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..3ab2fedd --- /dev/null +++ b/pytest_asyncio/_timeout.py @@ -0,0 +1,175 @@ +"""Cooperative delivery of pytest-timeout's signal failures.""" + +from __future__ import annotations + +import asyncio +import contextlib +import contextvars +import sys +import threading +from collections.abc import Coroutine, Iterator +from dataclasses import dataclass +from typing import Any, TypeVar + +import pytest + +if sys.version_info >= (3, 11): + from asyncio import Runner +else: + from backports.asyncio.runner import Runner + + +class _RunnerState(threading.local): + invocation: _Delivery | None = None + + +@dataclass +class _Delivery: + loop: asyncio.AbstractEventLoop + exception: BaseException | None = None + handle: asyncio.Handle | None = None + timeout_cancellation: bool = False + + def interrupt(self, state: _RunnerState) -> None: + raise NotImplementedError + + +@dataclass +class _Invocation(_Delivery): + task: asyncio.Task[Any] | None = None + cancellation_requested: bool = False + + def interrupt(self, state: _RunnerState) -> None: + self.handle = None + if state.invocation is self and self.task is not None: + self.cancellation_requested = self.task.cancel() + + +@dataclass +class _Shutdown(_Delivery): + def interrupt(self, state: _RunnerState) -> None: + self.handle = None + if state.invocation is self: + # Runner.close() owns the loop and closes it in a finally block. + # Stop only that final shutdown, never a reusable runner invocation. + self.loop.stop() + + +_RUNNER_STATE = pytest.StashKey[_RunnerState]() +_T = TypeVar("_T") + + +def configure(config: pytest.Config) -> None: + enabled = config.getoption("asyncio_cooperative_timeouts") or config.getini( + "asyncio_cooperative_timeouts" + ) + if not enabled: + return + if not config.hook.pytest_timeout_expired.has_spec(): + raise pytest.UsageError( + "asyncio_cooperative_timeouts requires pytest-timeout's " + "pytest_timeout_expired hook" + ) + config.stash[_RUNNER_STATE] = _RunnerState() + + +@pytest.hookimpl(tryfirst=True, optionalhook=True) +def pytest_timeout_expired(item: pytest.Item, exception: BaseException) -> bool | None: + state = item.config.stash.get(_RUNNER_STATE, None) + if state is None or state.invocation is None: + return None + invocation = state.invocation + 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. + if not invocation.loop.is_closed(): + invocation.handle = invocation.loop.call_soon_threadsafe( + invocation.interrupt, state + ) + return True + + +@contextlib.contextmanager +def _deliver(config: pytest.Config, invocation: _Delivery) -> Iterator[None]: + state = config.stash[_RUNNER_STATE] + previous = state.invocation + state.invocation = invocation + try: + try: + yield + finally: + # Once the runner returns, a new signal can fail synchronously. + # Stop claiming it before deciding which outcome to propagate. + state.invocation = previous + if invocation.handle is not None: + invocation.handle.cancel() + except (KeyboardInterrupt, SystemExit, pytest.exit.Exception): + raise + except asyncio.CancelledError as exc: + if invocation.exception is None: + raise + if invocation.timeout_cancellation: + raise invocation.exception from exc + # Preserve cancellation when another caller requested it, or when a + # custom Task cannot tell us whose cancellation is being delivered. + raise exc from invocation.exception + except BaseException as exc: + if invocation.exception is None or exc is invocation.exception: + raise + raise invocation.exception from exc + if invocation.exception is not None: + raise invocation.exception + + +def run( + runner: Runner, + coro: Coroutine[Any, Any, _T], + *, + context: contextvars.Context, + config: pytest.Config, +) -> _T: + if _RUNNER_STATE not in config.stash: + return runner.run(coro, context=context) + + invocation = _Invocation(runner.get_loop()) + + async def invoke() -> _T: + task = asyncio.current_task() + assert task is not None + invocation.task = task + if invocation.exception is not None: + coro.close() + raise invocation.exception + # A custom Python 3.10 task factory can bypass the runner backport's + # Task, which provides the cancellation-count methods added in 3.11. + get_cancelling = getattr(task, "cancelling", None) + uncancel = getattr(task, "uncancel", None) + cancelling = get_cancelling() if get_cancelling is not None else 0 + try: + return await coro + finally: + if invocation.cancellation_requested and uncancel is not None: + # Remove only our cancellation before Runner handles SIGINT. + # A concurrent external cancellation must still propagate. + remaining = uncancel() + invocation.timeout_cancellation = ( + get_cancelling is not None and remaining <= cancelling + ) + + wrapped = invoke() + try: + with _deliver(config, invocation): + return runner.run(wrapped, context=context) + finally: + if invocation.task is None: + wrapped.close() + coro.close() + + +def close(runner: Runner, *, config: pytest.Config) -> None: + if _RUNNER_STATE not in config.stash: + runner.close() + return + with _deliver(config, _Shutdown(runner.get_loop())): + runner.close() diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 38b75e41..20c11d58 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -54,6 +54,13 @@ PytestPluginManager, ) +from ._timeout import ( + close as _close_with_timeout, + configure as _configure_timeouts, + pytest_timeout_expired as pytest_timeout_expired, + run as _run_with_timeout, +) + if sys.version_info >= (3, 11): from asyncio import Runner else: @@ -123,6 +130,11 @@ def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager) -> None default=None, help="enable asyncio debug mode for the default event loop", ) + group.addoption( + "--asyncio-cooperative-timeouts", + action="store_true", + help="deliver pytest-timeout signal failures by cancelling the active task", + ) parser.addini( "asyncio_mode", help="default value for --asyncio-mode", @@ -134,6 +146,12 @@ def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager) -> None type="bool", default="false", ) + parser.addini( + "asyncio_cooperative_timeouts", + help="default value for --asyncio-cooperative-timeouts", + type="bool", + default=False, + ) parser.addini( "asyncio_default_fixture_loop_scope", type="string", @@ -294,6 +312,7 @@ def _validate_scope(scope: str | None, option_name: str) -> None: def pytest_configure(config: Config) -> None: + _configure_timeouts(config) default_fixture_loop_scope = config.getini("asyncio_default_fixture_loop_scope") _validate_scope(default_fixture_loop_scope, "asyncio_default_fixture_loop_scope") if not default_fixture_loop_scope: @@ -405,7 +424,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 +443,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 +476,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 +588,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) @@ -894,6 +919,7 @@ 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 @@ -903,7 +929,7 @@ def _synchronize_coroutine( @functools.wraps(func) def inner(*args, **kwargs): coro = func(*args, **kwargs) - runner.run(coro, context=context) + _run_with_timeout(runner, coro, context=context, config=config) return inner @@ -1042,15 +1068,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..8116b441 --- /dev/null +++ b/tests/test_timeout.py @@ -0,0 +1,320 @@ +from __future__ import annotations + +import signal +import sys +from textwrap import dedent + +import pytest +from pytest import Pytester + +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.mark.parametrize("startup", ["timeout", "error"]) +def test_timeout_before_task_start( + pytester: Pytester, startup: str, timeout_plugin: None +): + pytester.makeini( + "[pytest]\nasyncio_default_fixture_loop_scope = function\n" + "asyncio_cooperative_timeouts = true" + ) + pytester.makeconftest(dedent("""\ + import pytest + + failures = [] + + @pytest.hookimpl(wrapper=True) + def pytest_timeout_expired(item, exception): + failures.append(exception) + return (yield) + """)) + pytester.makepyfile(dedent(f"""\ + import contextvars + import inspect + import signal + import pytest + from conftest import failures + from pytest_asyncio._timeout import Runner, run + + @pytest.mark.timeout(10, method="signal", func_only=True) + def test_startup(request): + startup = {startup!r} + entered = [] + wrappers = [] + failure = ValueError("task creation failed") + + async def body(): + entered.append(True) + + async def later(): + return 42 + + with Runner() as runner: + loop = runner.get_loop() + + def task_factory(loop, coro, **kwargs): + loop.set_task_factory(None) + wrappers.append(coro) + if startup == "error": + raise failure + signal.raise_signal(signal.SIGALRM) + return loop.create_task(coro, **kwargs) + + loop.set_task_factory(task_factory) + coro = body() + expected = ValueError if startup == "error" else pytest.fail.Exception + with pytest.raises(expected) as caught: + run( + runner, coro, context=contextvars.copy_context(), + config=request.config, + ) + assert caught.value is (failure if startup == "error" else failures[0]) + assert not entered + assert len(wrappers) == 1 and wrappers[0] is not coro + assert inspect.getcoroutinestate(coro) == inspect.CORO_CLOSED + assert inspect.getcoroutinestate(wrappers[0]) == inspect.CORO_CLOSED + assert run( + runner, later(), context=contextvars.copy_context(), + config=request.config, + ) == 42 + """)) + result = pytester.runpytest_subprocess(timeout=10) + result.assert_outcomes(passed=1) + assert "was never awaited" not in result.stdout.str() + result.stderr.str() + + +@pytest.mark.parametrize("trigger", ["timer", "reschedule"]) +def test_signal_timeout_preserves_shared_loop( + pytester: Pytester, trigger: str, timeout_plugin: None +): + pytester.makeini( + "[pytest]\nasyncio_default_fixture_loop_scope = function\n" + f"asyncio_cooperative_timeouts = {trigger == 'timer'}" + ) + pytester.makepyfile(dedent(f"""\ + import asyncio + import signal + import time + import pytest + + cleaned = [] + + async def application_wait(): + await asyncio.Future() + + @pytest.mark.timeout( + {0.1 if trigger == "timer" else 10}, method="signal", func_only=True + ) + @pytest.mark.asyncio(loop_scope="module") + async def test_timeout(): + 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!r} == "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(True) + + @pytest.mark.asyncio(loop_scope="module") + async def test_later(): + assert cleaned == [True] + + @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) + """)) + args = ["--tb=short"] + if trigger == "reschedule": + args.append("--asyncio-cooperative-timeouts") + result = pytester.runpytest_subprocess(*args, timeout=10) + result.assert_outcomes(failed=1, passed=2) + result.stdout.fnmatch_lines(["*Failed: Timeout*from pytest-timeout.*"]) + if trigger == "timer": + result.stdout.fnmatch_lines(["*in application_wait*", "*CancelledError*"]) + + +@pytest.mark.parametrize( + "cleanup", + [ + "return", + "xfail", + "error", + "interrupt", + "exit", + pytest.param( + "native_interrupt", + marks=pytest.mark.skipif( + sys.version_info >= (3, 11), + reason="native tasks have cancellation counters on Python 3.11+", + ), + ), + ], +) +def test_timeout_preserves_process_control( + pytester: Pytester, cleanup: str, timeout_plugin: None +): + pytester.makeini( + "[pytest]\nasyncio_default_fixture_loop_scope = function\n" + "asyncio_cooperative_timeouts = true" + ) + if cleanup == "native_interrupt": + pytester.makeconftest(dedent("""\ + import asyncio + import pytest + + NativeTask = asyncio.Task + + def loop_factory(): + loop = asyncio.new_event_loop() + loop.set_task_factory(lambda loop, coro: NativeTask(coro, loop=loop)) + return loop + + def pytest_asyncio_loop_factories(config, item): + return {"native": loop_factory} + + def pytest_runtest_makereport(item, call): + if call.when == "call": + error = call.excinfo.value + assert isinstance(error, asyncio.CancelledError) + assert isinstance(error.__cause__, pytest.fail.Exception) + assert "Timeout" in str(error.__cause__) + """)) + pytester.makepyfile(dedent(f"""\ + import asyncio + import signal + import pytest + + @pytest.mark.timeout(10, method="signal", func_only=True) + @pytest.mark.asyncio + async def test_timeout(): + loop = asyncio.get_running_loop() + loop.call_soon(signal.raise_signal, signal.SIGALRM) + try: + await asyncio.Future() + except asyncio.CancelledError: + if {cleanup!r} == "return": + return + if {cleanup!r} == "xfail": + pytest.xfail("cleanup xfail") + if {cleanup!r} == "error": + raise ValueError("cleanup error") + if {cleanup!r} == "exit": + pytest.exit("requested exit", returncode=4) + signal.raise_signal(signal.SIGINT) + await asyncio.sleep(0) + """)) + result = pytester.runpytest_subprocess(timeout=10) + if cleanup == "interrupt": + assert result.ret == pytest.ExitCode.INTERRUPTED + result.stdout.fnmatch_lines(["*KeyboardInterrupt*"]) + elif cleanup == "exit": + assert result.ret == pytest.ExitCode.USAGE_ERROR + result.stdout.fnmatch_lines(["*Exit: requested exit*"]) + elif cleanup == "native_interrupt": + result.assert_outcomes(failed=1) + result.stdout.fnmatch_lines(["*Failed: Timeout*", "*CancelledError*"]) + else: + result.assert_outcomes(failed=1) + result.stdout.fnmatch_lines(["*Failed: Timeout*from pytest-timeout.*"]) + + +@pytest.mark.parametrize( + "phase", ["coroutine_setup", "generator_setup", "teardown", "shutdown"] +) +def test_timeout_during_async_cleanup( + pytester: Pytester, phase: str, timeout_plugin: None +): + pytester.makeini( + "[pytest]\nasyncio_default_fixture_loop_scope = function\n" + "asyncio_cooperative_timeouts = true" + ) + pytester.makepyfile(dedent(f"""\ + import asyncio + import signal + import pytest + import pytest_asyncio + + 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(): + if {phase!r} == "coroutine_setup": + await timeout() + + @pytest_asyncio.fixture + async def fixture(): + if {phase!r} == "generator_setup": + await timeout() + yield + if {phase!r} == "teardown": + await timeout() + + @pytest.mark.timeout(10, method="signal") + @pytest.mark.asyncio + async def test_timeout(coroutine, fixture): + if {phase!r} == "shutdown": + asyncio.create_task(background()) + await asyncio.sleep(0) + + def test_later(): + pass + """)) + result = pytester.runpytest_subprocess(timeout=10) + result.assert_outcomes(errors=1, passed=1 if phase.endswith("setup") else 2) + result.stdout.fnmatch_lines(["*Failed: Timeout*from pytest-timeout.*"]) + + +def test_cooperative_timeout_is_opt_in(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) + + +def test_cooperative_timeout_requires_plugin(pytester: Pytester): + result = pytester.runpytest_subprocess( + "-p", "no:timeout", "--asyncio-cooperative-timeouts", timeout=10 + ) + assert result.ret == pytest.ExitCode.USAGE_ERROR + result.stderr.fnmatch_lines( + ["*asyncio_cooperative_timeouts requires pytest-timeout*"] + ) diff --git a/tox.ini b/tox.ini index 21ba3ead..2c013241 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, py310-lower-bounds, pytest-timeout-dev, docs, pyright isolated_build = true passenv = CI @@ -36,6 +36,16 @@ 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 +commands = + python -c "import pytest_timeout; assert hasattr(pytest_timeout.TimeoutHooks, 'pytest_timeout_expired')" + make test + [testenv:docs] allowlist_externals = git @@ -84,9 +94,9 @@ commands = pyright pytest_asyncio/ tests/ [gh-actions] python = - 3.10: py310, py310-lower-bounds, build + 3.10: py310, py310-lower-bounds, build, pytest-timeout-dev 3.11: py311 3.12: py312 3.13: py313, pyright - 3.14-dev: py314 + 3.14: py314, 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 = [ From b96aff4b33303d04daf41299d09d83c8dad4068e Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Sun, 23 Aug 2026 21:27:27 -0400 Subject: [PATCH 02/12] Simplify cooperative timeout integration Use asyncio.Timeout to own cancellation accounting on Python 3.11+ and activate the integration automatically when pytest-timeout provides its expiry hook [1]. Leave Python 3.10 on its existing signal path and remove the opt-in configuration and custom-task cancellation fallback. Document that cooperative cancellation cannot preempt Python code that never yields, while the thread method terminates the whole process. A stop at the boundary between Runner.close() phases can be consumed without interrupting the next phase [2]. Keep the stop callback queued while the final shutdown is active, and cancel its remaining handle on exit. This prevents later executor shutdown from outliving an already received expiry without introducing another timer. Hide internal delivery frames while preserving the interrupted await's traceback. Document the remaining coroutine wrapper rather than changing task creation to conceal it. [1]: https://docs.python.org/3/library/asyncio-task.html#asyncio.timeout [2]: https://github.com/pytest-dev/pytest-asyncio/pull/1548#discussion_r3840018306 --- changelog.d/215.added.rst | 2 +- docs/reference/configuration.rst | 7 -- docs/reference/timeouts.rst | 36 ++++--- pytest_asyncio/_timeout.py | 58 +++++------ pytest_asyncio/plugin.py | 11 --- tests/test_timeout.py | 165 +++++++++++++++++-------------- tox.ini | 2 +- 7 files changed, 135 insertions(+), 146 deletions(-) diff --git a/changelog.d/215.added.rst b/changelog.d/215.added.rst index c25324f7..d9ca5d2d 100644 --- a/changelog.d/215.added.rst +++ b/changelog.d/215.added.rst @@ -1 +1 @@ -Add opt-in cooperative delivery of pytest-timeout signal failures in asynchronous tests and fixtures. +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/configuration.rst b/docs/reference/configuration.rst index 7775c955..5341f02a 100644 --- a/docs/reference/configuration.rst +++ b/docs/reference/configuration.rst @@ -56,10 +56,3 @@ The value can also be set via the ``--asyncio-mode`` command-line option: If the asyncio mode is set in both the pytest configuration file and the command-line option, the command-line option takes precedence. If no asyncio mode is specified, the mode defaults to `strict`. - -asyncio_cooperative_timeouts -============================ - -Enable cooperative delivery of pytest-timeout signal failures. This is disabled -by default and can also be enabled with ``--asyncio-cooperative-timeouts``. -See :doc:`timeouts` for the required plugin support and cancellation limits. diff --git a/docs/reference/timeouts.rst b/docs/reference/timeouts.rst index d2d9cd18..32c00950 100644 --- a/docs/reference/timeouts.rst +++ b/docs/reference/timeouts.rst @@ -2,27 +2,25 @@ Timeouts ======== -Enable ``--asyncio-cooperative-timeouts`` or set -``asyncio_cooperative_timeouts = true`` to deliver pytest-timeout signal -failures by cancelling the active asynchronous test or fixture. This requires -a pytest-timeout version providing ``pytest_timeout_expired``. Cooperative -cleanup can then run before pytest reports the original timeout. +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. 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. It cannot -stop a callback that blocks indefinitely or a task that refuses cancellation. +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. A timeout during final event-loop shutdown -stops that shutdown and closes the loop; remaining resource cleanup may be -incomplete. +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. -Custom Python 3.10 task factories may return tasks without cancellation -counters. For those tasks, an escaping ``CancelledError`` is preserved and -the timeout is chained to it, because the runner cannot reliably distinguish -timeout cancellation from another cancellation request. - -Cooperative timeouts are disabled by default. Enabling them without a -compatible pytest-timeout plugin is a configuration error. The integration -does not take over runners managed by other async plugins or synchronous tests -that call ``asyncio.run()`` themselves. +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/pytest_asyncio/_timeout.py b/pytest_asyncio/_timeout.py index 3ab2fedd..cbfc509b 100644 --- a/pytest_asyncio/_timeout.py +++ b/pytest_asyncio/_timeout.py @@ -28,7 +28,6 @@ class _Delivery: loop: asyncio.AbstractEventLoop exception: BaseException | None = None handle: asyncio.Handle | None = None - timeout_cancellation: bool = False def interrupt(self, state: _RunnerState) -> None: raise NotImplementedError @@ -36,13 +35,13 @@ def interrupt(self, state: _RunnerState) -> None: @dataclass class _Invocation(_Delivery): - task: asyncio.Task[Any] | None = None - cancellation_requested: bool = False + timeout: asyncio.Timeout | None = None + started: bool = False def interrupt(self, state: _RunnerState) -> None: self.handle = None - if state.invocation is self and self.task is not None: - self.cancellation_requested = self.task.cancel() + if state.invocation is self and self.timeout is not None: + self.timeout.reschedule(self.loop.time()) @dataclass @@ -51,8 +50,10 @@ def interrupt(self, state: _RunnerState) -> None: self.handle = None if state.invocation is self: # Runner.close() owns the loop and closes it in a finally block. - # Stop only that final shutdown, never a reusable runner invocation. + # Each shutdown phase can consume a stop, so keep stopping until + # close() returns. Never stop a reusable runner invocation. self.loop.stop() + self.handle = self.loop.call_soon(self.interrupt, state) _RUNNER_STATE = pytest.StashKey[_RunnerState]() @@ -60,16 +61,10 @@ def interrupt(self, state: _RunnerState) -> None: def configure(config: pytest.Config) -> None: - enabled = config.getoption("asyncio_cooperative_timeouts") or config.getini( - "asyncio_cooperative_timeouts" - ) - if not enabled: + if sys.version_info < (3, 11): return if not config.hook.pytest_timeout_expired.has_spec(): - raise pytest.UsageError( - "asyncio_cooperative_timeouts requires pytest-timeout's " - "pytest_timeout_expired hook" - ) + return config.stash[_RUNNER_STATE] = _RunnerState() @@ -92,6 +87,7 @@ def pytest_timeout_expired(item: pytest.Item, exception: BaseException) -> bool @contextlib.contextmanager def _deliver(config: pytest.Config, invocation: _Delivery) -> Iterator[None]: + __tracebackhide__ = True state = config.stash[_RUNNER_STATE] previous = state.invocation state.invocation = invocation @@ -109,10 +105,8 @@ def _deliver(config: pytest.Config, invocation: _Delivery) -> Iterator[None]: except asyncio.CancelledError as exc: if invocation.exception is None: raise - if invocation.timeout_cancellation: - raise invocation.exception from exc - # Preserve cancellation when another caller requested it, or when a - # custom Task cannot tell us whose cancellation is being delivered. + # asyncio.Timeout converts only its own cancellation to TimeoutError. + # Preserve cancellation requested by another caller. raise exc from invocation.exception except BaseException as exc: if invocation.exception is None or exc is invocation.exception: @@ -129,45 +123,39 @@ def run( context: contextvars.Context, config: pytest.Config, ) -> _T: + __tracebackhide__ = True if _RUNNER_STATE not in config.stash: return runner.run(coro, context=context) invocation = _Invocation(runner.get_loop()) async def invoke() -> _T: - task = asyncio.current_task() - assert task is not None - invocation.task = task + __tracebackhide__ = True + invocation.started = True if invocation.exception is not None: coro.close() raise invocation.exception - # A custom Python 3.10 task factory can bypass the runner backport's - # Task, which provides the cancellation-count methods added in 3.11. - get_cancelling = getattr(task, "cancelling", None) - uncancel = getattr(task, "uncancel", None) - cancelling = get_cancelling() if get_cancelling is not None else 0 try: - return await coro + async with asyncio.timeout(None) as timeout: + invocation.timeout = timeout + return await coro finally: - if invocation.cancellation_requested and uncancel is not None: - # Remove only our cancellation before Runner handles SIGINT. - # A concurrent external cancellation must still propagate. - remaining = uncancel() - invocation.timeout_cancellation = ( - get_cancelling is not None and remaining <= cancelling - ) + # The signal may have queued delivery just as the coroutine exits. + # Do not reschedule a timeout whose context has already exited. + invocation.timeout = None wrapped = invoke() try: with _deliver(config, invocation): return runner.run(wrapped, context=context) finally: - if invocation.task is None: + if not invocation.started: wrapped.close() coro.close() def close(runner: Runner, *, config: pytest.Config) -> None: + __tracebackhide__ = True if _RUNNER_STATE not in config.stash: runner.close() return diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index 20c11d58..f0ceb429 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -130,11 +130,6 @@ def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager) -> None default=None, help="enable asyncio debug mode for the default event loop", ) - group.addoption( - "--asyncio-cooperative-timeouts", - action="store_true", - help="deliver pytest-timeout signal failures by cancelling the active task", - ) parser.addini( "asyncio_mode", help="default value for --asyncio-mode", @@ -146,12 +141,6 @@ def pytest_addoption(parser: Parser, pluginmanager: PytestPluginManager) -> None type="bool", default="false", ) - parser.addini( - "asyncio_cooperative_timeouts", - help="default value for --asyncio-cooperative-timeouts", - type="bool", - default=False, - ) parser.addini( "asyncio_default_fixture_loop_scope", type="string", diff --git a/tests/test_timeout.py b/tests/test_timeout.py index 8116b441..d9d10ea9 100644 --- a/tests/test_timeout.py +++ b/tests/test_timeout.py @@ -18,14 +18,17 @@ def timeout_plugin(request: pytest.FixtureRequest): 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("startup", ["timeout", "error"]) def test_timeout_before_task_start( - pytester: Pytester, startup: str, timeout_plugin: None + pytester: Pytester, startup: str, cooperative_timeout: None ): - pytester.makeini( - "[pytest]\nasyncio_default_fixture_loop_scope = function\n" - "asyncio_cooperative_timeouts = true" - ) + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makeconftest(dedent("""\ import pytest @@ -93,12 +96,9 @@ def task_factory(loop, coro, **kwargs): @pytest.mark.parametrize("trigger", ["timer", "reschedule"]) def test_signal_timeout_preserves_shared_loop( - pytester: Pytester, trigger: str, timeout_plugin: None + pytester: Pytester, trigger: str, cooperative_timeout: None ): - pytester.makeini( - "[pytest]\nasyncio_default_fixture_loop_scope = function\n" - f"asyncio_cooperative_timeouts = {trigger == 'timer'}" - ) + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makepyfile(dedent(f"""\ import asyncio import signal @@ -145,62 +145,44 @@ def test_synchronous(): with pytest.raises(pytest.fail.Exception, match="Timeout"): signal.raise_signal(signal.SIGALRM) """)) - args = ["--tb=short"] - if trigger == "reschedule": - args.append("--asyncio-cooperative-timeouts") - result = pytester.runpytest_subprocess(*args, timeout=10) + result = pytester.runpytest_subprocess("--tb=short", timeout=10) result.assert_outcomes(failed=1, passed=2) result.stdout.fnmatch_lines(["*Failed: Timeout*from pytest-timeout.*"]) if trigger == "timer": result.stdout.fnmatch_lines(["*in application_wait*", "*CancelledError*"]) + assert "_timeout.py" not in result.stdout.str() @pytest.mark.parametrize( "cleanup", - [ - "return", - "xfail", - "error", - "interrupt", - "exit", - pytest.param( - "native_interrupt", - marks=pytest.mark.skipif( - sys.version_info >= (3, 11), - reason="native tasks have cancellation counters on Python 3.11+", - ), - ), - ], + ["return", "xfail", "error", "interrupt", "exit", "system_exit", "cancel"], ) def test_timeout_preserves_process_control( - pytester: Pytester, cleanup: str, timeout_plugin: None + pytester: Pytester, cleanup: str, cooperative_timeout: None ): - pytester.makeini( - "[pytest]\nasyncio_default_fixture_loop_scope = function\n" - "asyncio_cooperative_timeouts = true" - ) - if cleanup == "native_interrupt": - pytester.makeconftest(dedent("""\ - import asyncio - import pytest - - NativeTask = asyncio.Task - - def loop_factory(): - loop = asyncio.new_event_loop() - loop.set_task_factory(lambda loop, coro: NativeTask(coro, loop=loop)) - return loop - - def pytest_asyncio_loop_factories(config, item): - return {"native": loop_factory} - - def pytest_runtest_makereport(item, call): - if call.when == "call": - error = call.excinfo.value + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makeconftest(dedent(f"""\ + import asyncio + import pytest + + failures = [] + + @pytest.hookimpl(wrapper=True) + def pytest_timeout_expired(item, exception): + failures.append(exception) + return (yield) + + def pytest_runtest_makereport(item, call): + if call.when == "call": + error = call.excinfo.value + if {cleanup!r} == "cancel": assert isinstance(error, asyncio.CancelledError) - assert isinstance(error.__cause__, pytest.fail.Exception) - assert "Timeout" in str(error.__cause__) - """)) + assert error.__cause__ is failures[0] + elif {cleanup!r} == "system_exit": + assert isinstance(error, SystemExit) and error.code == 7 + else: + assert error is failures[0] + """)) pytester.makepyfile(dedent(f"""\ import asyncio import signal @@ -222,7 +204,12 @@ async def test_timeout(): raise ValueError("cleanup error") if {cleanup!r} == "exit": pytest.exit("requested exit", returncode=4) - signal.raise_signal(signal.SIGINT) + if {cleanup!r} == "system_exit": + raise SystemExit(7) + if {cleanup!r} == "cancel": + asyncio.current_task().cancel() + else: + signal.raise_signal(signal.SIGINT) await asyncio.sleep(0) """)) result = pytester.runpytest_subprocess(timeout=10) @@ -232,7 +219,10 @@ async def test_timeout(): elif cleanup == "exit": assert result.ret == pytest.ExitCode.USAGE_ERROR result.stdout.fnmatch_lines(["*Exit: requested exit*"]) - elif cleanup == "native_interrupt": + elif cleanup == "system_exit": + result.assert_outcomes(failed=1) + result.stdout.fnmatch_lines(["*SystemExit: 7*"]) + elif cleanup == "cancel": result.assert_outcomes(failed=1) result.stdout.fnmatch_lines(["*Failed: Timeout*", "*CancelledError*"]) else: @@ -241,21 +231,29 @@ async def test_timeout(): @pytest.mark.parametrize( - "phase", ["coroutine_setup", "generator_setup", "teardown", "shutdown"] + "phase", + ["coroutine_setup", "generator_setup", "teardown", "shutdown", "shutdown_boundary"], ) def test_timeout_during_async_cleanup( - pytester: Pytester, phase: str, timeout_plugin: None + pytester: Pytester, phase: str, cooperative_timeout: None ): - pytester.makeini( - "[pytest]\nasyncio_default_fixture_loop_scope = function\n" - "asyncio_cooperative_timeouts = true" - ) + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makepyfile(dedent(f"""\ import asyncio import signal + import threading + from concurrent.futures import ThreadPoolExecutor import pytest import pytest_asyncio + executor = None + release = threading.Event() + finished = threading.Event() + + def worker(): + release.wait(5) + finished.set() + async def timeout(): asyncio.get_running_loop().call_soon(signal.raise_signal, signal.SIGALRM) await asyncio.Future() @@ -282,19 +280,38 @@ async def fixture(): @pytest.mark.timeout(10, method="signal") @pytest.mark.asyncio async def test_timeout(coroutine, fixture): + global executor if {phase!r} == "shutdown": asyncio.create_task(background()) await asyncio.sleep(0) + if {phase!r} == "shutdown_boundary": + loop = asyncio.get_running_loop() + executor = ThreadPoolExecutor() + loop.set_default_executor(executor) + loop.run_in_executor(None, worker) + original = loop.shutdown_asyncgens + + async def shutdown_asyncgens(): + await original() + signal.raise_signal(signal.SIGALRM) + + loop.shutdown_asyncgens = shutdown_asyncgens def test_later(): - pass + if executor is not None: + try: + assert not finished.is_set() + finally: + release.set() + executor.shutdown(wait=True) """)) result = pytester.runpytest_subprocess(timeout=10) result.assert_outcomes(errors=1, passed=1 if phase.endswith("setup") else 2) result.stdout.fnmatch_lines(["*Failed: Timeout*from pytest-timeout.*"]) -def test_cooperative_timeout_is_opt_in(pytester: Pytester, timeout_plugin: None): +@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 @@ -310,11 +327,15 @@ async def test_signal(): result.assert_outcomes(passed=1) -def test_cooperative_timeout_requires_plugin(pytester: Pytester): - result = pytester.runpytest_subprocess( - "-p", "no:timeout", "--asyncio-cooperative-timeouts", timeout=10 - ) - assert result.ret == pytest.ExitCode.USAGE_ERROR - result.stderr.fnmatch_lines( - ["*asyncio_cooperative_timeouts requires pytest-timeout*"] - ) +def test_asyncio_without_timeout_plugin(pytester: Pytester): + pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") + pytester.makepyfile(dedent("""\ + import asyncio + import pytest + + @pytest.mark.asyncio + async def test_asyncio(): + await asyncio.sleep(0) + """)) + result = pytester.runpytest_subprocess("-p", "no:timeout", timeout=10) + result.assert_outcomes(passed=1) diff --git a/tox.ini b/tox.ini index 2c013241..f223da49 100644 --- a/tox.ini +++ b/tox.ini @@ -95,7 +95,7 @@ commands = pyright pytest_asyncio/ tests/ [gh-actions] python = 3.10: py310, py310-lower-bounds, build, pytest-timeout-dev - 3.11: py311 + 3.11: py311, pytest-timeout-dev 3.12: py312 3.13: py313, pyright 3.14: py314, pytest-timeout-dev From 4a319fd1ac4bb6a97c63642187da3404e374c1de Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Sun, 23 Aug 2026 21:56:47 -0400 Subject: [PATCH 03/12] Remove redundant timeout bookkeeping Cancel delivery handles at their owning boundary instead of clearing them as callbacks run. Keep the active-invocation checks: a signal can interrupt scheduling after a callback is queued but before its handle is saved. Use asyncio.Runner in the helper's postponed annotations and import it directly in the Python 3.11+ startup regression. Remove the fieldless dataclass redeclaration and the subprocess smoke test already covered by test_asyncio_marker in the ordinary plugin-free environments. --- pytest_asyncio/_timeout.py | 14 ++++---------- tests/test_timeout.py | 17 ++--------------- 2 files changed, 6 insertions(+), 25 deletions(-) diff --git a/pytest_asyncio/_timeout.py b/pytest_asyncio/_timeout.py index cbfc509b..a3dc6f65 100644 --- a/pytest_asyncio/_timeout.py +++ b/pytest_asyncio/_timeout.py @@ -13,11 +13,6 @@ import pytest -if sys.version_info >= (3, 11): - from asyncio import Runner -else: - from backports.asyncio.runner import Runner - class _RunnerState(threading.local): invocation: _Delivery | None = None @@ -39,15 +34,12 @@ class _Invocation(_Delivery): started: bool = False def interrupt(self, state: _RunnerState) -> None: - self.handle = None if state.invocation is self and self.timeout is not None: self.timeout.reschedule(self.loop.time()) -@dataclass class _Shutdown(_Delivery): def interrupt(self, state: _RunnerState) -> None: - self.handle = None if state.invocation is self: # Runner.close() owns the loop and closes it in a finally block. # Each shutdown phase can consume a stop, so keep stopping until @@ -78,6 +70,8 @@ def pytest_timeout_expired(item: pytest.Item, exception: BaseException) -> bool 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. + # A signal can interrupt before the returned handle is saved, so the + # callback must also check that this invocation is still active. if not invocation.loop.is_closed(): invocation.handle = invocation.loop.call_soon_threadsafe( invocation.interrupt, state @@ -117,7 +111,7 @@ def _deliver(config: pytest.Config, invocation: _Delivery) -> Iterator[None]: def run( - runner: Runner, + runner: asyncio.Runner, coro: Coroutine[Any, Any, _T], *, context: contextvars.Context, @@ -154,7 +148,7 @@ async def invoke() -> _T: coro.close() -def close(runner: Runner, *, config: pytest.Config) -> None: +def close(runner: asyncio.Runner, *, config: pytest.Config) -> None: __tracebackhide__ = True if _RUNNER_STATE not in config.stash: runner.close() diff --git a/tests/test_timeout.py b/tests/test_timeout.py index d9d10ea9..4920b1b4 100644 --- a/tests/test_timeout.py +++ b/tests/test_timeout.py @@ -43,9 +43,10 @@ def pytest_timeout_expired(item, exception): import contextvars import inspect import signal + from asyncio import Runner import pytest from conftest import failures - from pytest_asyncio._timeout import Runner, run + from pytest_asyncio._timeout import run @pytest.mark.timeout(10, method="signal", func_only=True) def test_startup(request): @@ -325,17 +326,3 @@ async def test_signal(): """)) result = pytester.runpytest_subprocess(timeout=10) result.assert_outcomes(passed=1) - - -def test_asyncio_without_timeout_plugin(pytester: Pytester): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile(dedent("""\ - import asyncio - import pytest - - @pytest.mark.asyncio - async def test_asyncio(): - await asyncio.sleep(0) - """)) - result = pytester.runpytest_subprocess("-p", "no:timeout", timeout=10) - result.assert_outcomes(passed=1) From 2b9fc334ae29eafed9fafa48d985937620999113 Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Sun, 23 Aug 2026 22:27:06 -0400 Subject: [PATCH 04/12] Simplify timeout delivery and test isolation Collapse delivery state into one record with an explicit shutdown mode. Use the active invocation check to ignore stale callbacks and remove saved-handle cancellation. The check also covers SIGINT interrupting scheduling before a handle is returned. Queued callbacks retain their delivery until processed or the loop closes. Batch startup and outcome checks around the real Runner boundary, and group the public shared-loop and fixture cases in separate child sessions. This preserves the failure cases and subprocess watchdogs without launching an interpreter for each parameter. --- pytest_asyncio/_timeout.py | 42 +++---- tests/test_timeout.py | 224 +++++++++++++++++-------------------- 2 files changed, 115 insertions(+), 151 deletions(-) diff --git a/pytest_asyncio/_timeout.py b/pytest_asyncio/_timeout.py index a3dc6f65..777bcad6 100644 --- a/pytest_asyncio/_timeout.py +++ b/pytest_asyncio/_timeout.py @@ -21,31 +21,21 @@ class _RunnerState(threading.local): @dataclass class _Delivery: loop: asyncio.AbstractEventLoop + closing: bool = False exception: BaseException | None = None - handle: asyncio.Handle | None = None - - def interrupt(self, state: _RunnerState) -> None: - raise NotImplementedError - - -@dataclass -class _Invocation(_Delivery): timeout: asyncio.Timeout | None = None started: bool = False def interrupt(self, state: _RunnerState) -> None: - if state.invocation is self and self.timeout is not None: - self.timeout.reschedule(self.loop.time()) - - -class _Shutdown(_Delivery): - def interrupt(self, state: _RunnerState) -> None: - if state.invocation is self: - # Runner.close() owns the loop and closes it in a finally block. - # Each shutdown phase can consume a stop, so keep stopping until - # close() returns. Never stop a reusable runner invocation. + if state.invocation 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.handle = self.loop.call_soon(self.interrupt, state) + self.loop.call_soon(self.interrupt, state) + elif self.timeout is not None: + self.timeout.reschedule(self.loop.time()) _RUNNER_STATE = pytest.StashKey[_RunnerState]() @@ -70,12 +60,10 @@ def pytest_timeout_expired(item: pytest.Item, exception: BaseException) -> bool 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. - # A signal can interrupt before the returned handle is saved, so the - # callback must also check that this invocation is still active. + # 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.handle = invocation.loop.call_soon_threadsafe( - invocation.interrupt, state - ) + invocation.loop.call_soon_threadsafe(invocation.interrupt, state) return True @@ -92,8 +80,6 @@ def _deliver(config: pytest.Config, invocation: _Delivery) -> Iterator[None]: # Once the runner returns, a new signal can fail synchronously. # Stop claiming it before deciding which outcome to propagate. state.invocation = previous - if invocation.handle is not None: - invocation.handle.cancel() except (KeyboardInterrupt, SystemExit, pytest.exit.Exception): raise except asyncio.CancelledError as exc: @@ -121,7 +107,7 @@ def run( if _RUNNER_STATE not in config.stash: return runner.run(coro, context=context) - invocation = _Invocation(runner.get_loop()) + invocation = _Delivery(runner.get_loop()) async def invoke() -> _T: __tracebackhide__ = True @@ -153,5 +139,5 @@ def close(runner: asyncio.Runner, *, config: pytest.Config) -> None: if _RUNNER_STATE not in config.stash: runner.close() return - with _deliver(config, _Shutdown(runner.get_loop())): + with _deliver(config, _Delivery(runner.get_loop(), closing=True)): runner.close() diff --git a/tests/test_timeout.py b/tests/test_timeout.py index 4920b1b4..34c917eb 100644 --- a/tests/test_timeout.py +++ b/tests/test_timeout.py @@ -24,10 +24,7 @@ def cooperative_timeout(timeout_plugin: None): pytest.skip("cooperative timeouts require Python 3.11 or newer") -@pytest.mark.parametrize("startup", ["timeout", "error"]) -def test_timeout_before_task_start( - pytester: Pytester, startup: str, cooperative_timeout: None -): +def test_runner_timeout_delivery(pytester: Pytester, cooperative_timeout: None): pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makeconftest(dedent("""\ import pytest @@ -39,18 +36,19 @@ def pytest_timeout_expired(item, exception): failures.append(exception) return (yield) """)) - pytester.makepyfile(dedent(f"""\ + pytester.makepyfile(dedent("""\ + import asyncio import contextvars import inspect import signal - from asyncio import Runner import pytest from conftest import failures from pytest_asyncio._timeout import run - @pytest.mark.timeout(10, method="signal", func_only=True) - def test_startup(request): - startup = {startup!r} + pytestmark = pytest.mark.timeout(10, method="signal", func_only=True) + + @pytest.mark.parametrize("startup", ["timeout", "error"]) + def test_startup(startup, request): entered = [] wrappers = [] failure = ValueError("task creation failed") @@ -61,7 +59,7 @@ async def body(): async def later(): return 42 - with Runner() as runner: + with asyncio.Runner() as runner: loop = runner.get_loop() def task_factory(loop, coro, **kwargs): @@ -80,27 +78,76 @@ def task_factory(loop, coro, **kwargs): runner, coro, context=contextvars.copy_context(), config=request.config, ) - assert caught.value is (failure if startup == "error" else failures[0]) + assert caught.value is (failure if startup == "error" else failures[-1]) assert not entered - assert len(wrappers) == 1 and wrappers[0] is not coro - assert inspect.getcoroutinestate(coro) == inspect.CORO_CLOSED - assert inspect.getcoroutinestate(wrappers[0]) == inspect.CORO_CLOSED + assert all( + inspect.getcoroutinestate(c) == inspect.CORO_CLOSED + for c in [coro, *wrappers] + ) assert run( runner, later(), context=contextvars.copy_context(), config=request.config, ) == 42 + + @pytest.mark.parametrize( + "cleanup", + ["return", "xfail", "error", "interrupt", "exit", "system_exit", "cancel"], + ) + def test_cleanup(cleanup, request): + async def body(): + loop = asyncio.get_running_loop() + loop.call_soon(signal.raise_signal, signal.SIGALRM) + try: + await asyncio.Future() + except asyncio.CancelledError: + if cleanup == "return": + return + if cleanup == "xfail": + pytest.xfail("cleanup xfail") + if cleanup == "error": + raise ValueError("cleanup error") + if cleanup == "exit": + pytest.exit("requested exit", returncode=4) + if cleanup == "system_exit": + raise SystemExit(7) + if cleanup == "cancel": + asyncio.current_task().cancel() + else: + signal.raise_signal(signal.SIGINT) + await asyncio.sleep(0) + + expected = { + "interrupt": KeyboardInterrupt, + "exit": pytest.exit.Exception, + "system_exit": SystemExit, + "cancel": asyncio.CancelledError, + }.get(cleanup, pytest.fail.Exception) + with asyncio.Runner() as runner: + with pytest.raises(expected) as caught: + run( + runner, body(), context=contextvars.copy_context(), + config=request.config, + ) + if cleanup == "cancel": + assert caught.value.__cause__ is failures[-1] + elif cleanup == "system_exit": + assert caught.value.code == 7 + elif cleanup == "exit": + assert caught.value.returncode == 4 + assert str(caught.value) == "requested exit" + elif cleanup != "interrupt": + assert caught.value is failures[-1] """)) result = pytester.runpytest_subprocess(timeout=10) - result.assert_outcomes(passed=1) + result.assert_outcomes(passed=9) assert "was never awaited" not in result.stdout.str() + result.stderr.str() -@pytest.mark.parametrize("trigger", ["timer", "reschedule"]) def test_signal_timeout_preserves_shared_loop( - pytester: Pytester, trigger: str, cooperative_timeout: None + pytester: Pytester, cooperative_timeout: None ): pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile(dedent(f"""\ + pytester.makepyfile(dedent("""\ import asyncio import signal import time @@ -111,11 +158,19 @@ def test_signal_timeout_preserves_shared_loop( async def application_wait(): await asyncio.Future() - @pytest.mark.timeout( - {0.1 if trigger == "timer" else 10}, method="signal", func_only=True + @pytest.mark.parametrize( + "trigger", + [ + pytest.param("timer", marks=pytest.mark.timeout( + 0.1, method="signal", func_only=True, + )), + pytest.param("reschedule", marks=pytest.mark.timeout( + 10, method="signal", func_only=True, + )), + ], ) @pytest.mark.asyncio(loop_scope="module") - async def test_timeout(): + async def test_timeout(trigger): loop = asyncio.get_running_loop() original = loop.call_soon task = asyncio.current_task() @@ -126,7 +181,7 @@ def reschedule(callback, *args, context=None): signal.raise_signal(signal.SIGALRM) return original(callback, *args, context=context) - if {trigger!r} == "reschedule": + if trigger == "reschedule": loop.call_soon = reschedule else: loop.call_soon(time.sleep, 0.2) @@ -135,11 +190,11 @@ def reschedule(callback, *args, context=None): await application_wait() finally: loop.call_soon = original - cleaned.append(True) + cleaned.append(trigger) @pytest.mark.asyncio(loop_scope="module") async def test_later(): - assert cleaned == [True] + assert cleaned == ["timer", "reschedule"] @pytest.mark.timeout(10, method="signal", func_only=True) def test_synchronous(): @@ -147,99 +202,15 @@ def test_synchronous(): signal.raise_signal(signal.SIGALRM) """)) result = pytester.runpytest_subprocess("--tb=short", timeout=10) - result.assert_outcomes(failed=1, passed=2) - result.stdout.fnmatch_lines(["*Failed: Timeout*from pytest-timeout.*"]) - if trigger == "timer": - result.stdout.fnmatch_lines(["*in application_wait*", "*CancelledError*"]) + 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() -@pytest.mark.parametrize( - "cleanup", - ["return", "xfail", "error", "interrupt", "exit", "system_exit", "cancel"], -) -def test_timeout_preserves_process_control( - pytester: Pytester, cleanup: str, cooperative_timeout: None -): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makeconftest(dedent(f"""\ - import asyncio - import pytest - - failures = [] - - @pytest.hookimpl(wrapper=True) - def pytest_timeout_expired(item, exception): - failures.append(exception) - return (yield) - - def pytest_runtest_makereport(item, call): - if call.when == "call": - error = call.excinfo.value - if {cleanup!r} == "cancel": - assert isinstance(error, asyncio.CancelledError) - assert error.__cause__ is failures[0] - elif {cleanup!r} == "system_exit": - assert isinstance(error, SystemExit) and error.code == 7 - else: - assert error is failures[0] - """)) - pytester.makepyfile(dedent(f"""\ - import asyncio - import signal - import pytest - - @pytest.mark.timeout(10, method="signal", func_only=True) - @pytest.mark.asyncio - async def test_timeout(): - loop = asyncio.get_running_loop() - loop.call_soon(signal.raise_signal, signal.SIGALRM) - try: - await asyncio.Future() - except asyncio.CancelledError: - if {cleanup!r} == "return": - return - if {cleanup!r} == "xfail": - pytest.xfail("cleanup xfail") - if {cleanup!r} == "error": - raise ValueError("cleanup error") - if {cleanup!r} == "exit": - pytest.exit("requested exit", returncode=4) - if {cleanup!r} == "system_exit": - raise SystemExit(7) - if {cleanup!r} == "cancel": - asyncio.current_task().cancel() - else: - signal.raise_signal(signal.SIGINT) - await asyncio.sleep(0) - """)) - result = pytester.runpytest_subprocess(timeout=10) - if cleanup == "interrupt": - assert result.ret == pytest.ExitCode.INTERRUPTED - result.stdout.fnmatch_lines(["*KeyboardInterrupt*"]) - elif cleanup == "exit": - assert result.ret == pytest.ExitCode.USAGE_ERROR - result.stdout.fnmatch_lines(["*Exit: requested exit*"]) - elif cleanup == "system_exit": - result.assert_outcomes(failed=1) - result.stdout.fnmatch_lines(["*SystemExit: 7*"]) - elif cleanup == "cancel": - result.assert_outcomes(failed=1) - result.stdout.fnmatch_lines(["*Failed: Timeout*", "*CancelledError*"]) - else: - result.assert_outcomes(failed=1) - result.stdout.fnmatch_lines(["*Failed: Timeout*from pytest-timeout.*"]) - - -@pytest.mark.parametrize( - "phase", - ["coroutine_setup", "generator_setup", "teardown", "shutdown", "shutdown_boundary"], -) -def test_timeout_during_async_cleanup( - pytester: Pytester, phase: str, cooperative_timeout: None -): +def test_timeout_during_async_cleanup(pytester: Pytester, cooperative_timeout: None): pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile(dedent(f"""\ + pytester.makepyfile(dedent("""\ import asyncio import signal import threading @@ -266,26 +237,33 @@ async def background(): await timeout() @pytest_asyncio.fixture - async def coroutine(): - if {phase!r} == "coroutine_setup": + async def coroutine(phase): + if phase == "coroutine_setup": await timeout() @pytest_asyncio.fixture - async def fixture(): - if {phase!r} == "generator_setup": + async def fixture(phase): + if phase == "generator_setup": await timeout() yield - if {phase!r} == "teardown": + 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(coroutine, fixture): + async def test_timeout(phase, coroutine, fixture): global executor - if {phase!r} == "shutdown": + if phase == "shutdown": asyncio.create_task(background()) await asyncio.sleep(0) - if {phase!r} == "shutdown_boundary": + if phase == "shutdown_boundary": loop = asyncio.get_running_loop() executor = ThreadPoolExecutor() loop.set_default_executor(executor) @@ -307,8 +285,8 @@ def test_later(): executor.shutdown(wait=True) """)) result = pytester.runpytest_subprocess(timeout=10) - result.assert_outcomes(errors=1, passed=1 if phase.endswith("setup") else 2) - result.stdout.fnmatch_lines(["*Failed: Timeout*from pytest-timeout.*"]) + 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") From fab62f532517c653be01de01f99ae5ecdcf34075 Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Sun, 23 Aug 2026 23:12:43 -0400 Subject: [PATCH 05/12] Preserve coroutine ownership across interrupts The startup cleanup in 58aed6442 confuses first execution with Task ownership. An interrupt after Task creation can therefore close a wrapper that the Task still owns. Interruption inside timeout setup can instead leave the original coroutine unawaited. Use the actual pending Task to decide when cleanup is safe, and close only native coroutines that have never started. Closing a suspended coroutine can execute user cleanup synchronously and mask KeyboardInterrupt. Keep custom Coroutine objects on the plain Runner path. Their interface exposes no execution state, so wrapping them cannot preserve safe cleanup during interrupted startup. --- docs/reference/timeouts.rst | 4 +- pytest_asyncio/_timeout.py | 27 ++++++--- tests/test_timeout.py | 118 ++++++++++++++++++++++++++++++------ 3 files changed, 120 insertions(+), 29 deletions(-) diff --git a/docs/reference/timeouts.rst b/docs/reference/timeouts.rst index 32c00950..2de95d21 100644 --- a/docs/reference/timeouts.rst +++ b/docs/reference/timeouts.rst @@ -7,7 +7,9 @@ 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. +signal behavior. Runner invocations with custom coroutine objects retain +synchronous signal delivery; cooperative execution requires native coroutine +objects created by ``async def`` functions. pytest-timeout still controls the configured duration, covered test phases, debugger detection, and timeout diagnostics. diff --git a/pytest_asyncio/_timeout.py b/pytest_asyncio/_timeout.py index 777bcad6..5f2cf135 100644 --- a/pytest_asyncio/_timeout.py +++ b/pytest_asyncio/_timeout.py @@ -5,10 +5,12 @@ import asyncio import contextlib import contextvars +import inspect import sys import threading from collections.abc import Coroutine, Iterator from dataclasses import dataclass +from types import CoroutineType from typing import Any, TypeVar import pytest @@ -24,7 +26,6 @@ class _Delivery: closing: bool = False exception: BaseException | None = None timeout: asyncio.Timeout | None = None - started: bool = False def interrupt(self, state: _RunnerState) -> None: if state.invocation is not self: @@ -104,16 +105,14 @@ def run( config: pytest.Config, ) -> _T: __tracebackhide__ = True - if _RUNNER_STATE not in config.stash: + if _RUNNER_STATE not in config.stash or not isinstance(coro, CoroutineType): return runner.run(coro, context=context) invocation = _Delivery(runner.get_loop()) async def invoke() -> _T: __tracebackhide__ = True - invocation.started = True if invocation.exception is not None: - coro.close() raise invocation.exception try: async with asyncio.timeout(None) as timeout: @@ -128,10 +127,22 @@ async def invoke() -> _T: try: with _deliver(config, invocation): return runner.run(wrapped, context=context) - finally: - if not invocation.started: - wrapped.close() - coro.close() + except BaseException: + + def close_unstarted(_task: asyncio.Future[Any] | None = None) -> None: + # Never run suspended user cleanup outside its task. + for coroutine in (wrapped, coro): + if inspect.getcoroutinestate(coroutine) == inspect.CORO_CREATED: + coroutine.close() + + # A task can own the wrapper without having started it. + for task in asyncio.all_tasks(invocation.loop): + if task.get_coro() is wrapped: + task.add_done_callback(close_unstarted) + break + else: + close_unstarted() + raise def close(runner: asyncio.Runner, *, config: pytest.Config) -> None: diff --git a/tests/test_timeout.py b/tests/test_timeout.py index 34c917eb..3be64b84 100644 --- a/tests/test_timeout.py +++ b/tests/test_timeout.py @@ -41,20 +41,31 @@ def pytest_timeout_expired(item, exception): import contextvars import inspect import signal + import sys + from collections.abc import Coroutine import pytest from conftest import failures from pytest_asyncio._timeout import run pytestmark = pytest.mark.timeout(10, method="signal", func_only=True) - @pytest.mark.parametrize("startup", ["timeout", "error"]) - def test_startup(startup, request): + @pytest.mark.parametrize("startup", [ + "timeout", "error", "interrupt_before_start", "interrupt_during_timeout", + "interrupt_after_start", "custom_coroutine", + ]) + def test_startup(startup, request, monkeypatch): entered = [] wrappers = [] + tasks = [] failure = ValueError("task creation failed") async def body(): entered.append(True) + if startup == "interrupt_after_start": + try: + await asyncio.Future() + finally: + raise failure async def later(): return 42 @@ -67,27 +78,94 @@ def task_factory(loop, coro, **kwargs): wrappers.append(coro) if startup == "error": raise failure - signal.raise_signal(signal.SIGALRM) - return loop.create_task(coro, **kwargs) + if startup == "timeout": + signal.raise_signal(signal.SIGALRM) + task = loop.create_task(coro, **kwargs) + tasks.append(task) + if startup == "custom_coroutine": + raise failure + return task + + if startup in ("interrupt_before_start", "interrupt_during_timeout"): + target, name = ( + (loop, "run_until_complete") + if startup == "interrupt_before_start" + else (asyncio, "timeout") + ) + original = getattr(target, name) + + def interrupt(*args, **kwargs): + monkeypatch.setattr(target, name, original) + signal.raise_signal(signal.SIGINT) + signal.raise_signal(signal.SIGINT) + return original(*args, **kwargs) + + monkeypatch.setattr(target, name, interrupt) loop.set_task_factory(task_factory) coro = body() - expected = ValueError if startup == "error" else pytest.fail.Exception - with pytest.raises(expected) as caught: - run( - runner, coro, context=contextvars.copy_context(), - config=request.config, + argument = coro + if startup == "custom_coroutine": + class CustomCoroutine(Coroutine): + def send(self, value): + return coro.send(value) + + def throw(self, *args): + return coro.throw(*args) + + def __await__(self): + return coro.__await__() + + argument = CustomCoroutine() + + original_trace = sys.gettrace() + if startup == "interrupt_after_start": + def trace(frame, event, arg): + if ( + wrappers and frame.f_code is wrappers[0].cr_code + and event == "return" and arg is not None + ): + sys.settrace(None) + signal.raise_signal(signal.SIGINT) + signal.raise_signal(signal.SIGINT) + return trace + + sys.settrace(trace) + + expected = { + "timeout": pytest.fail.Exception, "error": ValueError, + "custom_coroutine": ValueError, + }.get(startup, KeyboardInterrupt) + try: + with pytest.raises(expected) as caught: + run( + runner, argument, context=contextvars.copy_context(), + config=request.config, + ) + finally: + sys.settrace(original_trace) + if startup in ("timeout", "error", "custom_coroutine"): + assert caught.value is ( + failures[-1] if startup == "timeout" else failure ) - assert caught.value is (failure if startup == "error" else failures[-1]) - assert not entered - assert all( - inspect.getcoroutinestate(c) == inspect.CORO_CLOSED - for c in [coro, *wrappers] - ) - assert run( - runner, later(), context=contextvars.copy_context(), - config=request.config, - ) == 42 + if startup == "interrupt_after_start": + with pytest.raises(ValueError) as cleanup: + coro.close() + assert cleanup.value is failure + if startup != "custom_coroutine": + assert run( + runner, later(), context=contextvars.copy_context(), + config=request.config, + ) == 42 + assert bool(entered) is (startup == "interrupt_after_start") + assert all( + inspect.getcoroutinestate(c) == inspect.CORO_CLOSED + for c in [coro, *wrappers] if inspect.iscoroutine(c) + ) + if startup in ("interrupt_before_start", "custom_coroutine"): + assert tasks[0].cancelled() + elif startup in ("interrupt_during_timeout", "interrupt_after_start"): + assert tasks[0].exception() is caught.value @pytest.mark.parametrize( "cleanup", @@ -139,7 +217,7 @@ async def body(): assert caught.value is failures[-1] """)) result = pytester.runpytest_subprocess(timeout=10) - result.assert_outcomes(passed=9) + result.assert_outcomes(passed=13) assert "was never awaited" not in result.stdout.str() + result.stderr.str() From 7c8ea750112a0553543d5493f61e53101c3ef2e5 Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Sun, 23 Aug 2026 23:27:32 -0400 Subject: [PATCH 06/12] Discover timeout support when runners start Test modules can load pytest-timeout through pytest_plugins during collection, after pytest-asyncio has already configured itself. The configure-time check in 58aed6442 then leaves managed runners on the synchronous fallback even though the expiry hook is available. Check hook support when entering a runner and initialize delivery state only when needed. This also removes registration-order-dependent configuration plumbing. --- pytest_asyncio/_timeout.py | 17 ++++++++--------- pytest_asyncio/plugin.py | 2 -- tests/test_timeout.py | 9 +++++++-- 3 files changed, 15 insertions(+), 13 deletions(-) diff --git a/pytest_asyncio/_timeout.py b/pytest_asyncio/_timeout.py index 5f2cf135..e06a1f6b 100644 --- a/pytest_asyncio/_timeout.py +++ b/pytest_asyncio/_timeout.py @@ -43,12 +43,9 @@ def interrupt(self, state: _RunnerState) -> None: _T = TypeVar("_T") -def configure(config: pytest.Config) -> None: - if sys.version_info < (3, 11): - return - if not config.hook.pytest_timeout_expired.has_spec(): - return - config.stash[_RUNNER_STATE] = _RunnerState() +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 config.hook.pytest_timeout_expired.has_spec() @pytest.hookimpl(tryfirst=True, optionalhook=True) @@ -71,7 +68,7 @@ def pytest_timeout_expired(item: pytest.Item, exception: BaseException) -> bool @contextlib.contextmanager def _deliver(config: pytest.Config, invocation: _Delivery) -> Iterator[None]: __tracebackhide__ = True - state = config.stash[_RUNNER_STATE] + state = config.stash.setdefault(_RUNNER_STATE, _RunnerState()) previous = state.invocation state.invocation = invocation try: @@ -105,7 +102,9 @@ def run( config: pytest.Config, ) -> _T: __tracebackhide__ = True - if _RUNNER_STATE not in config.stash or not isinstance(coro, CoroutineType): + if not _supports_cooperative_timeouts(config) or not isinstance( + coro, CoroutineType + ): return runner.run(coro, context=context) invocation = _Delivery(runner.get_loop()) @@ -147,7 +146,7 @@ def close_unstarted(_task: asyncio.Future[Any] | None = None) -> None: def close(runner: asyncio.Runner, *, config: pytest.Config) -> None: __tracebackhide__ = True - if _RUNNER_STATE not in config.stash: + if not _supports_cooperative_timeouts(config): runner.close() return with _deliver(config, _Delivery(runner.get_loop(), closing=True)): diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index f0ceb429..a6c6f093 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -56,7 +56,6 @@ from ._timeout import ( close as _close_with_timeout, - configure as _configure_timeouts, pytest_timeout_expired as pytest_timeout_expired, run as _run_with_timeout, ) @@ -301,7 +300,6 @@ def _validate_scope(scope: str | None, option_name: str) -> None: def pytest_configure(config: Config) -> None: - _configure_timeouts(config) default_fixture_loop_scope = config.getini("asyncio_default_fixture_loop_scope") _validate_scope(default_fixture_loop_scope, "asyncio_default_fixture_loop_scope") if not default_fixture_loop_scope: diff --git a/tests/test_timeout.py b/tests/test_timeout.py index 3be64b84..077ca28c 100644 --- a/tests/test_timeout.py +++ b/tests/test_timeout.py @@ -222,7 +222,7 @@ async def body(): def test_signal_timeout_preserves_shared_loop( - pytester: Pytester, cooperative_timeout: None + pytester: Pytester, cooperative_timeout: None, monkeypatch: pytest.MonkeyPatch ): pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makepyfile(dedent("""\ @@ -231,6 +231,8 @@ def test_signal_timeout_preserves_shared_loop( import time import pytest + pytest_plugins = "pytest_timeout" + cleaned = [] async def application_wait(): @@ -279,7 +281,10 @@ def test_synchronous(): with pytest.raises(pytest.fail.Exception, match="Timeout"): signal.raise_signal(signal.SIGALRM) """)) - result = pytester.runpytest_subprocess("--tb=short", timeout=10) + monkeypatch.setenv("PYTEST_DISABLE_PLUGIN_AUTOLOAD", "1") + result = pytester.runpytest_subprocess( + "-p", "pytest_asyncio.plugin", "--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*"]) From 8d7f18a990816fb4a6d0eda31db2e2fb64d8b348 Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Sun, 23 Aug 2026 23:43:08 -0400 Subject: [PATCH 07/12] Defer coroutine creation to the running task A task factory may run the supplied coroutine inside its own wrapper. The scan added in fab62f532 then misses the owning Task and can close a live coroutine after SIGINT, breaking subsequent fixture teardown. Create native user coroutines inside the running timeout task instead. Leave the Runner-input coroutine with Runner and its task factory, removing the extra lifetime and speculative cleanup. Check the original callable before binding arguments. Marked synchronous creators and partial subclasses must retain caller-side execution and legacy signal delivery. Their custom call behavior can otherwise move into the task or disappear when functools.partial flattens its input. Exercise pre-entry expiry and indirect ownership, replacing tests for states eliminated by lazy creation. --- docs/reference/timeouts.rst | 8 +- pytest_asyncio/_timeout.py | 39 ++--- pytest_asyncio/plugin.py | 30 +++- tests/test_timeout.py | 278 +++++++++++++++--------------------- 4 files changed, 154 insertions(+), 201 deletions(-) diff --git a/docs/reference/timeouts.rst b/docs/reference/timeouts.rst index 2de95d21..c5b675cc 100644 --- a/docs/reference/timeouts.rst +++ b/docs/reference/timeouts.rst @@ -7,9 +7,11 @@ 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. Runner invocations with custom coroutine objects retain -synchronous signal delivery; cooperative execution requires native coroutine -objects created by ``async def`` functions. +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. diff --git a/pytest_asyncio/_timeout.py b/pytest_asyncio/_timeout.py index e06a1f6b..e3629df5 100644 --- a/pytest_asyncio/_timeout.py +++ b/pytest_asyncio/_timeout.py @@ -5,12 +5,10 @@ import asyncio import contextlib import contextvars -import inspect import sys import threading -from collections.abc import Coroutine, Iterator +from collections.abc import Callable, Coroutine, Iterator from dataclasses import dataclass -from types import CoroutineType from typing import Any, TypeVar import pytest @@ -96,16 +94,15 @@ def _deliver(config: pytest.Config, invocation: _Delivery) -> Iterator[None]: def run( runner: asyncio.Runner, - coro: Coroutine[Any, Any, _T], + coro_factory: Callable[[], Coroutine[Any, Any, _T]], *, context: contextvars.Context, config: pytest.Config, ) -> _T: + """Run a native coroutine factory with cooperative timeout delivery.""" __tracebackhide__ = True - if not _supports_cooperative_timeouts(config) or not isinstance( - coro, CoroutineType - ): - return runner.run(coro, context=context) + if not _supports_cooperative_timeouts(config): + return runner.run(coro_factory(), context=context) invocation = _Delivery(runner.get_loop()) @@ -116,32 +113,16 @@ async def invoke() -> _T: try: async with asyncio.timeout(None) as timeout: invocation.timeout = timeout - return await coro + # 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 - wrapped = invoke() - try: - with _deliver(config, invocation): - return runner.run(wrapped, context=context) - except BaseException: - - def close_unstarted(_task: asyncio.Future[Any] | None = None) -> None: - # Never run suspended user cleanup outside its task. - for coroutine in (wrapped, coro): - if inspect.getcoroutinestate(coroutine) == inspect.CORO_CREATED: - coroutine.close() - - # A task can own the wrapper without having started it. - for task in asyncio.all_tasks(invocation.loop): - if task.get_coro() is wrapped: - task.add_done_callback(close_unstarted) - break - else: - close_unstarted() - raise + with _deliver(config, invocation): + return runner.run(invoke(), context=context) def close(runner: asyncio.Runner, *, config: pytest.Config) -> None: diff --git a/pytest_asyncio/plugin.py b/pytest_asyncio/plugin.py index a6c6f093..85ac9cc4 100644 --- a/pytest_asyncio/plugin.py +++ b/pytest_asyncio/plugin.py @@ -412,7 +412,7 @@ async def setup(): context = contextvars.copy_context() result = _run_with_timeout( - runner, setup(), context=context, config=request.config + runner, setup, context=context, config=request.config ) reset_contextvars = _apply_contextvar_changes(context) @@ -431,7 +431,7 @@ async def async_finalizer() -> None: raise ValueError(msg) _run_with_timeout( - runner, async_finalizer(), context=context, config=request.config + runner, async_finalizer, context=context, config=request.config ) if reset_contextvars is not None: reset_contextvars() @@ -464,7 +464,7 @@ async def setup(): context = contextvars.copy_context() result = _run_with_timeout( - runner, setup(), context=context, config=request.config + runner, setup, context=context, config=request.config ) # Copy the context vars modified by the setup task into the current @@ -902,6 +902,17 @@ 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, @@ -915,8 +926,17 @@ def _synchronize_coroutine( @functools.wraps(func) def inner(*args, **kwargs): - coro = func(*args, **kwargs) - _run_with_timeout(runner, coro, context=context, config=config) + 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 diff --git a/tests/test_timeout.py b/tests/test_timeout.py index 077ca28c..37d3a69c 100644 --- a/tests/test_timeout.py +++ b/tests/test_timeout.py @@ -26,198 +26,152 @@ def cooperative_timeout(timeout_plugin: None): def test_runner_timeout_delivery(pytester: Pytester, cooperative_timeout: None): pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makeconftest(dedent("""\ - import pytest - - failures = [] - - @pytest.hookimpl(wrapper=True) - def pytest_timeout_expired(item, exception): - failures.append(exception) - return (yield) - """)) pytester.makepyfile(dedent("""\ import asyncio import contextvars + import functools import inspect import signal - import sys - from collections.abc import Coroutine import pytest - from conftest import failures - from pytest_asyncio._timeout import run + from pytest_asyncio._timeout import pytest_timeout_expired, run + from pytest_asyncio.plugin import _synchronize_coroutine - pytestmark = pytest.mark.timeout(10, method="signal", func_only=True) - - @pytest.mark.parametrize("startup", [ - "timeout", "error", "interrupt_before_start", "interrupt_during_timeout", - "interrupt_after_start", "custom_coroutine", - ]) - def test_startup(startup, request, monkeypatch): + @pytest.mark.parametrize("event", ["timeout", "interrupt"]) + def test_startup(event, request): + expired = pytest.fail.Exception("original timeout") entered = [] - wrappers = [] tasks = [] - failure = ValueError("task creation failed") async def body(): entered.append(True) - if startup == "interrupt_after_start": - try: - await asyncio.Future() - finally: - raise failure - - async def later(): return 42 - with asyncio.Runner() as runner: - loop = runner.get_loop() - - def task_factory(loop, coro, **kwargs): - loop.set_task_factory(None) - wrappers.append(coro) - if startup == "error": - raise failure - if startup == "timeout": - signal.raise_signal(signal.SIGALRM) - task = loop.create_task(coro, **kwargs) - tasks.append(task) - if startup == "custom_coroutine": - raise failure - return task - - if startup in ("interrupt_before_start", "interrupt_during_timeout"): - target, name = ( - (loop, "run_until_complete") - if startup == "interrupt_before_start" - else (asyncio, "timeout") - ) - original = getattr(target, name) + def task_factory(loop, coro, **kwargs): + loop.set_task_factory(None) - def interrupt(*args, **kwargs): - monkeypatch.setattr(target, name, original) - signal.raise_signal(signal.SIGINT) - signal.raise_signal(signal.SIGINT) - return original(*args, **kwargs) - - monkeypatch.setattr(target, name, interrupt) - - loop.set_task_factory(task_factory) - coro = body() - argument = coro - if startup == "custom_coroutine": - class CustomCoroutine(Coroutine): - def send(self, value): - return coro.send(value) - - def throw(self, *args): - return coro.throw(*args) - - def __await__(self): - return coro.__await__() - - argument = CustomCoroutine() - - original_trace = sys.gettrace() - if startup == "interrupt_after_start": - def trace(frame, event, arg): - if ( - wrappers and frame.f_code is wrappers[0].cr_code - and event == "return" and arg is not None - ): - sys.settrace(None) - signal.raise_signal(signal.SIGINT) - signal.raise_signal(signal.SIGINT) - return trace - - sys.settrace(trace) - - expected = { - "timeout": pytest.fail.Exception, "error": ValueError, - "custom_coroutine": ValueError, - }.get(startup, KeyboardInterrupt) - try: - with pytest.raises(expected) as caught: - run( - runner, argument, context=contextvars.copy_context(), - config=request.config, - ) - finally: - sys.settrace(original_trace) - if startup in ("timeout", "error", "custom_coroutine"): - assert caught.value is ( - failures[-1] if startup == "timeout" else failure - ) - if startup == "interrupt_after_start": - with pytest.raises(ValueError) as cleanup: - coro.close() - assert cleanup.value is failure - if startup != "custom_coroutine": - assert run( - runner, later(), context=contextvars.copy_context(), + 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, - ) == 42 - assert bool(entered) is (startup == "interrupt_after_start") - assert all( - inspect.getcoroutinestate(c) == inspect.CORO_CLOSED - for c in [coro, *wrappers] if inspect.iscoroutine(c) - ) - if startup in ("interrupt_before_start", "custom_coroutine"): - assert tasks[0].cancelled() - elif startup in ("interrupt_during_timeout", "interrupt_after_start"): - assert tasks[0].exception() is caught.value + ) + 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", ["native", "synchronous", "partial_subclass"]) + def test_creator_context(kind, request): + value = contextvars.ContextVar("value", default="caller") + events = [] + expired = pytest.fail.Exception("original timeout") + + class Owner: + async def body(self, argument): + events.append(("body", value.get(), argument)) + value.set("updated") + if kind == "native": + pytest_timeout_expired(request.node, expired) + + native = functools.partial(Owner().body) + + def creator(argument): + events.append(("creator", value.get())) + return native(argument) + + if hasattr(inspect, "markcoroutinefunction"): + inspect.markcoroutinefunction(creator) + + class SyncPartial(functools.partial): + def __call__(self, argument): + events.append(("creator", value.get())) + return super().__call__(argument) + + func = { + "native": native, + "synchronous": creator, + "partial_subclass": SyncPartial(native), + }[kind] + context = contextvars.copy_context() + context.run(value.set, "task") + with asyncio.Runner() as runner: + synchronized = _synchronize_coroutine( + func, runner, context, request.config + ) + if kind == "native": + with pytest.raises(type(expired)) as caught: + synchronized(42) + assert caught.value is expired + else: + synchronized(42) + expected = [("body", "task", 42)] + if kind != "native": + expected.insert(0, ("creator", "caller")) + assert events == expected + assert value.get() == "caller" + assert context.get(value) == "updated" @pytest.mark.parametrize( - "cleanup", - ["return", "xfail", "error", "interrupt", "exit", "system_exit", "cancel"], + ("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, request): + def test_cleanup(cleanup, expected, request): + expired = pytest.fail.Exception("original timeout") + async def body(): loop = asyncio.get_running_loop() - loop.call_soon(signal.raise_signal, signal.SIGALRM) + loop.call_soon(pytest_timeout_expired, request.node, expired) try: await asyncio.Future() except asyncio.CancelledError: - if cleanup == "return": - return - if cleanup == "xfail": - pytest.xfail("cleanup xfail") - if cleanup == "error": - raise ValueError("cleanup error") - if cleanup == "exit": - pytest.exit("requested exit", returncode=4) - if cleanup == "system_exit": - raise SystemExit(7) - if cleanup == "cancel": - asyncio.current_task().cancel() - else: + if cleanup == "interrupt": signal.raise_signal(signal.SIGINT) + elif cleanup == "cancel": + asyncio.current_task().cancel() + elif cleanup is not None: + raise cleanup await asyncio.sleep(0) - expected = { - "interrupt": KeyboardInterrupt, - "exit": pytest.exit.Exception, - "system_exit": SystemExit, - "cancel": asyncio.CancelledError, - }.get(cleanup, pytest.fail.Exception) with asyncio.Runner() as runner: with pytest.raises(expected) as caught: run( - runner, body(), context=contextvars.copy_context(), + runner, body, context=contextvars.copy_context(), config=request.config, ) if cleanup == "cancel": - assert caught.value.__cause__ is failures[-1] - elif cleanup == "system_exit": - assert caught.value.code == 7 - elif cleanup == "exit": - assert caught.value.returncode == 4 - assert str(caught.value) == "requested exit" + assert caught.value.__cause__ is expired elif cleanup != "interrupt": - assert caught.value is failures[-1] + assert caught.value is ( + expired if expected is pytest.fail.Exception else cleanup + ) """)) result = pytester.runpytest_subprocess(timeout=10) - result.assert_outcomes(passed=13) + result.assert_outcomes(passed=12) assert "was never awaited" not in result.stdout.str() + result.stderr.str() @@ -302,12 +256,8 @@ def test_timeout_during_async_cleanup(pytester: Pytester, cooperative_timeout: N import pytest_asyncio executor = None + worker = None release = threading.Event() - finished = threading.Event() - - def worker(): - release.wait(5) - finished.set() async def timeout(): asyncio.get_running_loop().call_soon(signal.raise_signal, signal.SIGALRM) @@ -342,7 +292,7 @@ async def fixture(phase): @pytest.mark.timeout(10, method="signal") @pytest.mark.asyncio async def test_timeout(phase, coroutine, fixture): - global executor + global executor, worker if phase == "shutdown": asyncio.create_task(background()) await asyncio.sleep(0) @@ -350,7 +300,7 @@ async def test_timeout(phase, coroutine, fixture): loop = asyncio.get_running_loop() executor = ThreadPoolExecutor() loop.set_default_executor(executor) - loop.run_in_executor(None, worker) + worker = executor.submit(release.wait, 5) original = loop.shutdown_asyncgens async def shutdown_asyncgens(): @@ -362,7 +312,7 @@ async def shutdown_asyncgens(): def test_later(): if executor is not None: try: - assert not finished.is_set() + assert not worker.done() finally: release.set() executor.shutdown(wait=True) From 0de5523d22960dddc82af4c045c0a1ec739e0380 Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Sun, 23 Aug 2026 23:43:45 -0400 Subject: [PATCH 08/12] Cover timeouts on free-threaded Python The free-threaded CI job has no tox-gh-actions mapping and falls back to the generic py environment. Pytest-timeout is absent there, so the job passes while skipping every timeout integration test. Add an explicit py314t environment and select it alongside the pinned pytest-timeout environment for Python 3.14t. This preserves baseline coverage and exercises the integration with the free-threaded build. --- tox.ini | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tox.ini b/tox.ini index f223da49..84b3fb4b 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, pytest-timeout-dev, docs, pyright +envlist = build, py310, py311, py312, py313, py314, py314t, py310-lower-bounds, pytest-timeout-dev, docs, pyright isolated_build = true passenv = CI @@ -99,4 +99,5 @@ python = 3.12: py312 3.13: py313, pyright 3.14: py314, pytest-timeout-dev + 3.14t: py314t, pytest-timeout-dev pypy3: pypy3 From 86efcea90db310992957ae7761c0cbfcdd6317cc Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Mon, 24 Aug 2026 00:05:08 -0400 Subject: [PATCH 09/12] Run timeout adapter tests without a subprocess Only the real signal and shutdown regressions need process isolation. The direct runner checks use an unbounded wait, which makes missing cancellation hang instead of producing an assertion failure. Bound that wait and explicitly verify cancellation, then collect those cases as ordinary tests. This removes a generated test module and a pytest subprocess while exposing the code to normal lint and typing checks. --- tests/test_timeout.py | 294 +++++++++++++++++++++--------------------- 1 file changed, 148 insertions(+), 146 deletions(-) diff --git a/tests/test_timeout.py b/tests/test_timeout.py index 37d3a69c..ccef3ea7 100644 --- a/tests/test_timeout.py +++ b/tests/test_timeout.py @@ -1,12 +1,22 @@ 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" ) @@ -24,155 +34,147 @@ def cooperative_timeout(timeout_plugin: None): pytest.skip("cooperative timeouts require Python 3.11 or newer") -def test_runner_timeout_delivery(pytester: Pytester, cooperative_timeout: None): - pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") - pytester.makepyfile(dedent("""\ - import asyncio - import contextvars - import functools - import inspect - import signal - import pytest - from pytest_asyncio._timeout import pytest_timeout_expired, run - from pytest_asyncio.plugin import _synchronize_coroutine - - @pytest.mark.parametrize("event", ["timeout", "interrupt"]) - def test_startup(event, request): - 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.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", ["native", "synchronous", "partial_subclass"]) +def test_creator_context( + kind: str, request: pytest.FixtureRequest, cooperative_timeout: None +): + value = contextvars.ContextVar("value", default="caller") + events = [] + expired = pytest.fail.Exception("original timeout") + + class Owner: + async def body(self, argument): + events.append(("body", value.get(), argument)) + value.set("updated") + if kind == "native": 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", ["native", "synchronous", "partial_subclass"]) - def test_creator_context(kind, request): - value = contextvars.ContextVar("value", default="caller") - events = [] - expired = pytest.fail.Exception("original timeout") - - class Owner: - async def body(self, argument): - events.append(("body", value.get(), argument)) - value.set("updated") - if kind == "native": - pytest_timeout_expired(request.node, expired) - - native = functools.partial(Owner().body) - - def creator(argument): - events.append(("creator", value.get())) - return native(argument) - - if hasattr(inspect, "markcoroutinefunction"): - inspect.markcoroutinefunction(creator) - - class SyncPartial(functools.partial): - def __call__(self, argument): - events.append(("creator", value.get())) - return super().__call__(argument) - - func = { - "native": native, - "synchronous": creator, - "partial_subclass": SyncPartial(native), - }[kind] - context = contextvars.copy_context() - context.run(value.set, "task") - with asyncio.Runner() as runner: - synchronized = _synchronize_coroutine( - func, runner, context, request.config - ) - if kind == "native": - with pytest.raises(type(expired)) as caught: - synchronized(42) - assert caught.value is expired - else: - synchronized(42) - expected = [("body", "task", 42)] - if kind != "native": - expected.insert(0, ("creator", "caller")) - assert events == expected - 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), - ], + native = functools.partial(Owner().body) + + def creator(argument): + events.append(("creator", value.get())) + return native(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]] = { + "native": native, + "synchronous": creator, + "partial_subclass": SyncPartial(native), + } + context = contextvars.copy_context() + context.run(value.set, "task") + with asyncio.Runner() as runner: + synchronized = _synchronize_coroutine( + functions[kind], runner, context, request.config + ) + if kind == "native": + with pytest.raises(type(expired)) as caught: + synchronized(42) + assert caught.value is expired + else: + synchronized(42) + assert events[-1] == ("body", "task", 42) + assert events[:-1] == ([] if kind == "native" else [("creator", "caller")]) + 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_cleanup(cleanup, expected, request): - expired = pytest.fail.Exception("original timeout") - - async def body(): - loop = asyncio.get_running_loop() - loop.call_soon(pytest_timeout_expired, request.node, expired) - try: - await asyncio.Future() - except asyncio.CancelledError: - if cleanup == "interrupt": - signal.raise_signal(signal.SIGINT) - elif cleanup == "cancel": - asyncio.current_task().cancel() - elif cleanup is not None: - raise cleanup - await asyncio.sleep(0) - - with asyncio.Runner() as runner: - with pytest.raises(expected) as caught: - run( - runner, body, context=contextvars.copy_context(), - config=request.config, - ) - 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 - ) - """)) - result = pytester.runpytest_subprocess(timeout=10) - result.assert_outcomes(passed=12) - assert "was never awaited" not in result.stdout.str() + result.stderr.str() def test_signal_timeout_preserves_shared_loop( From fd2a673eda3a0223f0684ac18d8981a36775b5d1 Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Mon, 24 Aug 2026 00:05:20 -0400 Subject: [PATCH 10/12] Limit cooperative delivery to the main thread Pytest-timeout invokes the expiry hook from SIGALRM and uses its terminating thread method when timers are started off the main thread. Worker runners therefore have no cooperative signal-delivery path. Keep only the active main-thread invocation in the config stash. This removes thread-local state and leaves worker runners unwrapped. Both runner entry and the expiry hook check the thread, so a foreign hook call cannot claim the main thread's timeout. --- pytest_asyncio/_timeout.py | 35 ++++++++++++++++++----------------- 1 file changed, 18 insertions(+), 17 deletions(-) diff --git a/pytest_asyncio/_timeout.py b/pytest_asyncio/_timeout.py index e3629df5..60acb70b 100644 --- a/pytest_asyncio/_timeout.py +++ b/pytest_asyncio/_timeout.py @@ -14,10 +14,6 @@ import pytest -class _RunnerState(threading.local): - invocation: _Delivery | None = None - - @dataclass class _Delivery: loop: asyncio.AbstractEventLoop @@ -25,33 +21,39 @@ class _Delivery: exception: BaseException | None = None timeout: asyncio.Timeout | None = None - def interrupt(self, state: _RunnerState) -> None: - if state.invocation is not self: + def interrupt(self, config: pytest.Config) -> None: + if 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, state) + self.loop.call_soon(self.interrupt, config) elif self.timeout is not None: self.timeout.reschedule(self.loop.time()) -_RUNNER_STATE = pytest.StashKey[_RunnerState]() +# SIGALRM only reaches the main thread; worker runners keep their native behavior. +_CURRENT_DELIVERY = pytest.StashKey[_Delivery | None]() _T = TypeVar("_T") 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 config.hook.pytest_timeout_expired.has_spec() + 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: - state = item.config.stash.get(_RUNNER_STATE, None) - if state is None or state.invocation is 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 - invocation = state.invocation if invocation.exception is None: invocation.exception = exception # Raising here can interrupt asyncio before it schedules a task's next @@ -59,23 +61,22 @@ def pytest_timeout_expired(item: pytest.Item, exception: BaseException) -> bool # 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, state) + invocation.loop.call_soon_threadsafe(invocation.interrupt, item.config) return True @contextlib.contextmanager def _deliver(config: pytest.Config, invocation: _Delivery) -> Iterator[None]: __tracebackhide__ = True - state = config.stash.setdefault(_RUNNER_STATE, _RunnerState()) - previous = state.invocation - state.invocation = invocation + previous = config.stash.get(_CURRENT_DELIVERY, None) try: try: + config.stash[_CURRENT_DELIVERY] = invocation yield finally: # Once the runner returns, a new signal can fail synchronously. # Stop claiming it before deciding which outcome to propagate. - state.invocation = previous + config.stash[_CURRENT_DELIVERY] = previous except (KeyboardInterrupt, SystemExit, pytest.exit.Exception): raise except asyncio.CancelledError as exc: From 3ef5eb3ebbf254ea6998c216fa99f2917bd01880 Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Mon, 24 Aug 2026 00:41:07 -0400 Subject: [PATCH 11/12] Run timeout delivery through its owner Delivery owns the active runner state, but a separate generator context manager controls its lifetime. Execute the runner operation on _Delivery so registration, restoration, and failure propagation live together. Keep registration and restoration inside the protected region, and hide all delivery frames through the module's traceback flag. --- pytest_asyncio/_timeout.py | 81 ++++++++++++++++++-------------------- 1 file changed, 38 insertions(+), 43 deletions(-) diff --git a/pytest_asyncio/_timeout.py b/pytest_asyncio/_timeout.py index 60acb70b..e6021b4a 100644 --- a/pytest_asyncio/_timeout.py +++ b/pytest_asyncio/_timeout.py @@ -3,39 +3,67 @@ from __future__ import annotations import asyncio -import contextlib import contextvars import sys import threading -from collections.abc import Callable, Coroutine, Iterator +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 interrupt(self, config: pytest.Config) -> None: - if config.stash.get(_CURRENT_DELIVERY, None) is not self: + 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, config) + 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]() -_T = TypeVar("_T") def _supports_cooperative_timeouts(config: pytest.Config) -> bool: @@ -61,38 +89,10 @@ def pytest_timeout_expired(item: pytest.Item, exception: BaseException) -> bool # 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, item.config) + invocation.loop.call_soon_threadsafe(invocation.interrupt) return True -@contextlib.contextmanager -def _deliver(config: pytest.Config, invocation: _Delivery) -> Iterator[None]: - __tracebackhide__ = True - previous = config.stash.get(_CURRENT_DELIVERY, None) - try: - try: - config.stash[_CURRENT_DELIVERY] = invocation - yield - finally: - # Once the runner returns, a new signal can fail synchronously. - # Stop claiming it before deciding which outcome to propagate. - config.stash[_CURRENT_DELIVERY] = previous - except (KeyboardInterrupt, SystemExit, pytest.exit.Exception): - raise - except asyncio.CancelledError as exc: - if invocation.exception is None: - raise - # asyncio.Timeout converts only its own cancellation to TimeoutError. - # Preserve cancellation requested by another caller. - raise exc from invocation.exception - except BaseException as exc: - if invocation.exception is None or exc is invocation.exception: - raise - raise invocation.exception from exc - if invocation.exception is not None: - raise invocation.exception - - def run( runner: asyncio.Runner, coro_factory: Callable[[], Coroutine[Any, Any, _T]], @@ -101,14 +101,12 @@ def run( config: pytest.Config, ) -> _T: """Run a native coroutine factory with cooperative timeout delivery.""" - __tracebackhide__ = True if not _supports_cooperative_timeouts(config): return runner.run(coro_factory(), context=context) - invocation = _Delivery(runner.get_loop()) + invocation = _Delivery(config, runner.get_loop()) async def invoke() -> _T: - __tracebackhide__ = True if invocation.exception is not None: raise invocation.exception try: @@ -122,14 +120,11 @@ async def invoke() -> _T: # Do not reschedule a timeout whose context has already exited. invocation.timeout = None - with _deliver(config, invocation): - return runner.run(invoke(), context=context) + return invocation.run(lambda: runner.run(invoke(), context=context)) def close(runner: asyncio.Runner, *, config: pytest.Config) -> None: - __tracebackhide__ = True if not _supports_cooperative_timeouts(config): runner.close() return - with _deliver(config, _Delivery(runner.get_loop(), closing=True)): - runner.close() + _Delivery(config, runner.get_loop(), closing=True).run(runner.close) From f729f0f57dbc1619903c065dade220b48c2e756e Mon Sep 17 00:00:00 2001 From: Tamir Duberstein Date: Mon, 24 Aug 2026 00:41:59 -0400 Subject: [PATCH 12/12] Cover native timeouts across pytest versions Exercise partials and bound methods through the existing real-signal integration, leaving the direct creator test focused on synchronous call timing and context. The late-plugin bootstrap fails during collection on pytest before 9.1: option registration only supplies explicitly declared defaults [1]. Preload pytest-timeout there and retain late-loading coverage on newer pytest. Pin the existing Python 3.11 integration environment to pytest 8.4.0 so cooperative delivery covers the supported pytest floor. [1]: https://github.com/pytest-dev/pytest/blob/8.4.0/src/_pytest/config/__init__.py#L1206-L1212 --- tests/test_timeout.py | 60 +++++++++++++++++-------------------------- tox.ini | 2 ++ 2 files changed, 26 insertions(+), 36 deletions(-) diff --git a/tests/test_timeout.py b/tests/test_timeout.py index ccef3ea7..1d575f56 100644 --- a/tests/test_timeout.py +++ b/tests/test_timeout.py @@ -70,26 +70,20 @@ async def traced(): assert not entered -@pytest.mark.parametrize("kind", ["native", "synchronous", "partial_subclass"]) +@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 = [] - expired = pytest.fail.Exception("original timeout") - - class Owner: - async def body(self, argument): - events.append(("body", value.get(), argument)) - value.set("updated") - if kind == "native": - pytest_timeout_expired(request.node, expired) - native = functools.partial(Owner().body) + async def body(argument): + events.append(("body", value.get(), argument)) + value.set("updated") def creator(argument): events.append(("creator", value.get())) - return native(argument) + return body(argument) if hasattr(inspect, "markcoroutinefunction"): inspect.markcoroutinefunction(creator) @@ -100,9 +94,8 @@ def __call__(self, *args, **kwargs): return super().__call__(*args, **kwargs) functions: dict[str, Callable[..., CoroutineType]] = { - "native": native, "synchronous": creator, - "partial_subclass": SyncPartial(native), + "partial_subclass": SyncPartial(body), } context = contextvars.copy_context() context.run(value.set, "task") @@ -110,14 +103,8 @@ def __call__(self, *args, **kwargs): synchronized = _synchronize_coroutine( functions[kind], runner, context, request.config ) - if kind == "native": - with pytest.raises(type(expired)) as caught: - synchronized(42) - assert caught.value is expired - else: - synchronized(42) - assert events[-1] == ("body", "task", 42) - assert events[:-1] == ([] if kind == "native" else [("creator", "caller")]) + synchronized(42) + assert events == [("creator", "caller"), ("body", "task", 42)] assert value.get() == "caller" assert context.get(value) == "updated" @@ -183,6 +170,7 @@ def test_signal_timeout_preserves_shared_loop( pytester.makeini("[pytest]\nasyncio_default_fixture_loop_scope = function") pytester.makepyfile(dedent("""\ import asyncio + import functools import signal import time import pytest @@ -194,19 +182,9 @@ def test_signal_timeout_preserves_shared_loop( async def application_wait(): await asyncio.Future() - @pytest.mark.parametrize( - "trigger", - [ - pytest.param("timer", marks=pytest.mark.timeout( - 0.1, method="signal", func_only=True, - )), - pytest.param("reschedule", marks=pytest.mark.timeout( - 10, method="signal", func_only=True, - )), - ], - ) + @pytest.mark.timeout(0.1, method="signal", func_only=True) @pytest.mark.asyncio(loop_scope="module") - async def test_timeout(trigger): + async def timeout(trigger="timer"): loop = asyncio.get_running_loop() original = loop.call_soon task = asyncio.current_task() @@ -228,6 +206,14 @@ def reschedule(callback, *args, context=None): 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"] @@ -238,9 +224,11 @@ def test_synchronous(): signal.raise_signal(signal.SIGALRM) """)) monkeypatch.setenv("PYTEST_DISABLE_PLUGIN_AUTOLOAD", "1") - result = pytester.runpytest_subprocess( - "-p", "pytest_asyncio.plugin", "--tb=short", timeout=10 - ) + 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*"]) diff --git a/tox.ini b/tox.ini index 84b3fb4b..eab431ae 100644 --- a/tox.ini +++ b/tox.ini @@ -42,6 +42,8 @@ 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