diff --git a/src/specify_cli/workflows/step/shell/__init__.py b/src/specify_cli/workflows/step/shell/__init__.py index 0b614b462f..42684b83e1 100644 --- a/src/specify_cli/workflows/step/shell/__init__.py +++ b/src/specify_cli/workflows/step/shell/__init__.py @@ -2,7 +2,9 @@ from __future__ import annotations +import io import json +import locale import math import os import subprocess @@ -57,16 +59,28 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: run_cmd, shell=True, capture_output=True, - text=True, cwd=cwd, env=env, timeout=timeout, ) - output = { - "exit_code": proc.returncode, - "stdout": proc.stdout, - "stderr": proc.stderr, - } + # Decode in this thread so malformed output becomes a step + # failure on Windows too, where subprocess uses reader threads. + # Preserve text=True's locale encoding and universal newlines. + output = {"exit_code": proc.returncode, "stdout": "", "stderr": ""} + encoding = locale.getpreferredencoding(False) + decode_errors = [] + for name in ("stdout", "stderr"): + try: + with io.TextIOWrapper(io.BytesIO(getattr(proc, name)), encoding=encoding) as stream: + output[name] = stream.read() + except UnicodeDecodeError as exc: + decode_errors.append(f"Shell command {name} could not be decoded: {exc}") + if decode_errors: + return StepResult( + status=StepStatus.FAILED, + error="; ".join(decode_errors), + output=output, + ) if proc.returncode != 0: return StepResult( status=StepStatus.FAILED, @@ -79,7 +93,7 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: # (e.g. a fan-out's ``items:``). A parse failure fails the # step — declaring ``output_format: json`` is a contract. try: - output["data"] = json.loads(proc.stdout) + output["data"] = json.loads(output["stdout"]) except json.JSONDecodeError as exc: return StepResult( status=StepStatus.FAILED, diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 410faa13d8..86377b37eb 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -2807,7 +2807,7 @@ def test_configured_timeout_is_passed_to_subprocess(self, monkeypatch): def fake_run(*args, **kwargs): captured["timeout"] = kwargs.get("timeout") return subprocess.CompletedProcess( - args=args[0] if args else "", returncode=0, stdout="", stderr="" + args=args[0] if args else "", returncode=0, stdout=b"", stderr=b"" ) monkeypatch.setattr(subprocess, "run", fake_run) @@ -2830,7 +2830,7 @@ def test_default_timeout_preserved_when_omitted(self, monkeypatch): def fake_run(*args, **kwargs): captured["timeout"] = kwargs.get("timeout") return subprocess.CompletedProcess( - args=args[0] if args else "", returncode=0, stdout="", stderr="" + args=args[0] if args else "", returncode=0, stdout=b"", stderr=b"" ) monkeypatch.setattr(subprocess, "run", fake_run) diff --git a/tests/workflows/test_shell_step.py b/tests/workflows/test_shell_step.py new file mode 100644 index 0000000000..5acdc5efd4 --- /dev/null +++ b/tests/workflows/test_shell_step.py @@ -0,0 +1,168 @@ +"""Real-process regression tests for shell step output decoding.""" + +from __future__ import annotations + +import locale +import sys +from pathlib import Path + +import pytest +import yaml + +from specify_cli.workflows.base import RunStatus, StepContext, StepStatus +from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine +from specify_cli.workflows.step.shell import ShellStep + + +@pytest.fixture +def utf8_locale(monkeypatch): + """Use the same subprocess text encoding on Unix and Windows.""" + monkeypatch.setattr(locale, "getencoding", lambda: "utf-8") + + +def _python_run(project_dir: Path, body: str) -> str: + script = project_dir / "emit.py" + script.write_text(body, encoding="utf-8") + return f'"{sys.executable}" "{script}"' + + +@pytest.mark.parametrize("stream", [1, 2, "both"], ids=["stdout", "stderr", "both"]) +@pytest.mark.parametrize("exit_code", [0, 3], ids=["zero-exit", "nonzero-exit"]) +@pytest.mark.parametrize("output_format", [None, "json"], ids=["text", "json"]) +def test_undecodable_output_fails_step(project_dir, utf8_locale, stream, exit_code, output_format): + stdout = b'{"message": "\xff"}' if stream in (1, "both") else b'{"message": "ok"}' + stderr = b"\xff" if stream in (2, "both") else b"diagnostic\r\n" + run = _python_run( + project_dir, + f"import os\nos.write(1, {stdout!r})\nos.write(2, {stderr!r})\n" + f"raise SystemExit({exit_code})\n", + ) + config = {"id": "invalid-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.FAILED + assert "could not be decoded" in result.error + assert "utf-8" in result.error + assert result.output["exit_code"] == exit_code + assert result.output["stdout"] == ("" if stream in (1, "both") else '{"message": "ok"}') + assert result.output["stderr"] == ("" if stream in (2, "both") else "diagnostic\n") + for name, number in (("stdout", 1), ("stderr", 2)): + if stream in (number, "both"): + assert f"{name} could not be decoded" in result.error + assert "data" not in result.output + + +@pytest.mark.parametrize("output_format", [None, "json"], ids=["text", "json"]) +def test_valid_unicode_output_is_preserved(project_dir, utf8_locale, output_format): + stdout = '{\r\n"message": "caf\u00e9"\r}' + stderr = "diagnostic: caf\u00e9\r\nnext line\rlast line\n" + run = _python_run( + project_dir, + f"import os\nos.write(1, {stdout.encode('utf-8')!r})\n" + f"os.write(2, {stderr.encode('utf-8')!r})\n", + ) + config = {"id": "valid-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"] == '{\n"message": "caf\u00e9"\n}' + assert result.output["stderr"] == "diagnostic: caf\u00e9\nnext line\nlast line\n" + assert result.output["exit_code"] == 0 + if output_format == "json": + assert result.output["data"] == {"message": "caf\u00e9"} + else: + assert "data" not in result.output + + +@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") + definition = WorkflowDefinition.from_string(yaml.safe_dump({ + "schema_version": "1.0", + "workflow": { + "id": "shell-output-decoding", + "name": "Shell Output Decoding", + "version": "1.0.0", + }, + "steps": [ + { + "id": "invalid-output", + "type": "shell", + "run": run, + "continue_on_error": continue_on_error, + }, + {"id": "after", "type": "shell", "run": "echo did-run"}, + ], + })) + + state = WorkflowEngine(project_dir).execute(definition) + + assert state.step_results["invalid-output"]["status"] == "failed" + assert "could not be decoded" in state.step_results["invalid-output"]["error"] + if continue_on_error: + assert state.status == RunStatus.COMPLETED + assert state.step_results["after"]["status"] == "completed" + assert state.step_results["after"]["output"]["stdout"].strip() == "did-run" + else: + assert state.status == RunStatus.FAILED + assert "after" not in state.step_results + + +@pytest.mark.parametrize("branch_type", ["if", "switch"]) +def test_decoding_failure_preserves_exit_code_for_branching(project_dir, utf8_locale, branch_type): + run = _python_run( + project_dir, + "import os\nos.write(1, b'usable output')\nos.write(2, bytes([255]))\n" + "raise SystemExit(3)\n", + ) + recovery = [{"id": "recovery", "type": "shell", "run": "echo recovery-ran"}] + fallback = [{"id": "wrong-exit", "type": "shell", "run": "echo wrong-exit"}] + branch = {"id": "route", "type": branch_type} + if branch_type == "if": + branch.update({ + "condition": "{{ steps.invalid-output.output.exit_code == 3 }}", + "then": recovery, + "else": fallback, + }) + else: + branch.update({ + "expression": "{{ steps.invalid-output.output.exit_code }}", + "cases": {"3": recovery}, + "default": fallback, + }) + definition = WorkflowDefinition.from_string(yaml.safe_dump({ + "schema_version": "1.0", + "workflow": { + "id": "shell-decode-branching", + "name": "Shell Decode Branching", + "version": "1.0.0", + }, + "steps": [ + { + "id": "invalid-output", + "type": "shell", + "run": run, + "continue_on_error": True, + }, + branch, + ], + })) + + state = WorkflowEngine(project_dir).execute(definition) + + assert state.status == RunStatus.COMPLETED + assert "recovery" in state.step_results + assert "wrong-exit" not in state.step_results + result = state.step_results["invalid-output"] + assert result["status"] == "failed" + assert result["output"]["exit_code"] == 3 + assert result["output"]["stdout"] == "usable output" + assert result["output"]["stderr"] == "" + assert "stderr could not be decoded" in result["error"]