implement pseudo tty on stdout/stderr

docs/changelog/1773.bugfix.rst

@@ -0,0 +1,2 @@
+If tox is running in a tty, allocate a pty (pseudo terminal) for commands
+and copy termios attributes to show colors and improve interactive use - by :user:`masenf`.

src/tox/execute/local_sub_process/__init__.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import fnmatch
+import io
 import logging
 import os
 import shutil
@@ -240,12 +241,19 @@ def __exit__(
 
     @staticmethod
     def get_stream_file_no(key: str) -> Generator[int, Popen[bytes], None]:
-        process = yield PIPE
-        stream = getattr(process, key)
-        if sys.platform == "win32":  # explicit check for mypy # pragma: win32 cover
-            yield stream.handle
+        allocated_pty = _pty(key)
+        if allocated_pty is not None:
+            main_fd, child_fd = allocated_pty
+            yield child_fd
+            os.close(child_fd)  # close the child process pipe
+            yield main_fd
         else:
-            yield stream.name
+            process = yield PIPE
+            stream = getattr(process, key)
+            if sys.platform == "win32":  # explicit check for mypy # pragma: win32 cover
+                yield stream.handle
+            else:
+                yield stream.name
 
     def set_out_err(self, out: SyncWrite, err: SyncWrite) -> tuple[SyncWrite, SyncWrite]:
         prev = self._out, self._err
@@ -256,6 +264,56 @@ def set_out_err(self, out: SyncWrite, err: SyncWrite) -> tuple[SyncWrite, SyncWr
         return prev
 
 
+def _pty(key: str) -> tuple[int, int] | None:
+    """
+    Allocate a virtual terminal (pty) for a subprocess.
+
+    A virtual terminal allows a process to perform syscalls that fetch attributes related to the tty,
+    for example to determine whether to use colored output or enter interactive mode.
+
+    The termios attributes of the controlling terminal stream will be copied to the allocated pty.
+
+    :param key: The stream to copy attributes from. Either "stdout" or "stderr".
+    :return: (main_fd, child_fd) of an allocated pty; or None on error or if unsupported (win32).
+    """
+    if sys.platform == "win32":  # explicit check for mypy # pragma: win32 cover
+        return None
+
+    stream: io.TextIOWrapper = getattr(sys, key)
+
+    # when our current stream is a tty, emulate pty for the child
+    #   to allow host streams traits to be inherited
+    if not stream.isatty():
+        return None
+
+    try:
+        import fcntl
+        import pty
+        import struct
+        import termios
+    except ImportError:  # pragma: no cover
+        return None  # cannot proceed on platforms without pty support
+
+    try:
+        main, child = pty.openpty()
+    except OSError:  # could not open a tty
+        return None  # pragma: no cover
+
+    try:
+        mode = termios.tcgetattr(stream)
+        termios.tcsetattr(child, termios.TCSANOW, mode)
+    except (termios.error, OSError):  # could not inherit traits
+        return None  # pragma: no cover
+
+    # adjust sub-process terminal size
+    columns, lines = shutil.get_terminal_size(fallback=(-1, -1))
+    if columns != -1 and lines != -1:
+        size = struct.pack("HHHH", columns, lines, 0, 0)
+        fcntl.ioctl(child, termios.TIOCSWINSZ, size)
+
+    return main, child
+
+
 __all__ = (
     "SIG_INTERRUPT",
     "CREATION_FLAGS",

src/tox/execute/stream.py

@@ -55,12 +55,12 @@ def handler(self, content: bytes) -> None:
             at = content.rfind(b"\n")
             if at != -1:  # pragma: no branch
                 at = len(self._content) - len(content) + at + 1
-        if at != -1:
-            self._cancel()
-            try:
+        self._cancel()
+        try:
+            if at != -1:
                 self._write(at)
-            finally:
-                self._start()
+        finally:
+            self._start()
 
     def _start(self) -> None:
         self.timer = Timer(self.REFRESH_RATE, self._trigger_timer)

tests/execute/local_subprocess/test_local_subprocess.py

@@ -267,6 +267,15 @@ def test_local_subprocess_tty(monkeypatch: MonkeyPatch, mocker: MockerFixture, t
     tty = tty_mode == "on"
     mocker.patch("sys.stdout.isatty", return_value=tty)
     mocker.patch("sys.stderr.isatty", return_value=tty)
+    try:
+        import termios  # noqa: F401
+    except ImportError:
+        exp_tty = False  # platforms without tty support at all
+    else:
+        # to avoid trying (and failing) to copy mode bits
+        exp_tty = tty
+        mocker.patch("termios.tcgetattr")
+        mocker.patch("termios.tcsetattr")
 
     executor = LocalSubProcessExecutor(colored=False)
     cmd: list[str] = [sys.executable, str(Path(__file__).parent / "tty_check.py")]
@@ -281,8 +290,8 @@ def test_local_subprocess_tty(monkeypatch: MonkeyPatch, mocker: MockerFixture, t
     assert outcome
     info = json.loads(outcome.out)
     assert info == {
-        "stdout": False,
-        "stderr": False,
+        "stdout": exp_tty,
+        "stderr": exp_tty,
         "stdin": False,
         "terminal": [100, 100],
     }

whitelist.txt

@@ -108,6 +108,7 @@ nonlocal
 notset
 nox
 objtype
+openpty
 ov
 pathname
 pep517
@@ -119,6 +120,7 @@ posix
 prereleases
 prj
 psutil
+pty
 purelib
 py311
 py38
@@ -156,6 +158,11 @@ statemachine
 string2lines
 stringify
 subparsers
+tcgetattr
+TCSANOW
+tcsetattr
+TIOCSWINSZ
+termios
 termux
 testenv
 tmpdir