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
2 changes: 2 additions & 0 deletions README.rst
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
65 changes: 35 additions & 30 deletions pytest_timeout.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import time
import traceback
from collections import namedtuple
from contextlib import ExitStack

import pytest

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