Skip to content
Open
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
22 changes: 22 additions & 0 deletions src/specify_cli/workflows/steps/if_then/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,30 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:

if result:
branch = config.get("then", [])
branch_name = "then"
else:
branch = config.get("else", [])
branch_name = "else"

if not isinstance(branch, list):
# The engine does not auto-validate step config and feeds
# ``next_steps`` straight into ``_execute_steps``, which iterates them
# as step mappings. A non-list ``then``/``else`` (a single mapping or
# scalar authoring mistake) would otherwise be iterated element-wise
# — a dict yields its keys, a str its characters — and crash the whole
# run with AttributeError on ``.get()``. ``validate`` already rejects
# a non-list branch; fail this step loudly on an unvalidated run
# instead, mirroring the while/do-while (#3519) / switch / fan-out
# steps. The selected branch is always returned as next_steps, so the
# guard is unconditional.
return StepResult(
status=StepStatus.FAILED,
error=(
f"If step {config.get('id', '?')!r}: {branch_name!r} must be a "
f"list of steps, got {type(branch).__name__}."
),
output={"condition_result": result},
)

return StepResult(
status=StepStatus.COMPLETED,
Expand Down
24 changes: 24 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -2060,6 +2060,30 @@ def test_execute_else_branch(self):
assert result.output["condition_result"] is False
assert result.next_steps[0]["id"] == "b"

@pytest.mark.parametrize("bad_branch", ["oops", {"a": 1}, 42])
def test_execute_rejects_non_list_branch(self, bad_branch):
"""execute() fails cleanly on a non-list then/else branch. The engine
doesn't auto-validate step config, so a non-iterable next_steps would
otherwise crash the run — completing the while/do-while guard (#3519) for
the if step (mirrors the switch step's 'cases' guard)."""
from specify_cli.workflows.steps.if_then import IfThenStep
from specify_cli.workflows.base import StepContext, StepStatus

step = IfThenStep()
cond = "{{ inputs.scope == 'full' }}"
res_then = step.execute(
{"id": "i", "condition": cond, "then": bad_branch},
StepContext(inputs={"scope": "full"}),
)
assert res_then.status is StepStatus.FAILED
assert "'then' must be a list" in (res_then.error or "")
res_else = step.execute(
{"id": "i", "condition": cond, "then": [], "else": bad_branch},
StepContext(inputs={"scope": "backend"}),
)
assert res_else.status is StepStatus.FAILED
assert "'else' must be a list" in (res_else.error or "")

def test_validate_missing_condition(self):
from specify_cli.workflows.steps.if_then import IfThenStep

Expand Down