Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 33 additions & 3 deletions README.rst
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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``
----------------

Expand Down Expand Up @@ -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
-----

Expand Down
26 changes: 19 additions & 7 deletions pytest_timeout.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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
Expand All @@ -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):
Expand Down
82 changes: 82 additions & 0 deletions test_pytest_timeout.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down