safe decoding of build backend errors

src/poetry/installation/chef.py

@@ -17,6 +17,7 @@
 from poetry.core.utils.helpers import temporary_directory
 from pyproject_hooks import quiet_subprocess_runner  # type: ignore[import]
 
+from poetry.utils._compat import decode
 from poetry.utils.env import ephemeral_environment
 
 
@@ -135,9 +136,9 @@ def _prepare(
                     e.exception.stdout is not None or e.exception.stderr is not None
                 ):
                     message_parts.append(
-                        e.exception.stderr.decode()
+                        decode(e.exception.stderr)
                         if e.exception.stderr is not None
-                        else e.exception.stdout.decode()
+                        else decode(e.exception.stdout)
                     )
 
                 error = ChefBuildError("\n\n".join(message_parts))

src/poetry/utils/cache.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import contextlib
 import dataclasses
 import hashlib
 import json
@@ -15,6 +14,8 @@
 from typing import Generic
 from typing import TypeVar
 
+from poetry.utils._compat import decode
+from poetry.utils._compat import encode
 from poetry.utils.wheel import InvalidWheelName
 from poetry.utils.wheel import Wheel
 
@@ -32,42 +33,6 @@
 logger = logging.getLogger(__name__)
 
 
-def decode(string: bytes, encodings: list[str] | None = None) -> str:
-    """
-    Compatiblity decode function pulled from cachy.
-
-    :param string: The byte string to decode.
-    :param encodings: List of encodings to apply
-    :return: Decoded string
-    """
-    if encodings is None:
-        encodings = ["utf-8", "latin1", "ascii"]
-
-    for encoding in encodings:
-        with contextlib.suppress(UnicodeDecodeError):
-            return string.decode(encoding)
-
-    return string.decode(encodings[0], errors="ignore")
-
-
-def encode(string: str, encodings: list[str] | None = None) -> bytes:
-    """
-    Compatibility encode function from cachy.
-
-    :param string: The string to encode.
-    :param encodings: List of encodings to apply
-    :return: Encoded byte string
-    """
-    if encodings is None:
-        encodings = ["utf-8", "latin1", "ascii"]
-
-    for encoding in encodings:
-        with contextlib.suppress(UnicodeDecodeError):
-            return string.encode(encoding)
-
-    return string.encode(encodings[0], errors="ignore")
-
-
 def _expiration(minutes: int) -> int:
     """
     Calculates the time in seconds since epoch that occurs 'minutes' from now.

tests/installation/test_executor.py

@@ -1240,7 +1240,6 @@ def test_build_backend_errors_are_reported_correctly_if_caused_by_subprocess(
     config: Config,
     pool: RepositoryPool,
     io: BufferedIO,
-    tmp_dir: str,
     mock_file_downloads: None,
     env: MockEnv,
     fixture_dir: FixtureDirGetter,
@@ -1265,11 +1264,7 @@ def test_build_backend_errors_are_reported_correctly_if_caused_by_subprocess(
     # must not be included in the error message
     directory_package.python_versions = ">=3.7"
 
-    return_code = executor.execute(
-        [
-            Install(directory_package),
-        ]
-    )
+    return_code = executor.execute([Install(directory_package)])
 
     assert return_code == 1
 
@@ -1306,6 +1301,47 @@ def test_build_backend_errors_are_reported_correctly_if_caused_by_subprocess(
     assert output.endswith(expected_end)
 
 
+@pytest.mark.parametrize("encoding", ["utf-8", "latin-1"])
+@pytest.mark.parametrize("stderr", [None, "Errör on stderr"])
+def test_build_backend_errors_are_reported_correctly_if_caused_by_subprocess_encoding(
+    encoding: str,
+    stderr: str | None,
+    mocker: MockerFixture,
+    config: Config,
+    pool: RepositoryPool,
+    io: BufferedIO,
+    mock_file_downloads: None,
+    env: MockEnv,
+    fixture_dir: FixtureDirGetter,
+) -> None:
+    """Test that the output of the subprocess is decoded correctly."""
+    stdout = "Errör on stdout"
+    error = BuildBackendException(
+        CalledProcessError(
+            1,
+            ["pip"],
+            output=stdout.encode(encoding),
+            stderr=stderr.encode(encoding) if stderr else None,
+        )
+    )
+    mocker.patch.object(ProjectBuilder, "get_requires_for_build", side_effect=error)
+    io.set_verbosity(Verbosity.NORMAL)
+
+    executor = Executor(env, pool, config, io)
+
+    directory_package = Package(
+        "simple-project",
+        "1.2.3",
+        source_type="directory",
+        source_url=fixture_dir("simple_project").resolve().as_posix(),
+    )
+
+    return_code = executor.execute([Install(directory_package)])
+
+    assert return_code == 1
+    assert (stderr or stdout) in io.fetch_output()
+
+
 def test_build_system_requires_not_available(
     config: Config,
     pool: RepositoryPool,