Add tty replacer

This allows users to force isatty until we fix
https://github.com/tox-dev/tox/issues/1773.

Signed-off-by: Bernát Gábor <bgabor8@bloomberg.net>

5 files changed, 43 insertions, 9 deletions

diff --git a/src/tox/config/loader/ini/replace.py b/src/tox/config/loader/ini/replace.py
index faff1551..3a6bb98f 100644
--- a/src/tox/config/loader/ini/replace.py
+++ b/src/tox/config/loader/ini/replace.py
@@ -3,6 +3,7 @@ Apply value substitution (replacement) on tox strings.
 """
 import os
 import re
+import sys
 from configparser import SectionProxy
 from typing import TYPE_CHECKING, Iterator, List, Optional, Sequence, Tuple, Union
 
@@ -71,6 +72,8 @@ def _replace_match(conf: Config, current_env: Optional[str], loader: "IniLoader"
     of_type, *args = ARGS_GROUP.split(value)
     if of_type == "env":
         replace_value: Optional[str] = replace_env(args)
+    elif of_type == "tty":
+        replace_value = replace_tty(args)
     elif of_type == "posargs":
         replace_value = replace_pos_args(args, conf.pos_args)
     else:
@@ -175,6 +178,14 @@ def replace_env(args: List[str]) -> str:
     return os.environ.get(key, default)
 
 
+def replace_tty(args: List[str]) -> str:
+    if sys.stdout.isatty():
+        result = args[0] if len(args) > 0 else ""
+    else:
+        result = args[1] if len(args) > 1 else ""
+    return result
+
+
 __all__ = (
     "CORE_PREFIX",
     "BASE_TEST_ENV",
diff --git a/src/tox/execute/local_sub_process/__init__.py b/src/tox/execute/local_sub_process/__init__.py
index db52d270..b456267b 100644
--- a/src/tox/execute/local_sub_process/__init__.py
+++ b/src/tox/execute/local_sub_process/__init__.py
@@ -13,7 +13,7 @@ from ..request import ExecuteRequest, StdinSource
 from ..stream import SyncWrite
 from .read_via_thread import WAIT_GENERAL
 
-if sys.platform == "win32":  # check explicilty in this form so mypy understands # pragma: win32 cover
+if sys.platform == "win32":  # explicit check for mypy # pragma: win32 cover
     # needs stdin/stdout handlers backed by overlapped IO
     if TYPE_CHECKING:  # the typeshed libraries don't contain this, so replace it with normal one
         from subprocess import Popen
@@ -76,7 +76,7 @@ class LocalSubprocessExecuteStatus(ExecuteStatus):
                 while proc.poll() is None and (time.monotonic() - start) < WAIT_INTERRUPT:
                     continue
                 if proc.poll() is None:  # pragma: no branch
-                    if IS_WIN:  # pragma: no branch
+                    if sys.platform == "win32":  # explicit check for mypy # pragma: no branch
                         logging.warning("terminate %d from %d", to_pid, host_pid)  # pragma: no cover
                     else:
                         logging.warning(
@@ -88,7 +88,8 @@ class LocalSubprocessExecuteStatus(ExecuteStatus):
                         )
                     proc.terminate()
                     start = time.monotonic()
-                    if not IS_WIN:  # Windows terminate is UNIX kill  # pragma: no branch
+                    if sys.platform != "win32":  # explicit check for mypy # pragma: no branch
+                        # Windows terminate is UNIX kill
                         while proc.poll() is None and (time.monotonic() - start) < WAIT_TERMINATE:
                             continue
                         if proc.poll() is None:  # pragma: no branch
@@ -113,7 +114,7 @@ class LocalSubprocessExecuteStatus(ExecuteStatus):
             return  # pragma: no cover
         bytes_content = content.encode()
         try:
-            if IS_WIN:  # pragma: win32 cover
+            if sys.platform == "win32":  # explicit check for mypy  # pragma: win32 cover
                 # on Windows we have a PipeHandle object here rather than a file stream
                 import _overlapped  # type: ignore[import]
 
@@ -210,7 +211,7 @@ class LocalSubProcessExecuteInstance(ExecuteInstance):
         self._read_stdout = ReadViaThread(stdout.send(process), self.out_handler, name=f"out-{pid}", drain=drain)
         self._read_stdout.__enter__()
 
-        if IS_WIN:  # pragma: win32 cover
+        if sys.platform == "win32":  # explicit check for mypy:  # pragma: win32 cover
             process.stderr.read = self._read_stderr._drain_stream  # type: ignore[assignment,union-attr]
             process.stdout.read = self._read_stdout._drain_stream  # type: ignore[assignment,union-attr]
         return status
@@ -227,7 +228,10 @@ class LocalSubProcessExecuteInstance(ExecuteInstance):
     def get_stream_file_no(key: str) -> Generator[int, "Popen[bytes]", None]:
         process = yield PIPE
         stream = getattr(process, key)
-        yield stream.handle if IS_WIN else stream.name
+        if sys.platform == "win32":  # explicit check for mypy
+            yield stream.handle
+        else:
+            yield stream.name
 
     def set_out_err(self, out: SyncWrite, err: SyncWrite) -> Tuple[SyncWrite, SyncWrite]:
         prev = self._out, self._err
diff --git a/src/tox/execute/local_sub_process/read_via_thread.py b/src/tox/execute/local_sub_process/read_via_thread.py
index 14ee1a33..1d558e0d 100644
--- a/src/tox/execute/local_sub_process/read_via_thread.py
+++ b/src/tox/execute/local_sub_process/read_via_thread.py
@@ -1,7 +1,6 @@
 """
 A reader that drain a stream via its file no on a background thread.
 """
-import os
 from abc import ABC, abstractmethod
 from threading import Event, Thread
 from types import TracebackType
@@ -29,7 +28,6 @@ class ReadViaThread(ABC):
         while self.thread.is_alive():  # wait until it stops
             self.thread.join(WAIT_GENERAL)
         self._drain_stream()  # read anything left
-        os.close(self.file_no)
 
     @abstractmethod
     def _read_stream(self) -> None:
diff --git a/tests/config/loader/ini/replace/test_replace_tty.py b/tests/config/loader/ini/replace/test_replace_tty.py
new file mode 100644
index 00000000..95718bd6
--- /dev/null
+++ b/tests/config/loader/ini/replace/test_replace_tty.py
@@ -0,0 +1,20 @@
+import sys
+
+import pytest
+from pytest_mock import MockFixture
+
+from tests.config.loader.ini.replace.conftest import ReplaceOne
+
+
+@pytest.mark.parametrize("is_atty", [True, False])
+def test_replace_env_set(replace_one: ReplaceOne, mocker: MockFixture, is_atty: bool) -> None:
+    mocker.patch.object(sys.stdout, "isatty", return_value=is_atty)
+
+    result = replace_one("1 {tty} 2")
+    assert result == "1  2"
+
+    result = replace_one("1 {tty:a} 2")
+    assert result == f"1 {'a' if is_atty else ''} 2"
+
+    result = replace_one("1 {tty:a:b} 2")
+    assert result == f"1 {'a' if is_atty else 'b'} 2"
diff --git a/tox.ini b/tox.ini
index 3fc37d89..f4a17ed4 100644
--- a/tox.ini
+++ b/tox.ini
@@ -26,7 +26,7 @@ setenv =
 extras =
     testing
 commands =
-    pytest {posargs: \
+    pytest {tty:--color=yes} {posargs: \
       --junitxml {toxworkdir}/junit.{envname}.xml --cov {envsitepackagesdir}/tox --cov {toxinidir}/tests \
       --cov-config=setup.cfg --no-cov-on-fail --cov-report term-missing:skip-covered --cov-context=test \
       --cov-report html:{envtmpdir}/htmlcov \
@@ -49,6 +49,7 @@ commands =
 
 [testenv:type]
 description = run type check on code base
+setenv = {tty:MYPY_FORCE_COLOR = 1}
 deps =
     mypy==0.790
 commands =