Merge pull request #13728 from bluetech/terminal-tab-progress

terminal: notify terminal emulator about test session progress

Author: bluetech

AUTHORS

@@ -39,6 +39,7 @@ Andrzej Klajnert
 Andrzej Ostrowski
 Andy Freeland
 Anita Hammer
+Anna Tasiopoulou
 Anthon van der Neut
 Anthony Shaw
 Anthony Sottile

changelog/13072.feature.rst

@@ -0,0 +1,8 @@
+Added support for displaying test session progress in the terminal tab using the `OSC 9;4; <https://conemu.github.io/en/AnsiEscapeCodes.html#ConEmu_specific_OSC>`_ ANSI sequence.
+When pytest runs in a supported terminal emulator like ConEmu, Gnome Terminal, Ptyxis, Windows Terminal, Kitty or Ghostty,
+you'll see the progress in the terminal tab or window,
+allowing you to monitor pytest's progress at a glance.
+
+This feature is automatically enabled when running in a TTY. It is implemented as an internal plugin. If needed, it can be disabled as follows:
+- On a user level, using ``-p no:terminalprogress`` on the command line or via an environment variable ``PYTEST_ADDOPTS='-p no:terminalprogress'``.
+- On a project configuration level, using ``addopts = "-p no:terminalprogress"``.

src/_pytest/_io/terminalwriter.py

@@ -161,20 +161,23 @@ def write(self, msg: str, *, flush: bool = False, **markup: bool) -> None:
 
             msg = self.markup(msg, **markup)
 
-            try:
-                self._file.write(msg)
-            except UnicodeEncodeError:
-                # Some environments don't support printing general Unicode
-                # strings, due to misconfiguration or otherwise; in that case,
-                # print the string escaped to ASCII.
-                # When the Unicode situation improves we should consider
-                # letting the error propagate instead of masking it (see #7475
-                # for one brief attempt).
-                msg = msg.encode("unicode-escape").decode("ascii")
-                self._file.write(msg)
-
-            if flush:
-                self.flush()
+            self.write_raw(msg, flush=flush)
+
+    def write_raw(self, msg: str, *, flush: bool = False) -> None:
+        try:
+            self._file.write(msg)
+        except UnicodeEncodeError:
+            # Some environments don't support printing general Unicode
+            # strings, due to misconfiguration or otherwise; in that case,
+            # print the string escaped to ASCII.
+            # When the Unicode situation improves we should consider
+            # letting the error propagate instead of masking it (see #7475
+            # for one brief attempt).
+            msg = msg.encode("unicode-escape").decode("ascii")
+            self._file.write(msg)
+
+        if flush:
+            self.flush()
 
     def line(self, s: str = "", **markup: bool) -> None:
         self.write(s, **markup)

src/_pytest/terminal.py

@@ -295,6 +295,10 @@ def mywriter(tags, args):
 
         config.trace.root.setprocessor("pytest:config", mywriter)
 
+    if reporter.isatty():
+        plugin = TerminalProgressPlugin(reporter)
+        config.pluginmanager.register(plugin, "terminalprogress")
+
 
 def getreportopt(config: Config) -> str:
     reportchars: str = config.option.reportchars
@@ -454,6 +458,14 @@ def showfspath(self, value: bool | None) -> None:
     def showlongtestinfo(self) -> bool:
         return self.config.get_verbosity(Config.VERBOSITY_TEST_CASES) > 0
 
+    @property
+    def reported_progress(self) -> int:
+        """The amount of items reported in the progress so far.
+
+        :meta private:
+        """
+        return len(self._progress_nodeids_reported)
+
     def hasopt(self, char: str) -> bool:
         char = {"xfailed": "x", "skipped": "s"}.get(char, char)
         return char in self.reportchars
@@ -508,6 +520,9 @@ def wrap_write(
     def write(self, content: str, *, flush: bool = False, **markup: bool) -> None:
         self._tw.write(content, flush=flush, **markup)
 
+    def write_raw(self, content: str, *, flush: bool = False) -> None:
+        self._tw.write_raw(content, flush=flush)
+
     def flush(self) -> None:
         self._tw.flush()
 
@@ -681,7 +696,7 @@ def pytest_runtest_logreport(self, report: TestReport) -> None:
     @property
     def _is_last_item(self) -> bool:
         assert self._session is not None
-        return len(self._progress_nodeids_reported) == self._session.testscollected
+        return self.reported_progress == self._session.testscollected
 
     @hookimpl(wrapper=True)
     def pytest_runtestloop(self) -> Generator[None, object, object]:
@@ -691,7 +706,7 @@ def pytest_runtestloop(self) -> Generator[None, object, object]:
         if (
             self.config.get_verbosity(Config.VERBOSITY_TEST_CASES) <= 0
             and self._show_progress_info
-            and self._progress_nodeids_reported
+            and self.reported_progress
         ):
             self._write_progress_information_filling_space()
 
@@ -702,7 +717,7 @@ def _get_progress_information_message(self) -> str:
         collected = self._session.testscollected
         if self._show_progress_info == "count":
             if collected:
-                progress = len(self._progress_nodeids_reported)
+                progress = self.reported_progress
                 counter_format = f"{{:{len(str(collected))}d}}"
                 format_string = f" [{counter_format}/{{}}]"
                 return format_string.format(progress, collected)
@@ -739,7 +754,7 @@ def _get_progress_information_message(self) -> str:
                 )
             return ""
         if collected:
-            return f" [{len(self._progress_nodeids_reported) * 100 // collected:3d}%]"
+            return f" [{self.reported_progress * 100 // collected:3d}%]"
         return " [100%]"
 
     def _write_progress_information_if_past_edge(self) -> None:
@@ -1641,3 +1656,92 @@ def _get_raw_skip_reason(report: TestReport) -> str:
         elif reason == "Skipped":
             reason = ""
         return reason
+
+
+class TerminalProgressPlugin:
+    """Terminal progress reporting plugin using OSC 9;4 ANSI sequences.
+
+    Emits OSC 9;4 sequences to indicate test progress to terminal
+    tabs/windows/etc.
+
+    Not all terminal emulators support this feature.
+
+    Ref: https://conemu.github.io/en/AnsiEscapeCodes.html#ConEmu_specific_OSC
+    """
+
+    def __init__(self, tr: TerminalReporter) -> None:
+        self._tr = tr
+        self._session: Session | None = None
+        self._has_failures = False
+
+    def _emit_progress(
+        self,
+        state: Literal["remove", "normal", "error", "indeterminate", "paused"],
+        progress: int | None = None,
+    ) -> None:
+        """Emit OSC 9;4 sequence for indicating progress to the terminal.
+
+        :param state:
+            Progress state to set.
+        :param progress:
+            Progress value 0-100. Required for "normal", optional for "error"
+            and "paused", otherwise ignored.
+        """
+        assert progress is None or 0 <= progress <= 100
+
+        # OSC 9;4 sequence: ESC ] 9 ; 4 ; state ; progress ST
+        # ST can be ESC \ or BEL. ESC \ seems better supported.
+        match state:
+            case "remove":
+                sequence = "\x1b]9;4;0;\x1b\\"
+            case "normal":
+                assert progress is not None
+                sequence = f"\x1b]9;4;1;{progress}\x1b\\"
+            case "error":
+                if progress is not None:
+                    sequence = f"\x1b]9;4;2;{progress}\x1b\\"
+                else:
+                    sequence = "\x1b]9;4;2;\x1b\\"
+            case "indeterminate":
+                sequence = "\x1b]9;4;3;\x1b\\"
+            case "paused":
+                if progress is not None:
+                    sequence = f"\x1b]9;4;4;{progress}\x1b\\"
+                else:
+                    sequence = "\x1b]9;4;4;\x1b\\"
+
+        self._tr.write_raw(sequence, flush=True)
+
+    @hookimpl
+    def pytest_sessionstart(self, session: Session) -> None:
+        self._session = session
+        # Show indeterminate progress during collection.
+        self._emit_progress("indeterminate")
+
+    @hookimpl
+    def pytest_collection_finish(self) -> None:
+        assert self._session is not None
+        if self._session.testscollected > 0:
+            # Switch from indeterminate to 0% progress.
+            self._emit_progress("normal", 0)
+
+    @hookimpl
+    def pytest_runtest_logreport(self, report: TestReport) -> None:
+        if report.failed:
+            self._has_failures = True
+
+        # Let's consider the "call" phase for progress.
+        if report.when != "call":
+            return
+
+        # Calculate and emit progress.
+        assert self._session is not None
+        collected = self._session.testscollected
+        if collected > 0:
+            reported = self._tr.reported_progress
+            progress = min(reported * 100 // collected, 100)
+            self._emit_progress("error" if self._has_failures else "normal", progress)
+
+    @hookimpl
+    def pytest_sessionfinish(self) -> None:
+        self._emit_progress("remove")

testing/test_terminal.py

@@ -10,7 +10,10 @@
 import textwrap
 from types import SimpleNamespace
 from typing import cast
+from typing import Literal
 from typing import NamedTuple
+from unittest.mock import Mock
+from unittest.mock import patch
 
 import pluggy
 
@@ -30,6 +33,7 @@
 from _pytest.terminal import _get_raw_skip_reason
 from _pytest.terminal import _plugin_nameversions
 from _pytest.terminal import getreportopt
+from _pytest.terminal import TerminalProgressPlugin
 from _pytest.terminal import TerminalReporter
 import pytest
 
@@ -3297,3 +3301,116 @@ def test_x(a):
                 r".*test_foo.py::test_x\[a::b/\] .*FAILED.*",
             ]
         )
+
+
+class TestTerminalProgressPlugin:
+    """Tests for the TerminalProgressPlugin."""
+
+    @pytest.fixture
+    def mock_file(self) -> StringIO:
+        return StringIO()
+
+    @pytest.fixture
+    def mock_tr(self, mock_file: StringIO) -> pytest.TerminalReporter:
+        tr = Mock(spec=pytest.TerminalReporter)
+
+        def write_raw(s: str, *, flush: bool = False) -> None:
+            mock_file.write(s)
+
+        tr.write_raw = write_raw
+        tr._progress_nodeids_reported = set()
+        return tr
+
+    def test_plugin_registration(self, pytester: pytest.Pytester) -> None:
+        """Test that the plugin is registered correctly on TTY output."""
+        # The plugin module should be registered as a default plugin.
+        with patch.object(sys.stdout, "isatty", return_value=True):
+            config = pytester.parseconfigure()
+            plugin = config.pluginmanager.get_plugin("terminalprogress")
+            assert plugin is not None
+
+    def test_disabled_for_non_tty(self, pytester: pytest.Pytester) -> None:
+        """Test that plugin is disabled for non-TTY output."""
+        with patch.object(sys.stdout, "isatty", return_value=False):
+            config = pytester.parseconfigure()
+            plugin = config.pluginmanager.get_plugin("terminalprogress")
+            assert plugin is None
+
+    @pytest.mark.parametrize(
+        ["state", "progress", "expected"],
+        [
+            ("indeterminate", None, "\x1b]9;4;3;\x1b\\"),
+            ("normal", 50, "\x1b]9;4;1;50\x1b\\"),
+            ("error", 75, "\x1b]9;4;2;75\x1b\\"),
+            ("paused", None, "\x1b]9;4;4;\x1b\\"),
+            ("paused", 80, "\x1b]9;4;4;80\x1b\\"),
+            ("remove", None, "\x1b]9;4;0;\x1b\\"),
+        ],
+    )
+    def test_emit_progress_sequences(
+        self,
+        mock_file: StringIO,
+        mock_tr: pytest.TerminalReporter,
+        state: Literal["remove", "normal", "error", "indeterminate", "paused"],
+        progress: int | None,
+        expected: str,
+    ) -> None:
+        """Test that progress sequences are emitted correctly."""
+        plugin = TerminalProgressPlugin(mock_tr)
+        plugin._emit_progress(state, progress)
+        assert expected in mock_file.getvalue()
+
+    def test_session_lifecycle(
+        self, mock_file: StringIO, mock_tr: pytest.TerminalReporter
+    ) -> None:
+        """Test progress updates during session lifecycle."""
+        plugin = TerminalProgressPlugin(mock_tr)
+
+        session = Mock(spec=pytest.Session)
+        session.testscollected = 3
+
+        # Session start - should emit indeterminate progress.
+        plugin.pytest_sessionstart(session)
+        assert "\x1b]9;4;3;\x1b\\" in mock_file.getvalue()
+        mock_file.truncate(0)
+        mock_file.seek(0)
+
+        # Collection finish - should emit 0% progress.
+        plugin.pytest_collection_finish()
+        assert "\x1b]9;4;1;0\x1b\\" in mock_file.getvalue()
+        mock_file.truncate(0)
+        mock_file.seek(0)
+
+        # First test - 33% progress.
+        report1 = pytest.TestReport(
+            nodeid="test_1",
+            location=("test.py", 0, "test_1"),
+            when="call",
+            outcome="passed",
+            keywords={},
+            longrepr=None,
+        )
+        mock_tr.reported_progress = 1  # type: ignore[misc]
+        plugin.pytest_runtest_logreport(report1)
+        assert "\x1b]9;4;1;33\x1b\\" in mock_file.getvalue()
+        mock_file.truncate(0)
+        mock_file.seek(0)
+
+        # Second test with failure - 66% in error state.
+        report2 = pytest.TestReport(
+            nodeid="test_2",
+            location=("test.py", 1, "test_2"),
+            when="call",
+            outcome="failed",
+            keywords={},
+            longrepr=None,
+        )
+        mock_tr.reported_progress = 2  # type: ignore[misc]
+        plugin.pytest_runtest_logreport(report2)
+        assert "\x1b]9;4;2;66\x1b\\" in mock_file.getvalue()
+        mock_file.truncate(0)
+        mock_file.seek(0)
+
+        # Session finish - should remove progress.
+        plugin.pytest_sessionfinish()
+        assert "\x1b]9;4;0;\x1b\\" in mock_file.getvalue()