fix(workflows): decode code-page shell output in Python's UTF-8 mode

src/specify_cli/workflows/step/shell/__init__.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import codecs
 import io
 import json
 import locale
@@ -15,6 +16,37 @@
 from specify_cli.workflows.expressions import evaluate_expression
 
 
+def _read_text(data: bytes, encoding: str) -> str:
+    # TextIOWrapper applies the same universal-newline translation as text=True.
+    with io.TextIOWrapper(io.BytesIO(data), encoding=encoding) as stream:
+        return stream.read()
+
+
+def _decode_output(data: bytes, encoding: str) -> str:
+    """Decode captured output with *encoding*, as ``text=True`` would.
+
+    In Python's UTF-8 mode (``PYTHONUTF8=1``, and the default from Python 3.15)
+    the preferred encoding is always UTF-8, but cmd.exe and other native
+    programs still write in the locale code page, e.g. cp936 on Chinese
+    Windows. If the output is not valid in *encoding*, retry with the locale
+    encoding that ``text=True`` used outside UTF-8 mode.
+    """
+    try:
+        return _read_text(data, encoding)
+    except UnicodeDecodeError as exc:
+        fallback = locale.getencoding()
+        try:
+            same = codecs.lookup(fallback).name == codecs.lookup(encoding).name
+        except LookupError:
+            same = True
+        if same:
+            raise
+        try:
+            return _read_text(data, fallback)
+        except UnicodeDecodeError:
+            raise exc from None
+
+
 class ShellStep(StepBase):
     """Run a local shell command (non-agent).
 
@@ -97,8 +129,7 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
             decode_errors = []
             for name in ("stdout", "stderr"):
                 try:
-                    with io.TextIOWrapper(io.BytesIO(getattr(proc, name)), encoding=encoding) as stream:
-                        output[name] = stream.read()
+                    output[name] = _decode_output(getattr(proc, name), encoding)
                 except UnicodeDecodeError as exc:
                     decode_errors.append(f"Shell command {name} could not be decoded: {exc}")
             if decode_errors:

tests/workflows/test_shell_step.py

@@ -472,6 +472,50 @@ def test_valid_unicode_output_is_preserved(project_dir, utf8_locale, output_form
         assert "data" not in result.output
 
 
+@pytest.fixture
+def utf8_mode_code_page_locale(monkeypatch):
+    """Python's UTF-8 mode on a system whose code page is cp936."""
+    monkeypatch.setattr(locale, "getpreferredencoding", lambda do_setlocale=True: "utf-8")
+    monkeypatch.setattr(locale, "getencoding", lambda: "cp936")
+
+
+@pytest.mark.parametrize("output_format", [None, "json"], ids=["text", "json"])
+def test_code_page_output_is_decoded_in_utf8_mode(project_dir, utf8_mode_code_page_locale, output_format):
+    # cmd.exe and other native programs write in the code page even when
+    # Python runs in UTF-8 mode (the default from Python 3.15).
+    stdout = '{"message": "\u5b8c\u6210"}\r\n'
+    stderr = "\u8b66\u544a\r\n"
+    run = _python_run(
+        project_dir,
+        f"import os\nos.write(1, {stdout.encode('cp936')!r})\n"
+        f"os.write(2, {stderr.encode('cp936')!r})\n",
+    )
+    config = {"id": "code-page-output", "type": "shell", "run": run}
+    if output_format is not None:
+        config["output_format"] = output_format
+
+    result = ShellStep().execute(config, StepContext(project_root=str(project_dir)))
+
+    assert result.status == StepStatus.COMPLETED
+    assert result.error is None
+    assert result.output["stdout"] == '{"message": "\u5b8c\u6210"}\n'
+    assert result.output["stderr"] == "\u8b66\u544a\n"
+    if output_format == "json":
+        assert result.output["data"] == {"message": "\u5b8c\u6210"}
+
+
+def test_output_invalid_in_code_page_too_fails_step(project_dir, utf8_mode_code_page_locale):
+    run = _python_run(project_dir, "import os\nos.write(1, b'\\xff')\n")
+    config = {"id": "invalid-output", "type": "shell", "run": run}
+
+    result = ShellStep().execute(config, StepContext(project_root=str(project_dir)))
+
+    assert result.status == StepStatus.FAILED
+    assert "stdout could not be decoded" in result.error
+    assert "utf-8" in result.error
+    assert result.output["stdout"] == ""
+
+
 @pytest.mark.parametrize("continue_on_error", [False, True], ids=["halt", "continue"])
 def test_decoding_failure_respects_continue_on_error(project_dir, utf8_locale, continue_on_error):
     run = _python_run(project_dir, "import os\nos.write(1, bytes([255]))\n")