diff --git a/src/specify_cli/workflows/steps/switch/__init__.py b/src/specify_cli/workflows/steps/switch/__init__.py index e58d3c23c3..a63b283432 100644 --- a/src/specify_cli/workflows/steps/switch/__init__.py +++ b/src/specify_cli/workflows/steps/switch/__init__.py @@ -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( diff --git a/tests/test_workflows.py b/tests/test_workflows.py index d7cff20f6d..69e73f9e9f 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -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