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
14 changes: 14 additions & 0 deletions src/specify_cli/workflows/steps/switch/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,20 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
str_value = str(value) if value is not None else ""

cases = config.get("cases", {})
if not isinstance(cases, dict):
# The engine does not auto-validate step config, so an unvalidated
# run with a non-mapping ``cases`` (a list/scalar authoring mistake)
# would otherwise raise AttributeError from ``.items()`` below and
# crash the whole run. Fail this step loudly instead, mirroring the
# fan-out step's non-list ``items`` handling.
return StepResult(
status=StepStatus.FAILED,
error=(
f"Switch step {config.get('id', '?')!r}: 'cases' must be a "
f"mapping, got {type(cases).__name__}."
),
output={"matched_case": None, "expression_value": value},
)
for case_key, case_steps in cases.items():
if str(case_key) == str_value:
return StepResult(
Expand Down
29 changes: 29 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -2052,6 +2052,35 @@ def test_execute_no_default_no_match(self):
assert result.output["matched_case"] == "__default__"
assert result.next_steps == []

def test_execute_non_dict_cases_fails_loudly(self):
"""A non-mapping ``cases`` must fail the step, not crash the run.

``validate`` rejects a non-dict ``cases``, but the engine's
``execute()`` does not auto-validate (see ``WorkflowEngine.load_workflow``
docstring). Before the guard, ``execute`` called ``cases.items()`` on the
raw value, so an unvalidated run with a list/scalar ``cases`` raised
AttributeError and took down the whole run instead of failing this step.
Mirrors the fan-out step's non-list ``items`` handling.
"""
from specify_cli.workflows.steps.switch import SwitchStep
from specify_cli.workflows.base import StepContext, StepStatus

step = SwitchStep()
ctx = StepContext(steps={"review": {"output": {"choice": "approve"}}})
for bad_cases in (["approve"], "approve", 5):
result = step.execute(
{
"id": "route",
"expression": "{{ steps.review.output.choice }}",
"cases": bad_cases,
},
ctx,
)
assert result.status == StepStatus.FAILED
assert "'cases' must be a mapping" in (result.error or "")
# expression is still evaluated, so its value is surfaced for context.
assert result.output["expression_value"] == "approve"

def test_validate_missing_expression(self):
from specify_cli.workflows.steps.switch import SwitchStep

Expand Down
Loading