Add cooperative signal timeouts

changelog.d/215.added.rst

@@ -0,0 +1 @@
+Automatically deliver pytest-timeout signal failures cooperatively in asynchronous tests and fixtures on Python 3.11 and newer when pytest-timeout provides its expiry hook.

docs/reference/index.rst

@@ -9,6 +9,7 @@ Reference
   fixtures/index
   functions
   hooks
+  timeouts
   markers/index
   decorators/index
   changelog

docs/reference/timeouts.rst

@@ -0,0 +1,30 @@
+========
+Timeouts
+========
+
+On Python 3.11 and newer, pytest-asyncio automatically delivers pytest-timeout
+signal failures by cancelling the active asynchronous test or fixture when
+pytest-timeout provides the ``pytest_timeout_expired`` hook. Cooperative cleanup
+can then run before pytest reports the original timeout. No configuration is
+needed. Python 3.10 and older pytest-timeout versions retain their existing
+signal behavior. Cooperative test execution requires a native ``async def``
+function, bound method, or plain ``functools.partial`` of one. Synchronous
+coroutine creators, including functions marked with
+``inspect.markcoroutinefunction()`` and subclasses of ``functools.partial``,
+retain their existing call timing, context, and synchronous signal delivery.
+pytest-timeout still controls the configured duration, covered test phases,
+debugger detection, and timeout diagnostics.
+
+Cancellation waits for the event loop and coroutine to cooperate. A raised
+signal exception can interrupt CPU-bound Python code, but cooperative
+cancellation cannot interrupt code that never yields. It also cannot stop
+an indefinitely blocking callback or a task that refuses cancellation.
+Use pytest-timeout's ``thread`` method or an independent process watchdog when
+the process must be terminated; these stop the entire process without normal
+test teardown. A timeout during final event-loop shutdown stops that shutdown
+and closes the loop; remaining resource cleanup may be incomplete.
+
+The integration does not take over runners managed by other async plugins or
+synchronous tests that call ``asyncio.run()`` themselves. It wraps asynchronous
+tests and fixtures in a timeout context, so ``asyncio.current_task().get_coro()``
+returns the wrapper coroutine rather than the original test coroutine.

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
 

pytest_asyncio/_timeout.py

@@ -0,0 +1,130 @@
+"""Cooperative delivery of pytest-timeout's signal failures."""
+
+from __future__ import annotations
+
+import asyncio
+import contextvars
+import sys
+import threading
+from collections.abc import Callable, Coroutine
+from dataclasses import dataclass
+from typing import Any, TypeVar
+
+import pytest
+
+__tracebackhide__ = True
+_T = TypeVar("_T")
+
+
+@dataclass
+class _Delivery:
+    config: pytest.Config
+    loop: asyncio.AbstractEventLoop
+    closing: bool = False
+    exception: BaseException | None = None
+    timeout: asyncio.Timeout | None = None
+
+    def run(self, operation: Callable[[], _T]) -> _T:
+        previous = self.config.stash.get(_CURRENT_DELIVERY, None)
+        try:
+            try:
+                self.config.stash[_CURRENT_DELIVERY] = self
+                result = operation()
+            finally:
+                # Once the runner returns, a new signal can fail synchronously.
+                # Stop claiming it before deciding which outcome to propagate.
+                self.config.stash[_CURRENT_DELIVERY] = previous
+        except (KeyboardInterrupt, SystemExit, pytest.exit.Exception):
+            raise
+        except asyncio.CancelledError as exc:
+            if self.exception is None:
+                raise
+            # asyncio.Timeout converts only its own cancellation to TimeoutError.
+            # Preserve cancellation requested by another caller.
+            raise exc from self.exception
+        except BaseException as exc:
+            if self.exception is None or exc is self.exception:
+                raise
+            raise self.exception from exc
+        if self.exception is not None:
+            raise self.exception
+        return result
+
+    def interrupt(self) -> None:
+        if self.config.stash.get(_CURRENT_DELIVERY, None) is not self:
+            return
+        if self.closing:
+            # A completed shutdown phase can consume stop(). Keep stopping
+            # until Runner.close() returns; never stop a reusable invocation.
+            self.loop.stop()
+            self.loop.call_soon(self.interrupt)
+        elif self.timeout is not None:
+            self.timeout.reschedule(self.loop.time())
+
+
+# SIGALRM only reaches the main thread; worker runners keep their native behavior.
+_CURRENT_DELIVERY = pytest.StashKey[_Delivery | None]()
+
+
+def _supports_cooperative_timeouts(config: pytest.Config) -> bool:
+    # Test modules can load pytest-timeout after pytest_configure has run.
+    return (
+        sys.version_info >= (3, 11)
+        and threading.current_thread() is threading.main_thread()
+        and config.hook.pytest_timeout_expired.has_spec()
+    )
+
+
+@pytest.hookimpl(tryfirst=True, optionalhook=True)
+def pytest_timeout_expired(item: pytest.Item, exception: BaseException) -> bool | None:
+    if threading.current_thread() is not threading.main_thread():
+        return None
+    invocation = item.config.stash.get(_CURRENT_DELIVERY, None)
+    if invocation is None:
+        return None
+    if invocation.exception is None:
+        invocation.exception = exception
+        # Raising here can interrupt asyncio before it schedules a task's next
+        # step. Return to the interrupted code and cancel at a safe loop turn.
+        # Late callbacks check ownership instead of relying on Handle.cancel():
+        # SIGINT can interrupt scheduling before the handle is returned.
+        if not invocation.loop.is_closed():
+            invocation.loop.call_soon_threadsafe(invocation.interrupt)
+    return True
+
+
+def run(
+    runner: asyncio.Runner,
+    coro_factory: Callable[[], Coroutine[Any, Any, _T]],
+    *,
+    context: contextvars.Context,
+    config: pytest.Config,
+) -> _T:
+    """Run a native coroutine factory with cooperative timeout delivery."""
+    if not _supports_cooperative_timeouts(config):
+        return runner.run(coro_factory(), context=context)
+
+    invocation = _Delivery(config, runner.get_loop())
+
+    async def invoke() -> _T:
+        if invocation.exception is not None:
+            raise invocation.exception
+        try:
+            async with asyncio.timeout(None) as timeout:
+                invocation.timeout = timeout
+                # Create the user coroutine only once its task owns execution.
+                # Runner and task factories retain ownership of invoke().
+                return await coro_factory()
+        finally:
+            # The signal may have queued delivery just as the coroutine exits.
+            # Do not reschedule a timeout whose context has already exited.
+            invocation.timeout = None
+
+    return invocation.run(lambda: runner.run(invoke(), context=context))
+
+
+def close(runner: asyncio.Runner, *, config: pytest.Config) -> None:
+    if not _supports_cooperative_timeouts(config):
+        runner.close()
+        return
+    _Delivery(config, runner.get_loop(), closing=True).run(runner.close)

pytest_asyncio/plugin.py

@@ -54,6 +54,12 @@
     PytestPluginManager,
 )
 
+from ._timeout import (
+    close as _close_with_timeout,
+    pytest_timeout_expired as pytest_timeout_expired,
+    run as _run_with_timeout,
+)
+
 if sys.version_info >= (3, 11):
     from asyncio import Runner
 else:
@@ -405,7 +411,9 @@ async def setup():
             return res
 
         context = contextvars.copy_context()
-        result = runner.run(setup(), context=context)
+        result = _run_with_timeout(
+            runner, setup, context=context, config=request.config
+        )
 
         reset_contextvars = _apply_contextvar_changes(context)
 
@@ -422,7 +430,9 @@ async def async_finalizer() -> None:
                     msg += "Yield only once."
                     raise ValueError(msg)
 
-            runner.run(async_finalizer(), context=context)
+            _run_with_timeout(
+                runner, async_finalizer, context=context, config=request.config
+            )
             if reset_contextvars is not None:
                 reset_contextvars()
 
@@ -453,7 +463,9 @@ async def setup():
             return res
 
         context = contextvars.copy_context()
-        result = runner.run(setup(), context=context)
+        result = _run_with_timeout(
+            runner, setup, context=context, config=request.config
+        )
 
         # Copy the context vars modified by the setup task into the current
         # context, and (if needed) add a finalizer to reset them.
@@ -563,7 +575,7 @@ def runtest(self) -> None:
         runner = self._request.getfixturevalue(runner_fixture_id)
         context = contextvars.copy_context()
         synchronized_obj = _synchronize_coroutine(
-            getattr(*self._synchronization_target_attr), runner, context
+            getattr(*self._synchronization_target_attr), runner, context, self.config
         )
         with MonkeyPatch.context() as c:
             c.setattr(*self._synchronization_target_attr, synchronized_obj)
@@ -890,10 +902,22 @@ def pytest_pyfunc_call(pyfuncitem: Function) -> object | None:
     return None
 
 
+def _is_native_coroutine_function(func: object) -> bool:
+    # Partial subclasses can override __call__; binding them can discard it.
+    if type(func) is functools.partial:
+        return _is_native_coroutine_function(func.func)
+    if inspect.ismethod(func):
+        return _is_native_coroutine_function(func.__func__)
+    return inspect.isfunction(func) and bool(
+        func.__code__.co_flags & inspect.CO_COROUTINE
+    )
+
+
 def _synchronize_coroutine(
     func: Callable[..., CoroutineType],
     runner: asyncio.Runner,
     context: contextvars.Context,
+    config: Config,
 ):
     """
     Return a sync wrapper around a coroutine executing it in the
@@ -902,8 +926,17 @@ def _synchronize_coroutine(
 
     @functools.wraps(func)
     def inner(*args, **kwargs):
-        coro = func(*args, **kwargs)
-        runner.run(coro, context=context)
+        if not _is_native_coroutine_function(func):
+            # Synchronous creators must run in the caller's context, even when
+            # inspect.markcoroutinefunction() marks them as coroutine functions.
+            runner.run(func(*args, **kwargs), context=context)
+            return
+        _run_with_timeout(
+            runner,
+            functools.partial(func, *args, **kwargs),
+            context=context,
+            config=config,
+        )
 
     return inner
 
@@ -1042,15 +1075,15 @@ def _scoped_runner(
                 _set_event_loop(runner.get_loop())
             try:
                 yield runner
-            except Exception as e:
-                runner.__exit__(type(e), e, e.__traceback__)
+            except Exception:
+                _close_with_timeout(runner, config=request.config)
             else:
                 with warnings.catch_warnings():
                     warnings.filterwarnings(
                         "ignore", ".*BaseEventLoop.shutdown_asyncgens.*", RuntimeWarning
                     )
                     try:
-                        runner.__exit__(None, None, None)
+                        _close_with_timeout(runner, config=request.config)
                     except RuntimeError:
                         warnings.warn(
                             _RUNNER_TEARDOWN_WARNING % traceback.format_exc(),