safe decoding of build backend errors (#7781)

Co-authored-by: Randy Döring <30527984+radoering@users.noreply.github.com>

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,