diff --git a/README.rst b/README.rst index 02e5e8f..d62c9a3 100644 --- a/README.rst +++ b/README.rst @@ -260,9 +260,8 @@ variable. Extending pytest-timeout with plugins ===================================== -``pytest-timeout`` provides two hooks that can be used for extending the tool. These -hooks are used for setting the timeout timer and cancelling it if the timeout is not -reached. +``pytest-timeout`` provides hooks for replacing the timeout timer or delivering a +signal timeout through another plugin's execution boundary. For example, ``pytest-asyncio`` can provide asyncio-specific code that generates better traceback and points on timed out ``await`` instead of the running loop iteration. @@ -321,6 +320,31 @@ The argument has ``Settings`` namedtuple type with the following fields: Can be overridden by plugins for alternative timeout implementation strategies. """ +``pytest_timeout_expired`` +-------------------------- + +.. code:: python + + @pytest.hookspec(firstresult=True) + def pytest_timeout_expired(item, settings, exception): + """Return True to take responsibility for reporting a signal timeout.""" + +This hook runs synchronously inside the ``SIGALRM`` handler, after debugger +detection and stack diagnostics. ``exception`` is the ``pytest.fail.Exception`` +instance that would normally interrupt the test. Return ``None`` to decline; +unless an implementation returns ``True``, pytest-timeout raises that same +exception immediately. + +A plugin that returns ``True`` must arrange to report the supplied exception at +its execution boundary. For example, an async test runner can request task +cancellation and raise the exception when the task finishes. The hook does not +rearm the one-shot signal timer or provide a cleanup deadline. Cooperative +cancellation cannot stop a blocked event loop or a task that refuses to finish; +use an independent process watchdog when termination must be guaranteed. + +The hook is not called for the ``thread`` method, which still terminates the +process. It does not change which test phases the configured timeout covers. + ``is_debugging`` ---------------- @@ -392,6 +416,12 @@ to 100 seconds:: Changelog ========= +Unreleased +---------- + +- Add ``pytest_timeout_expired`` so plugins can deliver a signal timeout at + their own execution boundary without replacing its timer or debugger handling. + 2.5.0 ----- diff --git a/pytest_timeout.py b/pytest_timeout.py index 192a16c..7ce4b69 100644 --- a/pytest_timeout.py +++ b/pytest_timeout.py @@ -129,6 +129,18 @@ def pytest_timeout_cancel_timer(self, item): """ + @pytest.hookspec(firstresult=True) + def pytest_timeout_expired(self, item, settings, exception): + """Deliver a signal timeout after debugger checks and stack diagnostics. + + Called synchronously from the signal handler. Return True to take + responsibility for reporting the supplied pytest.fail.Exception at a + safe execution boundary, or None to retain the default behavior. + The default behavior raises the same exception immediately. + Claiming delivery does not rearm the timer or guarantee termination. + This hook is not called for the thread timeout method. + """ + def pytest_addhooks(pluginmanager): """Register timeout-specific hooks.""" @@ -476,12 +488,7 @@ def _validate_disable_debugger_detection(disable_debugger_detection, where): def timeout_sigalrm(item, settings): - """Dump stack of threads and raise an exception. - - This will output the stacks of any threads other than the - current to stderr and then raise an AssertionError, thus - terminating the test. - """ + """Dump thread stacks and deliver the timeout failure.""" if not settings.disable_debugger_detection and is_debugging(): return __tracebackhide__ = True @@ -492,7 +499,12 @@ def timeout_sigalrm(item, settings): dump_stacks(terminal) if nthreads > 1: terminal.sep("+", title="Timeout") - pytest.fail(PYTEST_FAILURE_MESSAGE % settings.timeout) + exception = pytest.fail.Exception(PYTEST_FAILURE_MESSAGE % settings.timeout) + handled = item.ihook.pytest_timeout_expired( + item=item, settings=settings, exception=exception + ) + if handled is not True: + raise exception def timeout_timer(item, settings): diff --git a/test_pytest_timeout.py b/test_pytest_timeout.py index ff36c40..1a8ec8f 100644 --- a/test_pytest_timeout.py +++ b/test_pytest_timeout.py @@ -681,6 +681,88 @@ def test_foo(): ) +@pytest.mark.parametrize( + ("claim", "debugging", "disable_debugger_detection"), + [ + (None, False, False), + (False, False, False), + (True, False, False), + (True, True, False), + (True, True, True), + ], +) +def test_signal_expiry_hook( + request, monkeypatch, claim, debugging, disable_debugger_detection +): + import pytest_timeout + + settings = pytest_timeout.Settings(1.0, "signal", False, disable_debugger_detection) + events = [] + monkeypatch.setattr(pytest_timeout, "is_debugging", lambda: debugging) + monkeypatch.setattr( + pytest_timeout, "dump_stacks", lambda terminal: events.append(None) + ) + + class Plugin: + def pytest_timeout_expired(self, item, settings, exception): + events.append((item, settings, exception)) + return claim + + plugin = Plugin() + pluginmanager = request.config.pluginmanager + pluginmanager.register(plugin) + try: + suppressed = debugging and not disable_debugger_detection + if suppressed or claim is True: + pytest_timeout.timeout_sigalrm(request.node, settings) + else: + with pytest.raises(pytest.fail.Exception) as caught: + pytest_timeout.timeout_sigalrm(request.node, settings) + assert caught.value is events[-1][2] + finally: + pluginmanager.unregister(plugin) + + if suppressed: + assert events == [] + else: + assert len(events) == 2 + assert events[0] is None # Stack diagnostics precede delivery. + item, received_settings, exception = events[1] + assert item is request.node + assert received_settings is settings + assert isinstance(exception, pytest.fail.Exception) + assert str(exception) == PYTEST_FAILURE_MESSAGE % settings.timeout + + +def test_signal_expiry_hook_is_scoped_to_item(pytester): + pytester.makepyfile( + **{ + "scoped/conftest": """ + def pytest_timeout_expired(item, settings, exception): + return True + """, + "scoped/test_scoped": """ + import pytest_timeout + + def test_claimed(request): + settings = pytest_timeout.Settings(1, "signal", False, True) + pytest_timeout.timeout_sigalrm(request.node, settings) + """, + "test_unscoped": """ + import pytest + import pytest_timeout + + def test_unclaimed(request): + settings = pytest_timeout.Settings(1, "signal", False, True) + with pytest.raises(pytest.fail.Exception): + pytest_timeout.timeout_sigalrm(request.node, settings) + """, + } + ) + result = pytester.runpytest_subprocess() + result.assert_outcomes(passed=2) + + def test_session_timeout(pytester): # This is designed to timeout during the first test to ensure # - the first test still runs to completion