From b5116e4a4b34a9037e2d62bf4122cef8850a7453 Mon Sep 17 00:00:00 2001 From: gavenren Date: Thu, 22 Oct 2026 12:29:50 -0400 Subject: [PATCH 1/2] fix(workflows): handle undecodable shell output as a step failure Capture subprocess output as bytes and decode it in the calling thread so invalid output returns a failed StepResult on Unix and Windows. Preserve locale decoding and universal newlines, and keep JSON output strict. Add real-process regressions for stdout/stderr decoding, valid Unicode, and workflow continue_on_error behavior. Update two subprocess mocks for binary capture. Assisted-by: OpenAI Codex (model: not exposed by the runtime, autonomous) --- .../workflows/step/shell/__init__.py | 25 ++-- tests/test_workflows.py | 4 +- tests/workflows/test_shell_step.py | 110 ++++++++++++++++++ 3 files changed, 130 insertions(+), 9 deletions(-) create mode 100644 tests/workflows/test_shell_step.py diff --git a/src/specify_cli/workflows/step/shell/__init__.py b/src/specify_cli/workflows/step/shell/__init__.py index 0b614b462f..7553e608cc 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,18 @@ 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} + encoding = locale.getpreferredencoding(False) + for name in ("stdout", "stderr"): + with io.TextIOWrapper(io.BytesIO(getattr(proc, name)), encoding=encoding) as stream: + output[name] = stream.read() if proc.returncode != 0: return StepResult( status=StepStatus.FAILED, @@ -79,7 +83,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, @@ -100,6 +104,13 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: error=f"Shell command timed out after {timeout} seconds.", output={"exit_code": -1, "stdout": "", "stderr": "timeout"}, ) + except UnicodeDecodeError as exc: + error = f"Shell command output could not be decoded: {exc}" + return StepResult( + status=StepStatus.FAILED, + error=error, + output={"exit_code": -1, "stdout": "", "stderr": str(exc)}, + ) except OSError 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..51935a774e --- /dev/null +++ b/tests/workflows/test_shell_step.py @@ -0,0 +1,110 @@ +"""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], ids=["stdout", "stderr"]) +@pytest.mark.parametrize("output_format", [None, "json"], ids=["text", "json"]) +def test_undecodable_output_fails_step(project_dir, utf8_locale, stream, output_format): + stdout = b'{"message": "\xff"}' if stream == 1 else b'{"message": "ok"}' + stderr = b"\xff" if stream == 2 else b"" + run = _python_run( + project_dir, + f"import os\nos.write(1, {stdout!r})\nos.write(2, {stderr!r})\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"] == -1 + assert result.output["stdout"] == "" + assert "utf-8" in result.output["stderr"] + 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 From ea25484ccdb5e27242d9b1956e5f92ee75b2e428 Mon Sep 17 00:00:00 2001 From: gavenren Date: Mon, 26 Oct 2026 21:08:51 -0400 Subject: [PATCH 2/2] fix(workflows): retain shell results after output decode failures Decode stdout and stderr independently, keeping the real process exit code and every valid stream. Name each undecodable stream in the step error. Add regression cases for nonzero exits, both invalid streams, and downstream if/switch recovery paths with continue_on_error. Validation: 14 regression failures before this repair; 53 focused tests pass with the repair. Independent agent review reran all 18 decoding tests. Assisted-by: OpenAI Codex (model: unknown, autonomous) --- .../workflows/step/shell/__init__.py | 23 +++--- tests/workflows/test_shell_step.py | 74 +++++++++++++++++-- 2 files changed, 79 insertions(+), 18 deletions(-) diff --git a/src/specify_cli/workflows/step/shell/__init__.py b/src/specify_cli/workflows/step/shell/__init__.py index 7553e608cc..42684b83e1 100644 --- a/src/specify_cli/workflows/step/shell/__init__.py +++ b/src/specify_cli/workflows/step/shell/__init__.py @@ -66,11 +66,21 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: # 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} + output = {"exit_code": proc.returncode, "stdout": "", "stderr": ""} encoding = locale.getpreferredencoding(False) + decode_errors = [] for name in ("stdout", "stderr"): - with io.TextIOWrapper(io.BytesIO(getattr(proc, name)), encoding=encoding) as stream: - output[name] = stream.read() + 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, @@ -104,13 +114,6 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: error=f"Shell command timed out after {timeout} seconds.", output={"exit_code": -1, "stdout": "", "stderr": "timeout"}, ) - except UnicodeDecodeError as exc: - error = f"Shell command output could not be decoded: {exc}" - return StepResult( - status=StepStatus.FAILED, - error=error, - output={"exit_code": -1, "stdout": "", "stderr": str(exc)}, - ) except OSError as exc: return StepResult( status=StepStatus.FAILED, diff --git a/tests/workflows/test_shell_step.py b/tests/workflows/test_shell_step.py index 51935a774e..5acdc5efd4 100644 --- a/tests/workflows/test_shell_step.py +++ b/tests/workflows/test_shell_step.py @@ -26,14 +26,16 @@ def _python_run(project_dir: Path, body: str) -> str: return f'"{sys.executable}" "{script}"' -@pytest.mark.parametrize("stream", [1, 2], ids=["stdout", "stderr"]) +@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, output_format): - stdout = b'{"message": "\xff"}' if stream == 1 else b'{"message": "ok"}' - stderr = b"\xff" if stream == 2 else b"" +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"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: @@ -44,9 +46,12 @@ def test_undecodable_output_fails_step(project_dir, utf8_locale, stream, output_ assert result.status == StepStatus.FAILED assert "could not be decoded" in result.error assert "utf-8" in result.error - assert result.output["exit_code"] == -1 - assert result.output["stdout"] == "" - assert "utf-8" in result.output["stderr"] + 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 @@ -108,3 +113,56 @@ def test_decoding_failure_respects_continue_on_error(project_dir, utf8_locale, c 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"]