From bf2f25cee766f652cf24b2ff8a1828b20714ecf7 Mon Sep 17 00:00:00 2001 From: doquanghuy Date: Tue, 2 Jun 2026 17:10:30 +0700 Subject: [PATCH 1/9] fix(workflows): render gate show_file contents in the interactive prompt The gate step read and recorded `show_file` but never displayed its contents at the interactive prompt, so the operator approved/rejected without seeing the referenced file. Render the file inside the prompt when stdin is a TTY, with a graceful notice for missing/unreadable files. Non-interactive PAUSED behaviour, exit codes, resume semantics, and no-`show_file` output are unchanged. Closes #2809. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../workflows/steps/gate/__init__.py | 57 ++++++++- tests/test_workflows.py | 109 ++++++++++++++++++ 2 files changed, 162 insertions(+), 4 deletions(-) diff --git a/src/specify_cli/workflows/steps/gate/__init__.py b/src/specify_cli/workflows/steps/gate/__init__.py index d4d32d763c..466f4e300f 100644 --- a/src/specify_cli/workflows/steps/gate/__init__.py +++ b/src/specify_cli/workflows/steps/gate/__init__.py @@ -3,6 +3,7 @@ from __future__ import annotations import sys +from pathlib import Path from typing import Any from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus @@ -23,6 +24,10 @@ class GateStep(StepBase): type_key = "gate" + #: Maximum number of ``show_file`` lines rendered at the prompt, so a + #: large file cannot flood the terminal before the choice. + MAX_SHOW_FILE_LINES = 200 + def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: message = config.get("message", "Review required.") if isinstance(message, str) and "{{" in message: @@ -32,8 +37,14 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: on_reject = config.get("on_reject", "abort") show_file = config.get("show_file") - if show_file and isinstance(show_file, str) and "{{" in show_file: + if isinstance(show_file, str) and "{{" in show_file: show_file = evaluate_expression(show_file, context) + # ``evaluate_expression`` can return a non-string for a single + # expression (e.g. a number from a prior step), and a literal + # non-string is also possible; coerce so it is rendered rather + # than silently skipped at the prompt. + if show_file is not None: + show_file = str(show_file) output = { "message": message, @@ -48,7 +59,7 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: return StepResult(status=StepStatus.PAUSED, output=output) # Interactive: prompt the user - choice = self._prompt(message, options) + choice = self._prompt(message, options, show_file) output["choice"] = choice if choice in ("reject", "abort"): @@ -68,10 +79,20 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: return StepResult(status=StepStatus.COMPLETED, output=output) @staticmethod - def _prompt(message: str, options: list[str]) -> str: - """Display gate message and prompt for a choice.""" + def _prompt(message: str, options: list[str], show_file: str | None = None) -> str: + """Display gate message and prompt for a choice. + + When ``show_file`` names a readable file, its contents are shown + before the options so the operator can review the material the + gate refers to. + """ print("\n ┌─ Gate ─────────────────────────────────────") print(f" │ {message}") + if show_file: + print(" │") + print(f" │ {show_file}:") + for line in GateStep._read_show_file(show_file): + print(f" │ {line}") print(" │") for i, opt in enumerate(options, 1): print(f" │ [{i}] {opt}") @@ -90,6 +111,34 @@ def _prompt(message: str, options: list[str]) -> str: return next(o for o in options if o.lower() == raw.lower()) print(f" Invalid choice. Enter 1-{len(options)} or an option name.") + @staticmethod + def _read_show_file(show_file: str) -> list[str]: + """Return the lines of ``show_file`` for display. + + Reads at most ``MAX_SHOW_FILE_LINES`` lines so a large file cannot + flood the prompt, and returns a short notice instead of raising + when the file is missing or cannot be decoded, so a misconfigured + path never breaks the interactive prompt. + """ + lines: list[str] = [] + truncated = False + try: + with Path(show_file).open(encoding="utf-8") as handle: + for line in handle: + if len(lines) >= GateStep.MAX_SHOW_FILE_LINES: + truncated = True + break + lines.append(line.rstrip("\n")) + except (OSError, UnicodeDecodeError) as exc: + return [f"(could not read file: {exc})"] + if not lines and not truncated: + return ["(file is empty)"] + if truncated: + lines.append( + f"… (output truncated at {GateStep.MAX_SHOW_FILE_LINES} lines)" + ) + return lines + def validate(self, config: dict[str, Any]) -> list[str]: errors = super().validate(config) if "message" not in config: diff --git a/tests/test_workflows.py b/tests/test_workflows.py index a9c8c4c606..21f0c1a9bf 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -822,6 +822,115 @@ def test_validate_invalid_on_reject(self): }) assert any("on_reject" in e for e in errors) + def test_interactive_prompt_renders_show_file(self, tmp_path, monkeypatch, capsys): + from specify_cli.workflows.steps.gate import GateStep + from specify_cli.workflows.base import StepContext, StepStatus + + review = tmp_path / "spec.md" + review.write_text("LINE-ONE\nLINE-TWO\n", encoding="utf-8") + + monkeypatch.setattr("sys.stdin.isatty", lambda: True) + monkeypatch.setattr("builtins.input", lambda _prompt="": "1") + + step = GateStep() + config = { + "id": "review", + "message": "Review the spec.", + "show_file": str(review), + "options": ["approve", "reject"], + } + result = step.execute(config, StepContext()) + out = capsys.readouterr().out + + assert "LINE-ONE" in out and "LINE-TWO" in out + assert str(review) in out + assert result.status == StepStatus.COMPLETED + assert result.output["choice"] == "approve" + + def test_interactive_prompt_missing_show_file_does_not_crash( + self, tmp_path, monkeypatch, capsys + ): + from specify_cli.workflows.steps.gate import GateStep + from specify_cli.workflows.base import StepContext, StepStatus + + missing = tmp_path / "does-not-exist.md" + + monkeypatch.setattr("sys.stdin.isatty", lambda: True) + monkeypatch.setattr("builtins.input", lambda _prompt="": "1") + + step = GateStep() + config = { + "id": "review", + "message": "Review.", + "show_file": str(missing), + "options": ["approve", "reject"], + } + result = step.execute(config, StepContext()) + out = capsys.readouterr().out + + assert "could not read file" in out + assert result.status == StepStatus.COMPLETED + + def test_non_interactive_show_file_still_pauses_without_reading( + self, tmp_path, monkeypatch + ): + from specify_cli.workflows.steps.gate import GateStep + from specify_cli.workflows.base import StepContext, StepStatus + + review = tmp_path / "spec.md" + review.write_text("CONTENT\n", encoding="utf-8") + + monkeypatch.setattr("sys.stdin.isatty", lambda: False) + + step = GateStep() + config = { + "id": "review", + "message": "Review.", + "show_file": str(review), + "options": ["approve", "reject"], + } + result = step.execute(config, StepContext()) + assert result.status == StepStatus.PAUSED + assert result.output["show_file"] == str(review) + + def test_read_show_file_empty(self, tmp_path): + from specify_cli.workflows.steps.gate import GateStep + + empty = tmp_path / "empty.md" + empty.write_text("", encoding="utf-8") + assert GateStep._read_show_file(str(empty)) == ["(file is empty)"] + + def test_read_show_file_truncates_large_file(self, tmp_path): + from specify_cli.workflows.steps.gate import GateStep + + big = tmp_path / "big.md" + big.write_text( + "\n".join(f"line{i}" for i in range(GateStep.MAX_SHOW_FILE_LINES + 50)), + encoding="utf-8", + ) + rendered = GateStep._read_show_file(str(big)) + # MAX_SHOW_FILE_LINES content lines + one truncation notice line. + assert len(rendered) == GateStep.MAX_SHOW_FILE_LINES + 1 + assert "truncated" in rendered[-1] + + def test_templated_show_file_resolving_to_non_string_is_coerced(self): + from specify_cli.workflows.steps.gate import GateStep + from specify_cli.workflows.base import StepContext, StepStatus + + # A single-expression template can resolve to a non-string (e.g. a + # number from a prior step); it must be coerced to str, not skipped. + step = GateStep() + ctx = StepContext(steps={"prev": {"output": {"ref": 123}}}) + config = { + "id": "review", + "message": "Review.", + "show_file": "{{ steps.prev.output.ref }}", + "options": ["approve", "reject"], + } + result = step.execute(config, ctx) # non-interactive -> PAUSED + assert result.status == StepStatus.PAUSED + assert result.output["show_file"] == "123" + class TestIfThenStep: """Test the if/then/else step type.""" From bf376353299d97322a6d22793a42a5d464e04b29 Mon Sep 17 00:00:00 2001 From: Huy Do Date: Thu, 4 Jun 2026 00:08:56 +0700 Subject: [PATCH 2/9] fix(workflows): keep gate _prompt signature stable and harden show_file reads The gate prompt rendered show_file by passing it as a third positional argument to _prompt. A test that stubs _prompt with a two-argument lambda (test_gate_abort_still_halts_with_continue_on_error) then failed once the branch caught up to main, because the call site passed three arguments to the two-argument stub. Compose the show_file material into the displayed message in execute() and keep _prompt to its (message, options) contract. Display data no longer widens the interactive seam, so stubbing _prompt stays stable and future review material can be added without breaking callers. _prompt now renders a multi-line message inside the gate box. Also catch ValueError in _read_show_file so a path the OS rejects outright (e.g. an embedded NUL byte) degrades to a notice instead of crashing the prompt, matching the helper's stated contract. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../workflows/steps/gate/__init__.py | 51 ++++++++++++------- tests/test_workflows.py | 9 ++++ 2 files changed, 43 insertions(+), 17 deletions(-) diff --git a/src/specify_cli/workflows/steps/gate/__init__.py b/src/specify_cli/workflows/steps/gate/__init__.py index 466f4e300f..b40d39a501 100644 --- a/src/specify_cli/workflows/steps/gate/__init__.py +++ b/src/specify_cli/workflows/steps/gate/__init__.py @@ -54,12 +54,16 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: "choice": None, } - # Non-interactive: pause for later resume + # Non-interactive: pause for later resume (the file is not read here) if not sys.stdin.isatty(): return StepResult(status=StepStatus.PAUSED, output=output) - # Interactive: prompt the user - choice = self._prompt(message, options, show_file) + # Interactive: prompt the user. ``show_file`` contents are folded + # into the displayed message so the operator can review the + # referenced material before choosing. Composing the prompt text + # here keeps ``_prompt`` to its ``(message, options)`` contract, so + # adding review material never widens the interactive seam. + choice = self._prompt(self._compose_prompt(message, show_file), options) output["choice"] = choice if choice in ("reject", "abort"): @@ -78,21 +82,32 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: return StepResult(status=StepStatus.COMPLETED, output=output) + @classmethod + def _compose_prompt(cls, message: str, show_file: str | None) -> str: + """Build the gate's display text. + + When ``show_file`` names a file, its contents (read safely, see + ``_read_show_file``) are appended below the message so the operator + can review the referenced material before choosing. Returns a + possibly multi-line string that ``_prompt`` renders inside the box. + """ + if not show_file: + return message + body = "\n".join( + [f"{show_file}:", *(f" {line}" for line in cls._read_show_file(show_file))] + ) + return f"{message}\n\n{body}" + @staticmethod - def _prompt(message: str, options: list[str], show_file: str | None = None) -> str: - """Display gate message and prompt for a choice. + def _prompt(message: str, options: list[str]) -> str: + """Display the gate message and prompt for a choice. - When ``show_file`` names a readable file, its contents are shown - before the options so the operator can review the material the - gate refers to. + ``message`` may span multiple lines (e.g. when review material has + been folded in); each line is rendered inside the gate box. """ print("\n ┌─ Gate ─────────────────────────────────────") - print(f" │ {message}") - if show_file: - print(" │") - print(f" │ {show_file}:") - for line in GateStep._read_show_file(show_file): - print(f" │ {line}") + for line in message.split("\n"): + print(f" │ {line}" if line else " │") print(" │") for i, opt in enumerate(options, 1): print(f" │ [{i}] {opt}") @@ -117,8 +132,10 @@ def _read_show_file(show_file: str) -> list[str]: Reads at most ``MAX_SHOW_FILE_LINES`` lines so a large file cannot flood the prompt, and returns a short notice instead of raising - when the file is missing or cannot be decoded, so a misconfigured - path never breaks the interactive prompt. + when the file is missing, undecodable, or names an invalid path, + so a misconfigured ``show_file`` never breaks the interactive + prompt. ``ValueError`` covers paths the OS rejects outright (e.g. + an embedded NUL byte), which ``Path.open`` raises before any I/O. """ lines: list[str] = [] truncated = False @@ -129,7 +146,7 @@ def _read_show_file(show_file: str) -> list[str]: truncated = True break lines.append(line.rstrip("\n")) - except (OSError, UnicodeDecodeError) as exc: + except (OSError, UnicodeDecodeError, ValueError) as exc: return [f"(could not read file: {exc})"] if not lines and not truncated: return ["(file is empty)"] diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 7c7d28e342..a8c3e6165c 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -913,6 +913,15 @@ def test_read_show_file_truncates_large_file(self, tmp_path): assert len(rendered) == GateStep.MAX_SHOW_FILE_LINES + 1 assert "truncated" in rendered[-1] + def test_read_show_file_invalid_path_does_not_raise(self): + from specify_cli.workflows.steps.gate import GateStep + + # An embedded NUL byte makes the OS reject the path with ValueError + # before any I/O; it must degrade to a notice, not crash the prompt. + rendered = GateStep._read_show_file("bad\x00path.md") + assert len(rendered) == 1 + assert rendered[0].startswith("(could not read file:") + def test_templated_show_file_resolving_to_non_string_is_coerced(self): from specify_cli.workflows.steps.gate import GateStep from specify_cli.workflows.base import StepContext, StepStatus From 017e84d445c6da23299a38745f3bf9ae2fe911cb Mon Sep 17 00:00:00 2001 From: Huy Do Date: Thu, 4 Jun 2026 00:21:08 +0700 Subject: [PATCH 3/9] fix(workflows): coerce gate prompt message to str before rendering The multi-line render loop split the message on newlines, which assumes a str. A non-string message (e.g. a YAML numeric literal) previously rendered fine through the old f-string and would now raise on .split. Coerce with str() to preserve that tolerance, and add a regression test. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/specify_cli/workflows/steps/gate/__init__.py | 4 +++- tests/test_workflows.py | 16 ++++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/src/specify_cli/workflows/steps/gate/__init__.py b/src/specify_cli/workflows/steps/gate/__init__.py index b40d39a501..1f4041737a 100644 --- a/src/specify_cli/workflows/steps/gate/__init__.py +++ b/src/specify_cli/workflows/steps/gate/__init__.py @@ -106,7 +106,9 @@ def _prompt(message: str, options: list[str]) -> str: been folded in); each line is rendered inside the gate box. """ print("\n ┌─ Gate ─────────────────────────────────────") - for line in message.split("\n"): + # ``str(...)`` so a non-string message (e.g. a YAML numeric literal) + # renders rather than raising on ``.split``. + for line in str(message).split("\n"): print(f" │ {line}" if line else " │") print(" │") for i, opt in enumerate(options, 1): diff --git a/tests/test_workflows.py b/tests/test_workflows.py index a8c3e6165c..5fc55471a0 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -922,6 +922,22 @@ def test_read_show_file_invalid_path_does_not_raise(self): assert len(rendered) == 1 assert rendered[0].startswith("(could not read file:") + def test_interactive_non_string_message_renders(self, monkeypatch, capsys): + from specify_cli.workflows.steps.gate import GateStep + from specify_cli.workflows.base import StepContext, StepStatus + + # A YAML numeric literal reaches the prompt as a non-string; it must + # render rather than crash on the multi-line split. + monkeypatch.setattr("sys.stdin.isatty", lambda: True) + monkeypatch.setattr("builtins.input", lambda _prompt="": "1") + + step = GateStep() + config = {"id": "review", "message": 123, "options": ["approve", "reject"]} + result = step.execute(config, StepContext()) + out = capsys.readouterr().out + assert "123" in out + assert result.status == StepStatus.COMPLETED + def test_templated_show_file_resolving_to_non_string_is_coerced(self): from specify_cli.workflows.steps.gate import GateStep from specify_cli.workflows.base import StepContext, StepStatus From 06bcd43e72aac1fda2b4c4829c242c7b4b87a83b Mon Sep 17 00:00:00 2001 From: Huy Do Date: Thu, 4 Jun 2026 04:27:06 +0700 Subject: [PATCH 4/9] test(workflows): make gate stdin handling robust; tidy compose_prompt typing Address review feedback on the gate tests and helper: - Swap the gate module's sys.stdin for a fixed-isatty stub (shared _StubStdin / _force_gate_stdin helpers) instead of setattr on sys.stdin.isatty, which is not assignable under some pytest capture modes. This also forces the non-interactive tests to a non-TTY so they cannot block on input() when run in a real terminal. - The non-interactive show_file test now hard-fails if _read_show_file is called, proving the file is not read on the PAUSED path. - _compose_prompt accepts a non-string message (e.g. a YAML numeric literal) and always returns str via str(message), keeping its annotation and docstring accurate; the redundant coercion in _prompt is removed. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../workflows/steps/gate/__init__.py | 17 +++---- tests/test_workflows.py | 46 +++++++++++++++++-- 2 files changed, 50 insertions(+), 13 deletions(-) diff --git a/src/specify_cli/workflows/steps/gate/__init__.py b/src/specify_cli/workflows/steps/gate/__init__.py index 1f4041737a..c0b6ad2f8b 100644 --- a/src/specify_cli/workflows/steps/gate/__init__.py +++ b/src/specify_cli/workflows/steps/gate/__init__.py @@ -83,20 +83,23 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: return StepResult(status=StepStatus.COMPLETED, output=output) @classmethod - def _compose_prompt(cls, message: str, show_file: str | None) -> str: + def _compose_prompt(cls, message: object, show_file: str | None) -> str: """Build the gate's display text. + ``message`` may be a non-string (e.g. a YAML numeric literal that + ``execute`` does not coerce), so it is rendered through ``str``. When ``show_file`` names a file, its contents (read safely, see ``_read_show_file``) are appended below the message so the operator - can review the referenced material before choosing. Returns a - possibly multi-line string that ``_prompt`` renders inside the box. + can review the referenced material before choosing. Always returns a + ``str`` — possibly multi-line — for ``_prompt`` to render in the box. """ + text = str(message) if not show_file: - return message + return text body = "\n".join( [f"{show_file}:", *(f" {line}" for line in cls._read_show_file(show_file))] ) - return f"{message}\n\n{body}" + return f"{text}\n\n{body}" @staticmethod def _prompt(message: str, options: list[str]) -> str: @@ -106,9 +109,7 @@ def _prompt(message: str, options: list[str]) -> str: been folded in); each line is rendered inside the gate box. """ print("\n ┌─ Gate ─────────────────────────────────────") - # ``str(...)`` so a non-string message (e.g. a YAML numeric literal) - # renders rather than raising on ``.split``. - for line in str(message).split("\n"): + for line in message.split("\n"): print(f" │ {line}" if line else " │") print(" │") for i, opt in enumerate(options, 1): diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 5fc55471a0..62e4f336c0 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -784,6 +784,29 @@ def test_validate_missing_run(self): assert any("missing 'run'" in e for e in errors) +class _StubStdin: + """Stdin stub with a fixed ``isatty`` result. + + Swapped in via the gate module's ``sys.stdin`` rather than + ``setattr(sys.stdin, "isatty", …)`` because ``TextIOWrapper.isatty`` + is not assignable under some runners (e.g. pytest with capture + disabled), and because forcing the value keeps interactive/non- + interactive tests deterministic regardless of how the suite is run. + """ + + def __init__(self, tty: bool): + self._tty = tty + + def isatty(self) -> bool: + return self._tty + + +def _force_gate_stdin(monkeypatch, *, tty: bool): + from specify_cli.workflows.steps import gate as gate_module + + monkeypatch.setattr(gate_module.sys, "stdin", _StubStdin(tty=tty)) + + class TestGateStep: """Test the gate step type.""" @@ -829,7 +852,7 @@ def test_interactive_prompt_renders_show_file(self, tmp_path, monkeypatch, capsy review = tmp_path / "spec.md" review.write_text("LINE-ONE\nLINE-TWO\n", encoding="utf-8") - monkeypatch.setattr("sys.stdin.isatty", lambda: True) + _force_gate_stdin(monkeypatch, tty=True) monkeypatch.setattr("builtins.input", lambda _prompt="": "1") step = GateStep() @@ -855,7 +878,7 @@ def test_interactive_prompt_missing_show_file_does_not_crash( missing = tmp_path / "does-not-exist.md" - monkeypatch.setattr("sys.stdin.isatty", lambda: True) + _force_gate_stdin(monkeypatch, tty=True) monkeypatch.setattr("builtins.input", lambda _prompt="": "1") step = GateStep() @@ -880,7 +903,17 @@ def test_non_interactive_show_file_still_pauses_without_reading( review = tmp_path / "spec.md" review.write_text("CONTENT\n", encoding="utf-8") - monkeypatch.setattr("sys.stdin.isatty", lambda: False) + _force_gate_stdin(monkeypatch, tty=False) + # The non-interactive path must not read the file; hard-fail if it does. + monkeypatch.setattr( + GateStep, + "_read_show_file", + staticmethod( + lambda _p: (_ for _ in ()).throw( + AssertionError("show_file read on the non-interactive path") + ) + ), + ) step = GateStep() config = { @@ -928,7 +961,7 @@ def test_interactive_non_string_message_renders(self, monkeypatch, capsys): # A YAML numeric literal reaches the prompt as a non-string; it must # render rather than crash on the multi-line split. - monkeypatch.setattr("sys.stdin.isatty", lambda: True) + _force_gate_stdin(monkeypatch, tty=True) monkeypatch.setattr("builtins.input", lambda _prompt="": "1") step = GateStep() @@ -938,12 +971,15 @@ def test_interactive_non_string_message_renders(self, monkeypatch, capsys): assert "123" in out assert result.status == StepStatus.COMPLETED - def test_templated_show_file_resolving_to_non_string_is_coerced(self): + def test_templated_show_file_resolving_to_non_string_is_coerced(self, monkeypatch): from specify_cli.workflows.steps.gate import GateStep from specify_cli.workflows.base import StepContext, StepStatus # A single-expression template can resolve to a non-string (e.g. a # number from a prior step); it must be coerced to str, not skipped. + # Force a non-TTY so the path stays non-interactive (-> PAUSED) and + # cannot block on input under a real terminal. + _force_gate_stdin(monkeypatch, tty=False) step = GateStep() ctx = StepContext(steps={"prev": {"output": {"ref": 123}}}) config = { From 4d52dd942aaaf56277c23f4751b9898fab672011 Mon Sep 17 00:00:00 2001 From: Huy Do Date: Thu, 4 Jun 2026 05:38:14 +0700 Subject: [PATCH 5/9] fix(workflows): strip control chars from gate show_file; default tests non-TTY Address review feedback: - _read_show_file strips C0 control characters (except tab) from each line, so a show_file containing ANSI escape sequences (e.g. \x1b[2J) cannot clear the screen or spoof the prompt/options when rendered to a terminal. - Add an autouse fixture on TestGateStep that defaults every gate test to a non-TTY stdin, so no test can drop into the interactive prompt and block on input() when the suite runs under a real TTY. Interactive tests opt back in via _force_gate_stdin(tty=True); the now-redundant explicit non-TTY calls were removed. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../workflows/steps/gate/__init__.py | 11 +++++++- tests/test_workflows.py | 28 +++++++++++++++---- 2 files changed, 33 insertions(+), 6 deletions(-) diff --git a/src/specify_cli/workflows/steps/gate/__init__.py b/src/specify_cli/workflows/steps/gate/__init__.py index c0b6ad2f8b..e6526a634d 100644 --- a/src/specify_cli/workflows/steps/gate/__init__.py +++ b/src/specify_cli/workflows/steps/gate/__init__.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re import sys from pathlib import Path from typing import Any @@ -9,6 +10,11 @@ from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus from specify_cli.workflows.expressions import evaluate_expression +#: C0 control characters except tab — stripped from ``show_file`` content so a +#: file containing ANSI escapes (e.g. ``\x1b[2J``) cannot clear the screen or +#: spoof the prompt/options when its lines are printed to the terminal. +_CONTROL_CHARS = re.compile(r"[\x00-\x08\x0b-\x1f\x7f]") + class GateStep(StepBase): """Interactive review gate. @@ -139,6 +145,9 @@ def _read_show_file(show_file: str) -> list[str]: so a misconfigured ``show_file`` never breaks the interactive prompt. ``ValueError`` covers paths the OS rejects outright (e.g. an embedded NUL byte), which ``Path.open`` raises before any I/O. + + Control characters are stripped from each line so file content + cannot inject ANSI escape sequences into the terminal. """ lines: list[str] = [] truncated = False @@ -148,7 +157,7 @@ def _read_show_file(show_file: str) -> list[str]: if len(lines) >= GateStep.MAX_SHOW_FILE_LINES: truncated = True break - lines.append(line.rstrip("\n")) + lines.append(_CONTROL_CHARS.sub("", line.rstrip("\n"))) except (OSError, UnicodeDecodeError, ValueError) as exc: return [f"(could not read file: {exc})"] if not lines and not truncated: diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 62e4f336c0..bf0135826e 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -810,6 +810,14 @@ def _force_gate_stdin(monkeypatch, *, tty: bool): class TestGateStep: """Test the gate step type.""" + @pytest.fixture(autouse=True) + def _non_tty_stdin_by_default(self, monkeypatch): + # Default every gate test to a non-TTY stdin so none can drop into + # the interactive prompt and block on input() when the suite runs + # with a real TTY. Interactive tests opt back in with + # _force_gate_stdin(monkeypatch, tty=True). + _force_gate_stdin(monkeypatch, tty=False) + def test_execute_returns_paused(self): from specify_cli.workflows.steps.gate import GateStep from specify_cli.workflows.base import StepContext, StepStatus @@ -903,7 +911,7 @@ def test_non_interactive_show_file_still_pauses_without_reading( review = tmp_path / "spec.md" review.write_text("CONTENT\n", encoding="utf-8") - _force_gate_stdin(monkeypatch, tty=False) + # stdin defaults to non-TTY via the autouse fixture. # The non-interactive path must not read the file; hard-fail if it does. monkeypatch.setattr( GateStep, @@ -955,6 +963,17 @@ def test_read_show_file_invalid_path_does_not_raise(self): assert len(rendered) == 1 assert rendered[0].startswith("(could not read file:") + def test_read_show_file_strips_control_chars(self, tmp_path): + from specify_cli.workflows.steps.gate import GateStep + + # A file with ANSI/control bytes must not inject escapes into the + # terminal; ESC and other C0 controls are stripped, tab is kept. + f = tmp_path / "ansi.md" + f.write_text("a\x1b[2Jb\tc\x07d\n", encoding="utf-8") + rendered = GateStep._read_show_file(str(f)) + assert rendered == ["a[2Jb\tcd"] + assert "\x1b" not in rendered[0] and "\x07" not in rendered[0] + def test_interactive_non_string_message_renders(self, monkeypatch, capsys): from specify_cli.workflows.steps.gate import GateStep from specify_cli.workflows.base import StepContext, StepStatus @@ -971,15 +990,14 @@ def test_interactive_non_string_message_renders(self, monkeypatch, capsys): assert "123" in out assert result.status == StepStatus.COMPLETED - def test_templated_show_file_resolving_to_non_string_is_coerced(self, monkeypatch): + def test_templated_show_file_resolving_to_non_string_is_coerced(self): from specify_cli.workflows.steps.gate import GateStep from specify_cli.workflows.base import StepContext, StepStatus # A single-expression template can resolve to a non-string (e.g. a # number from a prior step); it must be coerced to str, not skipped. - # Force a non-TTY so the path stays non-interactive (-> PAUSED) and - # cannot block on input under a real terminal. - _force_gate_stdin(monkeypatch, tty=False) + # stdin defaults to non-TTY via the autouse fixture, so the path + # stays non-interactive (-> PAUSED) and cannot block on input. step = GateStep() ctx = StepContext(steps={"prev": {"output": {"ref": 123}}}) config = { From 8b653d7aee81e3d63aa25e6487bb9c708342eb6a Mon Sep 17 00:00:00 2001 From: Huy Do Date: Thu, 4 Jun 2026 19:27:18 +0700 Subject: [PATCH 6/9] test(workflows): localize gate stdin patch to the gate module's sys _force_gate_stdin rebinds the gate module's `sys` name to a stand-in whose stdin has a fixed isatty() and which delegates every other attribute to the real sys, instead of mutating the process-wide sys.stdin. This keeps the patch local to the gate module and leaves real stdin untouched. The gate abort test, which used the same process-wide swap, now shares the helper, so the pattern exists in exactly one place. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_workflows.py | 44 ++++++++++++++++++++++++----------------- 1 file changed, 26 insertions(+), 18 deletions(-) diff --git a/tests/test_workflows.py b/tests/test_workflows.py index bf0135826e..14dbd84666 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -15,6 +15,7 @@ import json import os import shutil +import sys import tempfile from pathlib import Path @@ -785,13 +786,12 @@ def test_validate_missing_run(self): class _StubStdin: - """Stdin stub with a fixed ``isatty`` result. + """Stdin stub exposing only a fixed ``isatty`` result. - Swapped in via the gate module's ``sys.stdin`` rather than - ``setattr(sys.stdin, "isatty", …)`` because ``TextIOWrapper.isatty`` - is not assignable under some runners (e.g. pytest with capture - disabled), and because forcing the value keeps interactive/non- - interactive tests deterministic regardless of how the suite is run. + A real ``TextIOWrapper.isatty`` is not assignable under some runners + (e.g. pytest with capture disabled), so the gate tests force the value + through this stub to stay deterministic regardless of how the suite is + run. """ def __init__(self, tty: bool): @@ -801,10 +801,26 @@ def isatty(self) -> bool: return self._tty +class _FakeSys: + """Stand-in for the gate module's ``sys`` with a fixed-``isatty`` stdin. + + Every other attribute delegates to the real ``sys``. Rebinding the gate + module's ``sys`` name (rather than mutating the process-wide + ``sys.stdin``) keeps the patch local to the gate module and leaves the + real stdin untouched. + """ + + def __init__(self, tty: bool): + self.stdin = _StubStdin(tty) + + def __getattr__(self, name): + return getattr(sys, name) + + def _force_gate_stdin(monkeypatch, *, tty: bool): from specify_cli.workflows.steps import gate as gate_module - monkeypatch.setattr(gate_module.sys, "stdin", _StubStdin(tty=tty)) + monkeypatch.setattr(gate_module, "sys", _FakeSys(tty=tty)) class TestGateStep: @@ -2735,19 +2751,11 @@ def test_gate_abort_still_halts_with_continue_on_error( from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine from specify_cli.workflows.base import RunStatus from specify_cli.workflows.steps.gate import GateStep - from specify_cli.workflows.steps import gate as gate_module # Force the gate step into interactive mode and feed a "reject" - # choice so the abort path actually runs in the test env - # (default behaviour returns StepStatus.PAUSED when stdin is not a TTY). - # Swap sys.stdin itself for a stub: setattr on the real - # TextIOWrapper's `isatty` method is not assignable under some - # runners (e.g. pytest with capture disabled). - class _TTYStdin: - def isatty(self) -> bool: - return True - - monkeypatch.setattr(gate_module.sys, "stdin", _TTYStdin()) + # choice so the abort path actually runs in the test env (default + # behaviour returns StepStatus.PAUSED when stdin is not a TTY). + _force_gate_stdin(monkeypatch, tty=True) monkeypatch.setattr( GateStep, "_prompt", staticmethod(lambda _msg, _opts: "reject") ) From 4f6f843d75d84e845b6ddad26a216b0c90b0991e Mon Sep 17 00:00:00 2001 From: Huy Do Date: Thu, 4 Jun 2026 21:35:13 +0700 Subject: [PATCH 7/9] fix(workflows): sanitize the displayed gate show_file path, not just content MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Control characters were stripped from show_file *contents* but the path was still printed verbatim as the header (`f"{show_file}:"`) and echoed in the read-error notice, so a show_file path containing ANSI escapes could still inject terminal sequences. Centralize stripping in `_sanitize_for_display` and apply it to every show_file-derived string that reaches the terminal — the displayed path, each file line, and the error notice — while still opening the file with the original path. Add a test for path sanitization. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../workflows/steps/gate/__init__.py | 26 ++++++++++++++----- tests/test_workflows.py | 10 +++++++ 2 files changed, 30 insertions(+), 6 deletions(-) diff --git a/src/specify_cli/workflows/steps/gate/__init__.py b/src/specify_cli/workflows/steps/gate/__init__.py index e6526a634d..fc712f1880 100644 --- a/src/specify_cli/workflows/steps/gate/__init__.py +++ b/src/specify_cli/workflows/steps/gate/__init__.py @@ -10,12 +10,22 @@ from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus from specify_cli.workflows.expressions import evaluate_expression -#: C0 control characters except tab — stripped from ``show_file`` content so a -#: file containing ANSI escapes (e.g. ``\x1b[2J``) cannot clear the screen or -#: spoof the prompt/options when its lines are printed to the terminal. +#: C0 control characters except tab. Stripped from anything derived from a +#: ``show_file`` before it is printed — both the file's contents and the path +#: itself — so a file (or a path) containing ANSI escapes (e.g. ``\x1b[2J``) +#: cannot clear the screen or spoof the prompt/options at the terminal. _CONTROL_CHARS = re.compile(r"[\x00-\x08\x0b-\x1f\x7f]") +def _sanitize_for_display(text: str) -> str: + """Strip control characters so untrusted text cannot inject ANSI escapes. + + Applied to every ``show_file``-derived string that reaches the terminal: + the displayed path, each file line, and the read-error notice. + """ + return _CONTROL_CHARS.sub("", text) + + class GateStep(StepBase): """Interactive review gate. @@ -102,8 +112,11 @@ def _compose_prompt(cls, message: object, show_file: str | None) -> str: text = str(message) if not show_file: return text + # The path is opened with the original value but displayed sanitized, + # so a path that itself contains escapes cannot spoof the terminal. + header = f"{_sanitize_for_display(show_file)}:" body = "\n".join( - [f"{show_file}:", *(f" {line}" for line in cls._read_show_file(show_file))] + [header, *(f" {line}" for line in cls._read_show_file(show_file))] ) return f"{text}\n\n{body}" @@ -157,9 +170,10 @@ def _read_show_file(show_file: str) -> list[str]: if len(lines) >= GateStep.MAX_SHOW_FILE_LINES: truncated = True break - lines.append(_CONTROL_CHARS.sub("", line.rstrip("\n"))) + lines.append(_sanitize_for_display(line.rstrip("\n"))) except (OSError, UnicodeDecodeError, ValueError) as exc: - return [f"(could not read file: {exc})"] + # ``exc`` echoes the (possibly hostile) path, so sanitize it too. + return [_sanitize_for_display(f"(could not read file: {exc})")] if not lines and not truncated: return ["(file is empty)"] if truncated: diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 14dbd84666..2cd5ca6433 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -990,6 +990,16 @@ def test_read_show_file_strips_control_chars(self, tmp_path): assert rendered == ["a[2Jb\tcd"] assert "\x1b" not in rendered[0] and "\x07" not in rendered[0] + def test_compose_prompt_sanitizes_show_file_path(self): + from specify_cli.workflows.steps.gate import GateStep + + # The displayed path header (and the read-error notice it produces) + # must not carry escapes even when the path string itself contains + # control characters; the file is still opened with the raw value. + out = GateStep._compose_prompt("Review.", "evil\x1b[2Jpath.md") + assert "\x1b" not in out + assert "evil[2Jpath.md:" in out + def test_interactive_non_string_message_renders(self, monkeypatch, capsys): from specify_cli.workflows.steps.gate import GateStep from specify_cli.workflows.base import StepContext, StepStatus From ead174cd65bf3df7e8dd1b903177e1b54a2aebd5 Mon Sep 17 00:00:00 2001 From: Huy Do Date: Thu, 4 Jun 2026 21:38:26 +0700 Subject: [PATCH 8/9] refactor(workflows): inline control-char stripping, drop the helper Reuse the existing _CONTROL_CHARS regex directly at the three display sites instead of wrapping it in a one-line helper. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../workflows/steps/gate/__init__.py | 19 +++++-------------- 1 file changed, 5 insertions(+), 14 deletions(-) diff --git a/src/specify_cli/workflows/steps/gate/__init__.py b/src/specify_cli/workflows/steps/gate/__init__.py index fc712f1880..1b7a47aed8 100644 --- a/src/specify_cli/workflows/steps/gate/__init__.py +++ b/src/specify_cli/workflows/steps/gate/__init__.py @@ -17,15 +17,6 @@ _CONTROL_CHARS = re.compile(r"[\x00-\x08\x0b-\x1f\x7f]") -def _sanitize_for_display(text: str) -> str: - """Strip control characters so untrusted text cannot inject ANSI escapes. - - Applied to every ``show_file``-derived string that reaches the terminal: - the displayed path, each file line, and the read-error notice. - """ - return _CONTROL_CHARS.sub("", text) - - class GateStep(StepBase): """Interactive review gate. @@ -112,9 +103,9 @@ def _compose_prompt(cls, message: object, show_file: str | None) -> str: text = str(message) if not show_file: return text - # The path is opened with the original value but displayed sanitized, + # The path is opened with the original value but displayed stripped, # so a path that itself contains escapes cannot spoof the terminal. - header = f"{_sanitize_for_display(show_file)}:" + header = f"{_CONTROL_CHARS.sub('', show_file)}:" body = "\n".join( [header, *(f" {line}" for line in cls._read_show_file(show_file))] ) @@ -170,10 +161,10 @@ def _read_show_file(show_file: str) -> list[str]: if len(lines) >= GateStep.MAX_SHOW_FILE_LINES: truncated = True break - lines.append(_sanitize_for_display(line.rstrip("\n"))) + lines.append(_CONTROL_CHARS.sub("", line.rstrip("\n"))) except (OSError, UnicodeDecodeError, ValueError) as exc: - # ``exc`` echoes the (possibly hostile) path, so sanitize it too. - return [_sanitize_for_display(f"(could not read file: {exc})")] + # ``exc`` echoes the (possibly hostile) path, so strip it too. + return [_CONTROL_CHARS.sub("", f"(could not read file: {exc})")] if not lines and not truncated: return ["(file is empty)"] if truncated: From 3905fbc18e53e4ac1da06127635cf10b78df1f3d Mon Sep 17 00:00:00 2001 From: Huy Do Date: Thu, 4 Jun 2026 23:09:45 +0700 Subject: [PATCH 9/9] fix(workflows): also strip LF and C1 controls from gate show_file display The control-char class skipped LF (so an embedded newline in a show_file path could break the boxed layout) and the C1 range (so \x9b CSI and other 8-bit controls survived). Widen the class to [\x00-\x08\x0a-\x1f\x7f-\x9f] (still keeping tab). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/specify_cli/workflows/steps/gate/__init__.py | 10 +++++----- tests/test_workflows.py | 7 ++++--- 2 files changed, 9 insertions(+), 8 deletions(-) diff --git a/src/specify_cli/workflows/steps/gate/__init__.py b/src/specify_cli/workflows/steps/gate/__init__.py index 1b7a47aed8..a2e473244e 100644 --- a/src/specify_cli/workflows/steps/gate/__init__.py +++ b/src/specify_cli/workflows/steps/gate/__init__.py @@ -10,11 +10,11 @@ from specify_cli.workflows.base import StepBase, StepContext, StepResult, StepStatus from specify_cli.workflows.expressions import evaluate_expression -#: C0 control characters except tab. Stripped from anything derived from a -#: ``show_file`` before it is printed — both the file's contents and the path -#: itself — so a file (or a path) containing ANSI escapes (e.g. ``\x1b[2J``) -#: cannot clear the screen or spoof the prompt/options at the terminal. -_CONTROL_CHARS = re.compile(r"[\x00-\x08\x0b-\x1f\x7f]") +#: Control characters except tab: C0 (incl. LF, so an embedded newline cannot +#: break the boxed layout), DEL, and C1 (incl. ``\x9b`` CSI). Stripped from +#: anything derived from a ``show_file`` before it is printed — the file's +#: contents and the path itself — so neither can inject ANSI/terminal escapes. +_CONTROL_CHARS = re.compile(r"[\x00-\x08\x0a-\x1f\x7f-\x9f]") class GateStep(StepBase): diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 2cd5ca6433..a0f0762c2e 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -995,9 +995,10 @@ def test_compose_prompt_sanitizes_show_file_path(self): # The displayed path header (and the read-error notice it produces) # must not carry escapes even when the path string itself contains - # control characters; the file is still opened with the raw value. - out = GateStep._compose_prompt("Review.", "evil\x1b[2Jpath.md") - assert "\x1b" not in out + # control characters — ESC, LF, and C1 CSI (\x9b); the file is still + # opened with the raw value. + out = GateStep._compose_prompt("Review.", "ev\x1bil\x9b[2J\npath.md") + assert "\x1b" not in out and "\x9b" not in out assert "evil[2Jpath.md:" in out def test_interactive_non_string_message_renders(self, monkeypatch, capsys):