Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 29 additions & 0 deletions src/specify_cli/workflows/steps/gate/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,35 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
options = config.get("options", ["approve", "reject"])
on_reject = config.get("on_reject", "abort")

# ``validate`` rejects a non-list (or empty) ``options``, and requires
# every option to be a string, but the engine does not auto-validate
# before ``execute``. An unvalidated run with a scalar/dict/None
# ``options`` would otherwise reach ``_prompt`` and crash the whole run
# with a raw ``TypeError`` (``enumerate``/``len`` on a non-iterable) or
# ``KeyError`` (indexing a dict); a non-string option would crash at the
# ``choice.lower()`` reject check with ``AttributeError``. Fail this step
# loudly instead — mirroring the switch 'cases' and command 'input'
# guards. Checked before the non-TTY short-circuit so the error surfaces
# in CI too, rather than PAUSING and crashing later on interactive resume.
if (
not isinstance(options, list)
or not options
or not all(isinstance(o, str) for o in options)
):
return StepResult(
status=StepStatus.FAILED,
error=(
f"Gate step {config.get('id', '?')!r}: 'options' must be a "
f"non-empty list of strings, got {type(options).__name__}."
),
output={
"message": message,
"options": options,
"on_reject": on_reject,
"choice": None,
},
)

show_file = config.get("show_file")
if isinstance(show_file, str) and "{{" in show_file:
show_file = evaluate_expression(show_file, context)
Expand Down
66 changes: 66 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -2093,6 +2093,72 @@ def test_templated_show_file_resolving_to_non_string_is_coerced(self):
assert result.status == StepStatus.PAUSED
assert result.output["show_file"] == "123"

@pytest.mark.parametrize(
"bad_options",
[5, {"a": "approve"}, None, [], "approve"],
)
def test_execute_non_list_options_fails_cleanly(self, monkeypatch, bad_options):
"""A malformed ``options`` must FAIL the step, not crash the run.

``validate`` rejects a non-list/empty ``options``, but the engine does
not auto-validate before ``execute``. On an interactive run a scalar/
dict/None ``options`` would otherwise reach ``_prompt`` and raise a raw
``TypeError`` (``enumerate``/``len`` on a non-iterable) or ``KeyError``
(indexing a dict), crashing the whole workflow. Mirrors the switch
'cases' and command 'input' unvalidated-execute guards."""
from specify_cli.workflows.steps.gate import GateStep
from specify_cli.workflows.base import StepContext, StepStatus

# Force an interactive TTY so the crash-prone _prompt path is reached;
# input() is stubbed so a (buggy) fall-through can't block the suite.
_force_gate_stdin(monkeypatch, tty=True)
monkeypatch.setattr("builtins.input", lambda _prompt="": "1")

step = GateStep()
config = {"id": "review", "message": "Review.", "options": bad_options}
result = step.execute(config, StepContext())

assert result.status == StepStatus.FAILED
assert "options" in (result.error or "")
assert result.output["choice"] is None

def test_execute_non_string_options_element_fails_cleanly(self, monkeypatch):
"""A non-string option element must FAIL the step, not crash.

A non-empty list with a non-string element passes the shape check but
would reach the reject test ``choice.lower()`` and raise a raw
``AttributeError`` at run time. ``validate`` reports "must be strings";
``execute`` must fail cleanly on an unvalidated run too."""
from specify_cli.workflows.steps.gate import GateStep
from specify_cli.workflows.base import StepContext, StepStatus

_force_gate_stdin(monkeypatch, tty=True)
monkeypatch.setattr("builtins.input", lambda _prompt="": "1")

step = GateStep()
config = {"id": "review", "message": "Review.", "options": [123, 456]}
result = step.execute(config, StepContext())

assert result.status == StepStatus.FAILED
assert "options" in (result.error or "")

def test_execute_non_list_options_fails_in_non_tty_too(self):
"""The guard runs before the non-TTY PAUSE short-circuit.

A malformed ``options`` should surface as FAILED in CI (non-TTY) rather
than PAUSING and only crashing later when an operator resumes on a real
terminal."""
from specify_cli.workflows.steps.gate import GateStep
from specify_cli.workflows.base import StepContext, StepStatus

# Autouse fixture already forces non-TTY stdin.
step = GateStep()
config = {"id": "review", "message": "Review.", "options": 5}
result = step.execute(config, StepContext())

assert result.status == StepStatus.FAILED
assert "options" in (result.error or "")


class TestIfThenStep:
"""Test the if/then/else step type."""
Expand Down