Make the test suite environment-independent, fix UTF-8 redirection, and a UI-call race (#1752)

* Render test output independently of the caller's environment

Rich and cmd2 consult several environment variables when deciding whether to
emit styling, and the suite inherited them. Exporting any one of them made large
numbers of unrelated tests fail depending on who ran the suite: NO_COLOR failed
15 tests, and FORCE_COLOR and TTY_COMPATIBLE 53 each. The failures look like
product regressions, which makes them expensive to diagnose.

Neutralize them for every test. Tests that exercise these variables set them
explicitly, which still works because a test's own monkeypatching runs after the
fixture. A guard test fails if any of them reaches a test again.

* Write redirected and piped output as UTF-8

Command output is rendered by Rich and routinely contains non-ASCII, but
redirection targets and pipes were opened with the locale's encoding. On any
system whose default is not UTF-8 -- a Windows console using a legacy code page,
for instance -- redirecting output raised UnicodeEncodeError, and the user was
left with an empty file and advice to set PYTHONIOENCODING.

Open both with UTF-8 explicitly. Two tests that read redirected output back were
relying on the locale encoding for decoding as well, so they now name it too.

* Fix a race that reported a timeout for a successful UI call

_call_in_ui() polls the pending future with a 0.1s timeout and, on expiry,
re-raises when the future is already done. That branch exists because
concurrent.futures.TimeoutError is TimeoutError on Python 3.11+, so a callback
raising a timeout of its own cannot be told apart by type from the poll expiring.

Re-raising the caught exception conflates the two. When the callback completes
in the window between the poll expiring and the future being inspected, the
caller is told the call timed out even though it succeeded. Ask the future for
its outcome instead: a callback that raised a timeout still propagates it, and
one that produced a value now returns it.

Found while investigating an intermittent failure of
test_command_toolbar_ui_call_propagates_failures, which reproduced once in 60
runs before this change and not once in 120 after. The added regression test
drives the interleaving deterministically rather than relying on timing.

(cherry picked from commit 84bc19e)

* Run the pipe encoding test through this interpreter, not cat

The test piped through `cat`, which cmd.exe does not provide. On a Windows
system without Unix utilities installed it would fail before reaching the
encoding behavior it exists to check -- and Windows is exactly what the UTF-8
redirection fix targets.

Use a sys.executable pass-through instead, matching the pipe tests already in
tests/test_command_toolbar.py. Reverting either the pipe or the redirect
encoding still fails these tests.

Author: tleonhardt

CHANGELOG.md

@@ -1,6 +1,10 @@
 ## 4.3.0 (TBD)
 
 - Bug Fixes
+    - Fixed output redirection and piping failing on systems whose default encoding is not UTF-8,
+      such as a Windows console using a legacy code page. Command output is rendered by Rich and
+      routinely contains non-ASCII, so redirecting it raised `UnicodeEncodeError` and left an empty
+      file behind. Redirection targets and pipes now use UTF-8 explicitly
     - Fixed the right prompt being redrawn beside every accepted command line in the scrollback.
       prompt-toolkit includes it in the final frame of each prompt, which is the frame left on the
       terminal; it is now hidden there, as the bottom toolbar already was, and stays on the live

cmd2/cmd2.py

@@ -3403,9 +3403,11 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
             # Create a pipe with read and write sides
             read_fd, write_fd = os.pipe()
 
-            # Open each side of the pipe
-            subproc_stdin = open(read_fd)  # noqa: SIM115
-            new_stdout: TextIO = cast(TextIO, open(write_fd, "w"))  # noqa: SIM115
+            # Open each side of the pipe. Both ends are given an explicit encoding:
+            # command output is rendered by Rich and routinely contains non-ASCII, which
+            # the locale encoding cannot always represent.
+            subproc_stdin = open(read_fd, encoding="utf-8")  # noqa: SIM115
+            new_stdout: TextIO = cast(TextIO, open(write_fd, "w", encoding="utf-8"))  # noqa: SIM115
 
             # Create pipe process in a separate group to isolate our signals from it. If a Ctrl-C event occurs,
             # our sigint handler will forward it only to the most recent pipe process. This makes sure pipe
@@ -3472,8 +3474,15 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
                 # statement.output can only contain REDIRECTION_APPEND or REDIRECTION_OUTPUT
                 mode = "a" if statement.redirector == constants.REDIRECTION_APPEND else "w"
                 try:
-                    # Use line buffering
-                    new_stdout = cast(TextIO, open(su.strip_quotes(statement.redirect_to), mode=mode, buffering=1))  # noqa: SIM115
+                    # Use line buffering. The encoding is explicit rather than the
+                    # locale's: command output is rendered by Rich and routinely contains
+                    # non-ASCII, so on a non-UTF-8 system -- a default Windows console,
+                    # for instance -- redirection would otherwise fail and leave an empty
+                    # file behind.
+                    new_stdout = cast(
+                        TextIO,
+                        open(su.strip_quotes(statement.redirect_to), mode=mode, buffering=1, encoding="utf-8"),  # noqa: SIM115
+                    )
                 except OSError as ex:
                     raise RedirectionError("Failed to redirect output") from ex
 

cmd2/command_toolbar.py

@@ -376,7 +376,12 @@ def call() -> None:
                 value = result.result(timeout=0.1)
             except FutureTimeoutError:
                 if result.done():
-                    raise
+                    # The callback finished while this poll was expiring, or raised a
+                    # TimeoutError of its own -- indistinguishable here, because
+                    # concurrent.futures.TimeoutError is TimeoutError on Python 3.11+.
+                    # Ask the future for its outcome rather than re-raising this poll's
+                    # timeout, which would report a failure for a call that succeeded.
+                    return result.result()
                 self._check_running()
             else:
                 return value

tests/conftest.py

@@ -86,6 +86,23 @@ def run_cmd(app: cmd2.Cmd, cmd: str) -> tuple[list[str], list[str]]:
     return normalize(out), normalize(err)
 
 
+#: Environment variables that change how Rich and cmd2 render output. Left inherited,
+#: they make unrelated tests fail depending on who runs the suite: NO_COLOR fails 15
+#: tests, FORCE_COLOR and TTY_COMPATIBLE 53 each.
+COLOR_ENVIRONMENT = ("NO_COLOR", "FORCE_COLOR", "TTY_COMPATIBLE", "TTY_INTERACTIVE")
+
+
+@pytest.fixture(autouse=True)
+def neutral_color_environment(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Render output the same way regardless of the caller's environment.
+
+    Tests that exercise these variables set them explicitly, which still works because
+    a test's own monkeypatching runs after this fixture.
+    """
+    for name in COLOR_ENVIRONMENT:
+        monkeypatch.delenv(name, raising=False)
+
+
 @pytest.fixture
 def base_app() -> cmd2.Cmd:
     return cmd2.Cmd(include_py=True, include_ipy=True)

tests/test_command_toolbar.py

@@ -3,7 +3,8 @@
 import sys
 import threading
 import time
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import Future, ThreadPoolExecutor
+from concurrent.futures import TimeoutError as FutureTimeoutError
 from types import SimpleNamespace
 from unittest import mock
 
@@ -58,7 +59,7 @@ def test_command_toolbar_redirected_output(toolbar_app, tmp_path) -> None:
     destination = tmp_path / "help.txt"
     with app._command_toolbar_context():
         app.onecmd_plus_hooks(f'help > "{destination}"')
-    text = destination.read_text()
+    text = destination.read_text(encoding="utf-8")
     assert "Cmd2 Commands" in text
     assert "STATUS" not in text
     assert "Cmd2 Commands" not in output.getvalue()
@@ -79,7 +80,7 @@ def command(statement, **kwargs):
         app.onecmd_plus_hooks(f'custom > "{destination}"')
         app.poutput("terminal output")
 
-    assert destination.read_text() == "before\nduring\nafter\n"
+    assert destination.read_text(encoding="utf-8") == "before\nduring\nafter\n"
     assert "before" not in output.getvalue()
     assert "during" not in output.getvalue()
     assert "after" not in output.getvalue()
@@ -134,7 +135,7 @@ def command(statement, **kwargs):
     assert running == [False]
     # A process given the terminal writes to it directly instead of through a captured pipe.
     assert readers[0]._proc.stdout is None
-    assert "PIPED" in destination.read_text()
+    assert "PIPED" in destination.read_text(encoding="utf-8")
 
 
 def test_command_toolbar_binary_output(toolbar_app) -> None:
@@ -431,6 +432,36 @@ def fail(exception: BaseException) -> None:
         assert toolbar._call_in_ui(lambda: time.sleep(0.2) or "finished") == "finished"
 
 
+def test_command_toolbar_ui_call_returns_a_result_that_lands_during_the_poll(toolbar_app, monkeypatch) -> None:
+    """A callback finishing while the poll expires must return its value, not a timeout.
+
+    `concurrent.futures.TimeoutError` is `TimeoutError` on Python 3.11+, so the poll
+    expiring and the callback raising a timeout of its own are indistinguishable by type.
+    Re-raising the caught exception once the future is done therefore reports a timeout for
+    a call that actually succeeded.
+    """
+    app, _, _ = toolbar_app
+
+    class RacyFuture(Future):
+        """Completes, and only then reports the poll as having expired."""
+
+        def __init__(self) -> None:
+            super().__init__()
+            self._polled = False
+
+        def result(self, timeout=None):  # type: ignore[no-untyped-def]
+            if timeout is not None and not self._polled:
+                self._polled = True
+                super().result(timeout=5)  # let the callback finish first
+                raise FutureTimeoutError  # then act as though the poll had expired
+            return super().result(timeout)
+
+    monkeypatch.setattr(command_toolbar, "Future", RacyFuture)
+    with app._command_toolbar_context():
+        toolbar = app._command_toolbar
+        assert toolbar._call_in_ui(lambda: "finished") == "finished"
+
+
 def test_command_toolbar_ui_call_after_display_stopped(toolbar_app) -> None:
     app, pipe, _ = toolbar_app
     with app._command_toolbar_context():
@@ -773,5 +804,5 @@ def test_builtin_pager_does_not_capture_redirected_output(toolbar_app, monkeypat
     with mock.patch("cmd2.command_toolbar.Pager") as pager, app._command_toolbar_context():
         app.onecmd_plus_hooks(f'help > "{target}"')
         pager.assert_not_called()
-    assert "Cmd2 Commands" in target.read_text()
+    assert "Cmd2 Commands" in target.read_text(encoding="utf-8")
     assert "Cmd2 Commands" not in output.getvalue()

tests/test_run_pyscript.py

@@ -246,7 +246,7 @@ def test_run_pyscript_print_redirection(base_app, request, tmp_path, capsys) ->
     out, err = capsys.readouterr()
 
     # Verify the output file contains what we expect from print()
-    content = pathlib.Path(out_file).read_text()
+    content = pathlib.Path(out_file).read_text(encoding="utf-8")
 
     # Look for everything written to self.stdout
     assert len(content.splitlines()) == 4

tests/test_suite_environment.py

@@ -0,0 +1,59 @@
+"""Guards that the suite renders output independently of the developer's environment.
+
+Rich and cmd2 both consult environment variables when deciding whether to emit styling.
+Inheriting them makes large numbers of unrelated tests fail depending on who runs them,
+which is expensive to diagnose because the failures look like product regressions.
+"""
+
+import os
+import sys
+
+import pytest
+
+import cmd2
+
+#: Variables that change how output is rendered. Rich reads all of these; cmd2 reads
+#: NO_COLOR directly. Tests that exercise them set them explicitly instead.
+COLOR_ENVIRONMENT = ("NO_COLOR", "FORCE_COLOR", "TTY_COMPATIBLE", "TTY_INTERACTIVE")
+
+
+@pytest.mark.parametrize("name", COLOR_ENVIRONMENT)
+def test_color_environment_does_not_leak_into_tests(name: str) -> None:
+    """A developer exporting any of these must not change the suite's results."""
+    assert name not in os.environ, (
+        f"{name} leaked into the test environment; output-rendering assertions would depend on who is running the suite"
+    )
+
+
+class EncodingProbe(cmd2.Cmd):
+    """Reports the encoding of whatever stream output is currently going to."""
+
+    def do_show_encoding(self, _: str) -> None:
+        """Print the current output stream's encoding."""
+        self.poutput(f"ENCODING={getattr(self.stdout, 'encoding', None)}")
+
+
+def test_redirection_to_a_file_uses_utf8(tmp_path) -> None:
+    """cmd2 renders non-ASCII, so a redirect target must not use the locale encoding.
+
+    Opened with the locale encoding, redirecting styled output raises UnicodeEncodeError
+    on any non-UTF-8 system -- which includes a default Windows console -- leaving the
+    user an empty file and an error.
+    """
+    app = EncodingProbe(allow_cli_args=False)
+    target = tmp_path / "out.txt"
+    app.onecmd_plus_hooks(f'show_encoding > "{target}"')
+    assert "ENCODING=utf-8" in target.read_text(encoding="utf-8")
+
+
+#: A pass-through filter, run with this interpreter so the test does not depend on Unix
+#: utilities being installed. `cmd.exe` has no `cat`, and this fix exists for Windows.
+PASS_THROUGH = "import sys; sys.stdin.reconfigure(encoding='utf-8'); sys.stdout.write(sys.stdin.read())"
+
+
+def test_piping_uses_utf8(tmp_path) -> None:
+    """The same applies to the pipe the subprocess reads from."""
+    app = EncodingProbe(allow_cli_args=False)
+    target = tmp_path / "piped.txt"
+    app.onecmd_plus_hooks(f'show_encoding | "{sys.executable}" -c "{PASS_THROUGH}" > "{target}"')
+    assert "ENCODING=utf-8" in target.read_text(encoding="utf-8")