Fix terminal reporter output not appearing with capture active

changelog/8973.bugfix.rst

@@ -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 :meth:`TerminalReporter.write() <pytest.TerminalReporter.write>` from within a test) always reaches the terminal instead of disappearing into the capture buffers.

src/_pytest/capture.py

@@ -41,6 +41,7 @@
 from _pytest.nodes import File
 from _pytest.nodes import Item
 from _pytest.reports import CollectReport
+from _pytest.stash import StashKey
 
 
 _CaptureMethod = Literal["fd", "sys", "no", "tee-sys"]
@@ -152,13 +153,102 @@ def _reopen_stdio(f, mode):
     sys.stderr = _reopen_stdio(sys.stderr, "wb")
 
 
+@final
+class TerminalStdout(io.TextIOWrapper):
+    """An unbuffered text file over a duplicate of stdout's file descriptor.
+
+    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 is a duplicate made before capture can start: it
+    always refers to the original stdout -- the same file capture restores on
+    suspend -- and stays usable no matter how capture is started, stopped or
+    reconfigured. Writing to it neither goes through capture nor affects its
+    state (#8973).
+
+    Owned by the config: closed by its cleanup at teardown.
+    """
+
+    def __init__(self, fd: int, *, original_stdout: TextIO) -> None:
+        super().__init__(
+            # ``open`` rather than ``io.FileIO``: only ``open`` picks
+            # ``_WindowsConsoleIO`` for a console descriptor, which writes
+            # through ``WriteConsoleW`` instead of handing UTF-8 bytes to a
+            # console whose code page is usually not UTF-8. This mirrors
+            # ``_reopen_stdio`` above, which duplicates stdout the same way
+            # and for the same reason. Buffered rather than raw: a raw stream
+            # may write fewer bytes than asked for, and ``TextIOWrapper`` does
+            # not retry those.
+            open(fd, "wb", closefd=True),
+            encoding=getattr(original_stdout, "encoding", None) or "utf-8",
+            errors=getattr(original_stdout, "errors", None) or "replace",
+            write_through=True,
+        )
+        self._original_stdout = original_stdout
+
+    def write(self, s: str) -> int:
+        # Output that legitimately targets the terminal may still be sitting in
+        # the buffers of the stream we duplicated (prints under ``-s`` or
+        # ``capsys.disabled()`` are block buffered when stdout is not a tty).
+        # Push it out first, so that the terminal keeps the order in which the
+        # writes were made. While capture is active this merely moves pending
+        # test output into the capture buffers, where it belongs.
+        streams = [self._original_stdout]
+        if sys.stdout is not self._original_stdout:
+            streams.append(sys.stdout)
+        for stream in streams:
+            if stream is not None and stream is not self:
+                try:
+                    stream.flush()
+                except (AttributeError, OSError, ValueError):
+                    pass
+        written = super().write(s)
+        # ``write_through`` only hands the text to the buffer; flush so the
+        # write is visible on the terminal immediately.
+        self.flush()
+        return written
+
+
+terminal_stdout_key = StashKey[TerminalStdout]()
+
+
+def get_terminal_stdout(config: Config) -> TextIO:
+    """Return the file terminal output should be written to.
+
+    Always usable. Falls back to ``sys.stdout`` when stdout could not be
+    duplicated (in-process pytester runs), or when the capture plugin is
+    disabled (``-p no:capture``) and ``sys.stdout`` is the terminal anyway.
+    """
+    terminal_stdout = config.stash.get(terminal_stdout_key, None)
+    if terminal_stdout is None:
+        return sys.stdout
+    return terminal_stdout
+
+
 @hookimpl(wrapper=True)
 def pytest_load_initial_conftests(early_config: Config) -> Generator[None]:
     ns = early_config.known_args_namespace
     if ns.capture == "fd":
         _windowsconsoleio_workaround(sys.stdout)
     _colorama_workaround()
     _readline_workaround()
+
+    # Duplicate stdout before any capture starts, so that terminal output can
+    # sidestep it (#8973). Must come after the windows console workaround,
+    # which replaces ``sys.stdout``, and before ``start_global_capturing()``.
+    # Registering the cleanup here -- before the capture manager's -- makes the
+    # LIFO cleanup stack close it last.
+    try:
+        fd = os.dup(sys.stdout.fileno())
+    except (AttributeError, OSError, ValueError):
+        # No usable file descriptor: in-process pytester runs, or a sys.stdout
+        # replaced by something not backed by a file. get_terminal_stdout()
+        # falls back to sys.stdout.
+        pass
+    else:
+        terminal_stdout = TerminalStdout(fd, original_stdout=sys.stdout)
+        early_config.stash[terminal_stdout_key] = terminal_stdout
+        early_config.add_cleanup(terminal_stdout.close)
+
     pluginmanager = early_config.pluginmanager
     capman = CaptureManager(ns.capture)
     pluginmanager.register(capman, "capturemanager")

src/_pytest/terminal.py

@@ -39,6 +39,7 @@
 from _pytest._io import TerminalWriter
 from _pytest._io.wcwidth import wcswidth
 import _pytest._version
+from _pytest.capture import get_terminal_stdout
 from _pytest.compat import running_on_ci
 from _pytest.config import _PluggyPlugin
 from _pytest.config import Config
@@ -297,7 +298,7 @@ def pytest_addoption(parser: Parser) -> None:
 def pytest_configure(config: Config) -> None:
     # Eagerly validate the value; it is only read lazily during reporting.
     config.getini("console_output_style")
-    reporter = TerminalReporter(config, sys.stdout)
+    reporter = TerminalReporter(config)
     config.pluginmanager.register(reporter, "terminalreporter")
     if config.option.debug or config.option.traceconfig:
 
@@ -398,7 +399,13 @@ def __init__(self, config: Config, file: TextIO | None = None) -> None:
         self._known_types: list[str] | None = None
         self.startpath = config.invocation_params.dir
         if file is None:
-            file = sys.stdout
+            # The terminal channel writes past output capture (#8973); it
+            # falls back to sys.stdout when there is none. Resolved from the
+            # default rather than in pytest_configure so that every caller
+            # that leaves `file` unset gets it -- including plugins that
+            # construct a reporter of their own (pytest-sugar does, despite
+            # the @final above).
+            file = get_terminal_stdout(config)
         self._tw = _pytest.config.create_terminal_writer(config, file)
         self._screen_width = self._tw.fullwidth
         self.currentfspath: Path | str | int | None = None
@@ -534,6 +541,18 @@ def wrap_write(
         self._tw.write(wrapped, flush=flush, **markup)
 
     def write(self, content: str, *, flush: bool = False, **markup: bool) -> None:
+        """Write content to the terminal.
+
+        This is the supported way for a plugin to write to the terminal: it
+        reaches the terminal even while output capture is active, and does so
+        without suspending capture (:issue:`8973`).
+
+        :param content: The text to write.
+        :param flush: Whether to flush the stream afterwards.
+        :param markup:
+            Markup to apply to the text, for example ``red=True`` or
+            ``bold=True``. An unknown markup name raises :class:`ValueError`.
+        """
         self._tw.write(content, flush=flush, **markup)
 
     def write_raw(self, content: str, *, flush: bool = False) -> None:

testing/test_capture.py

@@ -1770,3 +1770,79 @@ def pytest_terminal_summary(config):
     match = re.search(r"^value: '(.*)'\r?$", rest, re.MULTILINE)
     assert match is not None
     assert match.group(1) == "hi"
+
+
+class TestTerminalStdout:
+    """The capture-immune duplicate of stdout used for terminal output (#8973)."""
+
+    @pytest.fixture
+    def terminal_stdout(self, tmp_path):
+        """A TerminalStdout duplicated from a real file, plus that file."""
+
+        def make(name: str = "out.txt"):
+            path = tmp_path / name
+            f = path.open("w", encoding="utf-8")
+            terminal_stdout = capture.TerminalStdout(
+                os.dup(f.fileno()), original_stdout=f
+            )
+            return terminal_stdout, f, path
+
+        return make
+
+    def test_writes_are_unbuffered(self, terminal_stdout) -> None:
+        out, f, path = terminal_stdout()
+        with f:
+            try:
+                out.write("hello")
+                assert path.read_text(encoding="utf-8") == "hello"
+            finally:
+                out.close()
+
+    def test_flushes_the_duplicated_stream_first(self, terminal_stdout) -> None:
+        """Output still sitting in the original stream's buffer must reach the
+        terminal before ours, or the two appear out of order."""
+        out, f, path = terminal_stdout()
+        with f:
+            try:
+                f.write("buffered")
+                out.write("direct")
+                assert path.read_text(encoding="utf-8") == "buffereddirect"
+            finally:
+                out.close()
+
+    def test_close_releases_the_duplicated_descriptor(self, terminal_stdout) -> None:
+        out, f, _ = terminal_stdout()
+        with f:
+            fd = out.fileno()
+            out.close()
+            assert out.closed
+            with pytest.raises(OSError):
+                os.fstat(fd)
+            # Closing twice is harmless -- it must not close an unrelated fd
+            # that has meanwhile been handed out the same number.
+            out.close()
+            # The stream we duplicated from is untouched.
+            assert not f.closed
+
+    def test_falls_back_to_sys_stdout_when_absent(self, pytester: Pytester) -> None:
+        config = pytester.parseconfig()
+        # Whether one got created depends on the *outer* run's capture mode, so
+        # drop it explicitly rather than assume. The registered cleanup still
+        # holds it, so nothing leaks.
+        if capture.terminal_stdout_key in config.stash:
+            del config.stash[capture.terminal_stdout_key]
+        assert capture.get_terminal_stdout(config) is sys.stdout
+
+    def test_reporter_writes_without_the_capture_plugin(
+        self, pytester: Pytester
+    ) -> None:
+        """With -p no:capture nothing is duplicated; sys.stdout is the terminal."""
+        pytester.makepyfile("""
+            def test_foo(request):
+                reporter = request.config.pluginmanager.getplugin("terminalreporter")
+                reporter.ensure_newline()
+                reporter.write("NOCAPTURE_MARKER", flush=True)
+        """)
+        result = pytester.runpytest_subprocess("-p", "no:capture")
+        result.assert_outcomes(passed=1)
+        result.stdout.fnmatch_lines(["*NOCAPTURE_MARKER*"])

testing/test_config.py

@@ -1702,30 +1702,46 @@ def cleanup_first():
 
 
 class TestConfigFromdictargs:
-    def test_basic_behavior(self, _sys_snapshot) -> None:
+    @pytest.fixture
+    def fromdictargs(self, request: pytest.FixtureRequest):
+        """``Config.fromdictargs`` with teardown.
+
+        It parses, so ``pytest_load_initial_conftests`` runs and acquires
+        resources (capturing, the terminal channel); nothing unconfigures the
+        config otherwise, and those would leak into unrelated tests.
+        """
+
+        def make(option_dict: dict[str, object], args: list[str]) -> Config:
+            config = Config.fromdictargs(option_dict, args)
+            request.addfinalizer(config._ensure_unconfigure)
+            return config
+
+        return make
+
+    def test_basic_behavior(self, _sys_snapshot, fromdictargs) -> None:
         option_dict = {"verbose": 444, "foo": "bar", "capture": "no"}
         args = ["a", "b"]
 
-        config = Config.fromdictargs(option_dict, args)
+        config = fromdictargs(option_dict, args)
         with pytest.raises(AssertionError):
             config.parse(["should refuse to parse again"])
         assert config.option.verbose == 444
         assert config.option.foo == "bar"
         assert config.option.capture == "no"
         assert config.args == args
 
-    def test_invocation_params_args(self, _sys_snapshot) -> None:
+    def test_invocation_params_args(self, _sys_snapshot, fromdictargs) -> None:
         """Show that fromdictargs can handle args in their "orig" format"""
         option_dict: dict[str, object] = {}
         args = ["-vvvv", "-s", "a", "b"]
 
-        config = Config.fromdictargs(option_dict, args)
+        config = fromdictargs(option_dict, args)
         assert config.args == ["a", "b"]
         assert config.invocation_params.args == tuple(args)
         assert config.option.verbose == 4
         assert config.option.capture == "no"
 
-    def test_inifilename(self, tmp_path: Path) -> None:
+    def test_inifilename(self, tmp_path: Path, fromdictargs) -> None:
         d1 = tmp_path.joinpath("foo")
         d1.mkdir()
         p1 = d1.joinpath("bar.ini")
@@ -1759,7 +1775,7 @@ def test_inifilename(self, tmp_path: Path) -> None:
         )
         with MonkeyPatch.context() as mp:
             mp.chdir(cwd)
-            config = Config.fromdictargs(option_dict, [])
+            config = fromdictargs(option_dict, [])
             inipath = absolutepath(inifilename)
 
         assert config.args == [str(cwd)]

testing/test_terminal.py

@@ -3576,3 +3576,55 @@ 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 take the sys.stdout fallback
+    instead of the terminal channel under test.
+    """
+    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*")
+
+
+def test_terminalreporter_write_keeps_order_with_uncaptured_print(
+    pytester: pytest.Pytester,
+) -> None:
+    """Writes through the terminal reporter interleave correctly with output
+    that legitimately reaches the terminal via sys.stdout (#8973).
+
+    Under ``-s`` with a piped (non-tty) stdout, prints are block buffered, so
+    without flushing them first the reporter's write would overtake them.
+    """
+    pytester.makepyfile(
+        """
+        def test_foo(request):
+            reporter = request.config.pluginmanager.getplugin("terminalreporter")
+            print("FIRST_PRINT")
+            reporter.ensure_newline()
+            reporter.write("SECOND_WRITE\\n", flush=True)
+            print("THIRD_PRINT")
+        """
+    )
+    result = pytester.runpytest_subprocess("-s")
+    result.assert_outcomes(passed=1)
+    result.stdout.fnmatch_lines(
+        ["*FIRST_PRINT*", "*SECOND_WRITE*", "*THIRD_PRINT*"],
+    )