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
1 change: 1 addition & 0 deletions changelog/8973.bugfix.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
The terminal reporter now writes to an unbuffered duplicate of stdout created before output capture can start, so output written through it (for example by plugins calling :func:`TerminalReporter.write() <_pytest.terminal.TerminalReporter.write>` from within a test) always reaches the terminal instead of disappearing into the capture buffers.
57 changes: 57 additions & 0 deletions src/_pytest/config/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import importlib
import importlib.metadata
import inspect
import io
import os
import pathlib
import re
Expand Down Expand Up @@ -77,10 +78,45 @@
from _pytest.pathlib import resolve_package_path
from _pytest.pathlib import safe_exists
from _pytest.stash import Stash
from _pytest.stash import StashKey
from _pytest.warning_types import PytestConfigWarning
from _pytest.warning_types import warn_explicit_for


class _CaptureImmuneStdout(io.TextIOWrapper):
"""See :data:`capture_immune_stdout_key`."""

#: The ``sys.stdout`` at creation time. Output that legitimately targets
#: the terminal may sit in its buffers (e.g. prints under ``-s`` or
#: ``capsys.disabled()``, block-buffered when stdout is not a tty);
#: it is pushed out before each of our writes to preserve ordering on
#: the terminal.
_original_stdout: TextIO | None = None

def write(self, s: str) -> int:
for stdout in (self._original_stdout, sys.stdout):
if stdout is not None and stdout is not self:
try:
stdout.flush()
except (AttributeError, OSError, ValueError):
pass
return super().write(s)


capture_immune_stdout_key = StashKey[TextIO]()
"""An unbuffered text file over a duplicate of stdout's file descriptor,
created before output capture can start.

Output capture redirects file descriptor 1 and/or replaces ``sys.stdout``,
so anything written through them while capture is active ends up in the
capture buffers. This file always refers to the original stdout (the same
file capture restores on suspend) and is usable at any time no matter how
capture is started, stopped, or reconfigured — writing to it neither goes
through capture nor affects it. Used by the terminal reporter (#8973).

Owned by the config: closed by its cleanup at teardown."""


if TYPE_CHECKING:
from _pytest.assertion.rewrite import AssertionRewritingHook
from _pytest.cacheprovider import Cache
Expand Down Expand Up @@ -402,6 +438,27 @@ def _prepareconfig(
raise TypeError(msg.format(args, type(args)))

initial_config = get_config(args, plugins, prog=prog)

# Duplicate stdout early, before any output capture can start, so the
# terminal reporter can always write to the real terminal (#8973).
try:
dup_stdout_fd = os.dup(sys.stdout.fileno())
except (AttributeError, OSError, ValueError):
# stdout has no usable file descriptor (e.g. in-process pytester
# runs); the terminal reporter falls back to sys.stdout.
pass
else:
capture_immune_stdout = _CaptureImmuneStdout(
io.FileIO(dup_stdout_fd, mode="wb", closefd=True),
encoding=getattr(sys.stdout, "encoding", None) or "utf-8",
errors=getattr(sys.stdout, "errors", None) or "replace",
# Unbuffered: writes must be visible on the terminal immediately.
write_through=True,
)
capture_immune_stdout._original_stdout = sys.stdout
initial_config.stash[capture_immune_stdout_key] = capture_immune_stdout
initial_config.add_cleanup(capture_immune_stdout.close)

pluginmanager = initial_config.pluginmanager
try:
if plugins:
Expand Down
7 changes: 6 additions & 1 deletion src/_pytest/terminal.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
import _pytest._version
from _pytest.compat import running_on_ci
from _pytest.config import _PluggyPlugin
from _pytest.config import capture_immune_stdout_key
from _pytest.config import Config
from _pytest.config import ExitCode
from _pytest.config import hookimpl
Expand Down Expand Up @@ -289,7 +290,11 @@ def pytest_addoption(parser: Parser) -> None:


def pytest_configure(config: Config) -> None:
reporter = TerminalReporter(config, sys.stdout)
# Write to the capture-immune duplicate of stdout when available (#8973),
# fall back to sys.stdout otherwise (e.g. in-process pytester runs, where
# stdout has no real file descriptor to duplicate).
file = config.stash.get(capture_immune_stdout_key, None)
reporter = TerminalReporter(config, file)
config.pluginmanager.register(reporter, "terminalreporter")
if config.option.debug or config.option.traceconfig:

Expand Down
26 changes: 26 additions & 0 deletions testing/test_terminal.py
Original file line number Diff line number Diff line change
Expand Up @@ -3563,3 +3563,29 @@ def test_session_lifecycle(
# Session finish - should remove progress.
plugin.pytest_sessionfinish()
assert "\x1b]9;4;0;\x1b\\" in mock_file.getvalue()


def test_terminalreporter_write_during_capture_reaches_terminal(
pytester: pytest.Pytester,
) -> None:
"""Output written via the terminal reporter from within a test reaches
the terminal even while output capture is active (#8973).

Runs in a subprocess: in-process runs replace stdout with an object
without a real file descriptor, so they cannot exercise the
capture-immune stdout duplicate.
"""
pytester.makepyfile(
"""
def test_foo(request):
reporter = request.config.pluginmanager.getplugin("terminalreporter")
reporter.ensure_newline()
reporter.write("MAGIC_MARKER", flush=True)
print("PLAIN_PRINT")
"""
)
result = pytester.runpytest_subprocess()
result.assert_outcomes(passed=1)
result.stdout.fnmatch_lines(["*MAGIC_MARKER*"])
# Regular output stays captured (the test passes, so it is never shown).
result.stdout.no_fnmatch_line("*PLAIN_PRINT*")