diff --git a/README.rst b/README.rst index 14c3286..3ec0d0d 100644 --- a/README.rst +++ b/README.rst @@ -395,6 +395,8 @@ Changelog 2.5.0 ----- +- Preserve capture-fixture output and disable fixture capturing when reporting + thread timeouts. Thanks Arham Begani. - Fix fixture teardowns not timing out after failed tests. Thanks Timofey Peshin. - Add ruff and typos linters, drop black/isort/flake8 linters. Thanks Vladimir Roshchin. - Minimum support Python3.10 and pytest=8.0. Thanks Vladimir Roshchin. diff --git a/pytest_timeout.py b/pytest_timeout.py index 192a16c..811b5df 100644 --- a/pytest_timeout.py +++ b/pytest_timeout.py @@ -15,6 +15,7 @@ import time import traceback from collections import namedtuple +from contextlib import ExitStack import pytest @@ -498,41 +499,45 @@ def timeout_sigalrm(item, settings): def timeout_timer(item, settings): """Dump stack of threads and call os._exit(). - This disables the capturemanager and dumps stdout and stderr. + This disables global and fixture capturing and dumps stdout and stderr. Then the stacks are dumped and os._exit(1) is called. """ if not settings.disable_debugger_detection and is_debugging(): return terminal = item.config.get_terminal_writer() - try: - capman = item.config.pluginmanager.getplugin("capturemanager") - if capman: - capman.suspend_global_capture(item) - stdout, stderr = capman.read_global_capture() - else: - stdout, stderr = None, None - terminal.sep("+", title="Timeout") - caplog = item.config.pluginmanager.getplugin("_capturelog") - if caplog and hasattr(item, "capturelog_handler"): - log = item.capturelog_handler.stream.getvalue() - if log: - terminal.sep("~", title="Captured log") - terminal.write(log) - if stdout: - terminal.sep("~", title="Captured stdout") - terminal.write(stdout) - if stderr: - terminal.sep("~", title="Captured stderr") - terminal.write(stderr) - dump_stacks(terminal) - terminal.sep("+", title="Timeout") - except Exception: - traceback.print_exc() - finally: - terminal.flush() - sys.stdout.flush() - sys.stderr.flush() - os._exit(1) + with ExitStack() as stack: + try: + capman = item.config.pluginmanager.getplugin("capturemanager") + if capman: + # Flush fixture buffers into global capture before reading it. + # No test or fixture teardown will run after os._exit below. + capman.deactivate_fixture() + stack.enter_context(capman.global_and_fixture_disabled()) + stdout, stderr = capman.read_global_capture() + else: + stdout, stderr = None, None + terminal.sep("+", title="Timeout") + caplog = item.config.pluginmanager.getplugin("_capturelog") + if caplog and hasattr(item, "capturelog_handler"): + log = item.capturelog_handler.stream.getvalue() + if log: + terminal.sep("~", title="Captured log") + terminal.write(log) + if stdout: + terminal.sep("~", title="Captured stdout") + terminal.write(stdout) + if stderr: + terminal.sep("~", title="Captured stderr") + terminal.write(stderr) + dump_stacks(terminal) + terminal.sep("+", title="Timeout") + except Exception: + traceback.print_exc() + finally: + terminal.flush() + sys.stdout.flush() + sys.stderr.flush() + os._exit(1) def dump_stacks(terminal): diff --git a/test_pytest_timeout.py b/test_pytest_timeout.py index 178cb3e..ef01b62 100644 --- a/test_pytest_timeout.py +++ b/test_pytest_timeout.py @@ -71,6 +71,54 @@ def test_foo(): assert "++ Timeout ++" in result.stdout.lines[-1] +@pytest.mark.parametrize("capture", ["fd", "no"]) +@pytest.mark.parametrize("fixture", ["capsys", "capfd", "capsysbinary", "capfdbinary"]) +def test_thread_with_capture_fixture(pytester, capture, fixture): + pytester.makepyfile( + f""" + import sys + import time + + def test_foo({fixture}): + print("FIXTURE STDOUT") + print("FIXTURE STDERR", file=sys.stderr) + time.sleep(10) + """ + ) + + result = pytester.runpytest_subprocess( + "--timeout=1", "--timeout-method=thread", f"--capture={capture}", timeout=20 + ) + + assert result.ret == 1 + assert any(line.endswith("FIXTURE STDOUT") for line in result.stdout.lines) + assert any( + line.endswith("FIXTURE STDERR") + for line in result.stdout.lines + result.stderr.lines + ) + result.stdout.fnmatch_lines(["*++ Timeout ++*", "*~~ Stack of MainThread* ~~*"]) + + +def test_thread_without_capture_plugin(pytester): + pytester.makepyfile( + """ + import time + + def test_foo(): + print("UNCAPTURED STDOUT", flush=True) + time.sleep(10) + """ + ) + + result = pytester.runpytest_subprocess( + "--timeout=1", "--timeout-method=thread", "-p", "no:capture", timeout=20 + ) + + assert result.ret == 1 + assert any(line.endswith("UNCAPTURED STDOUT") for line in result.stdout.lines) + result.stdout.fnmatch_lines(["*++ Timeout ++*", "*~~ Stack of MainThread* ~~*"]) + + def test_cov(pytester): # This test requires pytest-cov pytest.importorskip("pytest_cov")