terminal: notify terminal emulator about test session progress

Use OSC 9;4 ANSI sequences terminal progress to notify the terminal
emulator about progress, so it can display it to the user (e.g. on the
terminal tab).

Fix #13072.

Co-Authored-By: Anna Tasiopoulou <[email protected]>

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,6 @@
+Added support for displaying test session progress in the terminal tab using the OSC 9;4; ANSI sequence.
+When pytest runs in a supported terminal emulator like ConEmu or Ptyxis,
+you'll see the progress in the terminal tab,
+allowing you to monitor pytest's progress at a glance without having to switch to the pytest tab.
+
+This feature is automatically enabled when running in a compatible terminal and requires no additional configuration.

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
@@ -1652,3 +1656,91 @@ 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.
+        if state == "remove":
+            sequence = "\x1b]9;4;0;\x1b\\"
+        elif state == "normal":
+            assert progress is not None
+            sequence = f"\x1b]9;4;1;{progress}\x1b\\"
+        elif state == "error":
+            if progress is not None:
+                sequence = f"\x1b]9;4;2;{progress}\x1b\\"
+            else:
+                sequence = "\x1b]9;4;2;\x1b\\"
+        elif state == "indeterminate":
+            sequence = "\x1b]9;4;3;\x1b\\"
+        elif state == "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

@@ -11,6 +11,8 @@
 from types import SimpleNamespace
 from typing import cast
 from typing import NamedTuple
+from unittest.mock import Mock
+from unittest.mock import patch
 
 import pluggy
 
@@ -30,6 +32,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 +3300,130 @@ 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
+
+    def test_emit_progress_sequences(
+        self, mock_file: StringIO, mock_tr: pytest.TerminalReporter
+    ) -> None:
+        """Test that progress sequences are emitted correctly."""
+        plugin = TerminalProgressPlugin(mock_tr)
+
+        plugin._emit_progress("indeterminate")
+        assert "\x1b]9;4;3;\x1b\\" in mock_file.getvalue()
+        mock_file.truncate(0)
+        mock_file.seek(0)
+
+        plugin._emit_progress("normal", 50)
+        assert "\x1b]9;4;1;50\x1b\\" in mock_file.getvalue()
+        mock_file.truncate(0)
+        mock_file.seek(0)
+
+        plugin._emit_progress("error", 75)
+        assert "\x1b]9;4;2;75\x1b\\" in mock_file.getvalue()
+        mock_file.truncate(0)
+        mock_file.seek(0)
+
+        # Paused state without progress.
+        plugin._emit_progress("paused")
+        assert "\x1b]9;4;4;\x1b\\" in mock_file.getvalue()
+        mock_file.truncate(0)
+        mock_file.seek(0)
+
+        # Paused state with progress.
+        plugin._emit_progress("paused", 80)
+        assert "\x1b]9;4;4;80\x1b\\" in mock_file.getvalue()
+        mock_file.truncate(0)
+        mock_file.seek(0)
+
+        plugin._emit_progress("remove")
+        assert "\x1b]9;4;0;\x1b\\" in mock_file.getvalue()
+        mock_file.truncate(0)
+        mock_file.seek(0)
+
+    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()