Handle encodings properly in SyncWrite, fixes #2422

src/tox/execute/api.py

@@ -119,16 +119,23 @@ def call(self, request: ExecuteRequest, show: bool, out_err: OutErr, env: ToxEnv
         start = time.monotonic()
         try:
             # collector is what forwards the content from the file streams to the standard streams
-            out, err = out_err[0].buffer, out_err[1].buffer
-            out_sync = SyncWrite(out.name, out if show else None)
-            err_sync = SyncWrite(err.name, err if show else None, Fore.RED if self._colored else None)
+            out, err = out_err
+            out_buffer, err_buffer = out.buffer, err.buffer
+            out_sync = SyncWrite(out_buffer.name, out_buffer if show else None, encoding=out.encoding)
+            err_sync = SyncWrite(
+                err_buffer.name,
+                err_buffer if show else None,
+                Fore.RED if self._colored else None,
+                encoding=err.encoding,
+            )
             with out_sync, err_sync:
                 instance = self.build_instance(request, self._option_class(env), out_sync, err_sync)
                 with instance as status:
                     yield status
                 exit_code = status.exit_code
         finally:
             end = time.monotonic()
+
         status.outcome = Outcome(
             request,
             show,

src/tox/execute/stream.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import sys
 from contextlib import contextmanager
 from threading import Event, Lock, Timer
 from types import TracebackType
@@ -17,7 +18,9 @@ class SyncWrite:
 
     REFRESH_RATE = 0.1
 
-    def __init__(self, name: str, target: IO[bytes] | None, color: str | None = None) -> None:
+    def __init__(
+        self, name: str, target: IO[bytes] | None, color: str | None = None, encoding: str | None = None
+    ) -> None:
         self._content = bytearray()
         self._target: IO[bytes] | None = target
         self._target_enabled: bool = target is not None
@@ -27,6 +30,7 @@ def __init__(self, name: str, target: IO[bytes] | None, color: str | None = None
         self._at: int = 0
         self._color: str | None = color
         self.name = name
+        self.encoding = sys.getdefaultencoding() if encoding is None else encoding
 
     def __repr__(self) -> str:
         return f"{self.__class__.__name__}(name={self.name!r}, target={self._target!r}, color={self._color!r})"
@@ -100,7 +104,7 @@ def colored(self) -> Iterator[None]:
     @property
     def text(self) -> str:
         with self._content_lock:
-            return self._content.decode("utf-8")
+            return self._content.decode(self.encoding)
 
     @property
     def content(self) -> bytearray:

tests/execute/test_stream.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+from io import BytesIO
+
+import pytest
 from colorama import Fore
 
 from tox.execute.stream import SyncWrite
@@ -8,3 +11,24 @@
 def test_sync_write_repr() -> None:
     sync_write = SyncWrite(name="a", target=None, color=Fore.RED)
     assert repr(sync_write) == f"SyncWrite(name='a', target=None, color={Fore.RED!r})"
+
+
+@pytest.mark.parametrize("encoding", ("utf-8", "latin-1", "cp1252"))
+def test_sync_write_encoding(encoding) -> None:
+    text = "Hello W\N{LATIN SMALL LETTER O WITH DIAERESIS}rld: "
+    io = BytesIO()
+
+    sync_write = SyncWrite(name="a", target=io, color=Fore.RED, encoding=encoding)
+    sync_write.handler(text.encode(encoding))
+    assert sync_write.text == text
+
+
+@pytest.mark.parametrize("encoding", ("latin-1", "cp1252"))
+def test_sync_invalid_encoding(encoding) -> None:
+    text = "Hello W\N{LATIN SMALL LETTER O WITH DIAERESIS}rld: "
+    io = BytesIO()
+
+    sync_write = SyncWrite(name="a", target=io, color=Fore.RED, encoding="utf-8")
+    sync_write.handler(text.encode(encoding))
+    with pytest.raises(UnicodeDecodeError):
+        assert sync_write.text == text