fix(cli): write CLI output as UTF-8 under legacy code pages

When stdout or stderr was opened with a non-UTF-8 encoding, such as the
Windows cp1252 code page used for pipes and redirects, sqb crashed with
E001 on the first non-ASCII character of its summary output.

The CLI entry point now reconfigures any non-UTF-8 standard stream to
UTF-8 with backslashreplace, so output never fails to encode and JSON
output stays valid. Library use and the rule-engine host process are
unaffected.

Author: kvlonge

src/sqlbuild/cli/entry/_helpers/__init__.py

@@ -0,0 +1,24 @@
+"""Standard output streams the CLI can always write, whatever the console code page."""
+
+import codecs
+import io
+import sys
+
+from sqlbuild.cli.entry.constants import OUTPUT_STREAM_ENCODING, OUTPUT_STREAM_ERRORS
+
+
+def use_utf8_output_streams() -> tuple[bool, ...]:
+    """Write stdout and stderr as UTF-8 when opened with another encoding; return which changed."""
+
+    return tuple(use_utf8_stream(stream) for stream in (sys.stdout, sys.stderr))
+
+
+def use_utf8_stream(stream: object) -> bool:
+    """Reconfigure a text stream that is not UTF-8 to UTF-8; return whether it changed."""
+
+    if not isinstance(stream, io.TextIOWrapper):
+        return False
+    if codecs.lookup(stream.encoding).name == codecs.lookup(OUTPUT_STREAM_ENCODING).name:
+        return False
+    stream.reconfigure(encoding=OUTPUT_STREAM_ENCODING, errors=OUTPUT_STREAM_ERRORS)
+    return True

src/sqlbuild/cli/entry/_helpers/streams.py

@@ -2,3 +2,5 @@
 
 NATIVE_ALLOCATOR_PURGE_DELAY_VARIABLE: str = "MIMALLOC_PURGE_DELAY"
 NATIVE_ALLOCATOR_PURGE_DELAY_MILLISECONDS: str = "10"
+OUTPUT_STREAM_ENCODING: str = "utf-8"
+OUTPUT_STREAM_ERRORS: str = "backslashreplace"

src/sqlbuild/cli/entry/constants.py

@@ -3,6 +3,7 @@
 import os
 from collections.abc import Sequence
 
+from sqlbuild.cli.entry._helpers.streams import use_utf8_output_streams
 from sqlbuild.cli.entry.constants import (
     NATIVE_ALLOCATOR_PURGE_DELAY_MILLISECONDS,
     NATIVE_ALLOCATOR_PURGE_DELAY_VARIABLE,
@@ -15,6 +16,7 @@ def main(argv: Sequence[str] | None = None) -> int:
     os.environ.setdefault(
         NATIVE_ALLOCATOR_PURGE_DELAY_VARIABLE, NATIVE_ALLOCATOR_PURGE_DELAY_MILLISECONDS
     )
+    _ = use_utf8_output_streams()
     from sqlbuild.cli.commands.main.entrypoint.entry import main as run_cli
 
     return run_cli(argv)

src/sqlbuild/cli/entry/main/entry.py

@@ -23,3 +23,11 @@ class UnicodeEmptyFixtureTestCase:
     description: str
     status: str
     expected_added_findings: int = 0
+
+
+@dataclass(frozen=True)
+class LegacyCodePageOutputTestCase:
+    description: str
+    code_page: str
+    expected_summary: str
+    expected_json_command: str

tests/e2e/src/sqlbuild/cli/commands/main/playground/_test_types.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import json
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from sqlbuild.cli.commands.main.workspace._playground import run_playground
+from sqlbuild.cli.commands.models import PlaygroundCommandRequest
+from tests.e2e.src.sqlbuild.cli.commands.main.playground._test_types import (
+    LegacyCodePageOutputTestCase,
+)
+from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import run_sqb
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        LegacyCodePageOutputTestCase(
+            description="cp1252 streams still get the UTF-8 summary and valid JSON",
+            code_page="cp1252",
+            expected_summary="\u2713 Project compiled",
+            expected_json_command="compile",
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_legacy_code_page_streams_when_compiling_then_output_is_utf8(
+    test_case: LegacyCodePageOutputTestCase, tmp_path: Path
+) -> None:
+    assert run_playground(PlaygroundCommandRequest(project_dir=tmp_path, target_path="shop")) == 0
+    environment: dict[str, str] = {"PYTHONIOENCODING": test_case.code_page}
+
+    summary: subprocess.CompletedProcess[str] = run_sqb(
+        command=("--no-color", "compile"), project_dir=tmp_path / "shop", env=environment
+    )
+    machine: subprocess.CompletedProcess[str] = run_sqb(
+        command=("compile", "--json"), project_dir=tmp_path / "shop", env=environment
+    )
+
+    assert (
+        summary.returncode,
+        test_case.expected_summary in summary.stdout,
+        machine.returncode,
+        json.loads(machine.stdout).get("command"),
+    ) == (0, True, 0, test_case.expected_json_command), summary.stderr + machine.stderr
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-vv"])

tests/e2e/src/sqlbuild/cli/commands/main/playground/test_output_encoding.py

@@ -0,0 +1,11 @@
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class OutputStreamEncodingTestCase:
+    description: str
+    encoding: str
+    errors: str
+    text: str
+    expected_changed: bool
+    expected_bytes: bytes

tests/unit/src/sqlbuild/cli/entry/__init__.py

@@ -0,0 +1,55 @@
+import io
+
+import pytest
+
+from sqlbuild.cli.entry._helpers.streams import use_utf8_stream
+from tests.unit.src.sqlbuild.cli.entry._helpers._test_types import OutputStreamEncodingTestCase
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        OutputStreamEncodingTestCase(
+            description="a cp1252 stream writes box drawing and symbols as UTF-8",
+            encoding="cp1252",
+            errors="strict",
+            text="\u2713 Project compiled \u2500\u2500 orders \u2192 customers\n",
+            expected_changed=True,
+            expected_bytes="\u2713 Project compiled \u2500\u2500 orders \u2192 customers\n".encode(),
+        ),
+        OutputStreamEncodingTestCase(
+            description="a lone surrogate is escaped instead of crashing",
+            encoding="cp1252",
+            errors="strict",
+            text="orders\ud800.sql\n",
+            expected_changed=True,
+            expected_bytes=b"orders\\ud800.sql\n",
+        ),
+        OutputStreamEncodingTestCase(
+            description="a UTF-8 stream under another spelling is left alone",
+            encoding="UTF8",
+            errors="backslashreplace",
+            text="\u2713 orders\n",
+            expected_changed=False,
+            expected_bytes="\u2713 orders\n".encode(),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_stream_encoding_when_preparing_cli_output_then_any_text_is_written(
+    test_case: OutputStreamEncodingTestCase,
+) -> None:
+    buffer: io.BytesIO = io.BytesIO()
+    stream: io.TextIOWrapper = io.TextIOWrapper(
+        buffer, encoding=test_case.encoding, errors=test_case.errors
+    )
+
+    changed: bool = use_utf8_stream(stream)
+    _ = stream.write(test_case.text)
+    stream.flush()
+
+    assert (changed, buffer.getvalue()) == (test_case.expected_changed, test_case.expected_bytes)
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-vv"])