diff --git a/src/specify_cli/workflows/steps/if_then/__init__.py b/src/specify_cli/workflows/steps/if_then/__init__.py index e7179a418a..b2ed880678 100644 --- a/src/specify_cli/workflows/steps/if_then/__init__.py +++ b/src/specify_cli/workflows/steps/if_then/__init__.py @@ -22,10 +22,33 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: result = evaluate_condition(condition, context) if result: + branch_name = "then" branch = config.get("then", []) else: + branch_name = "else" branch = config.get("else", []) + # The engine does not auto-validate step config (see + # ``WorkflowEngine.load_workflow``), and it feeds ``next_steps`` straight + # into ``_execute_steps`` which iterates them as step mappings. A + # non-list branch (a single mapping or scalar authoring mistake) would + # otherwise be iterated element-wise — a dict yields its string 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 switch/fan-out + # steps. A missing ``else`` defaults to ``[]`` and stays valid. + if branch is None and branch_name == "else": + branch = [] + elif not isinstance(branch, list): + return StepResult( + status=StepStatus.FAILED, + output={"condition_result": result}, + error=( + f"If step {config.get('id', '?')!r}: {branch_name!r} must be " + f"a list of steps, got {type(branch).__name__}." + ), + ) + return StepResult( status=StepStatus.COMPLETED, output={"condition_result": result}, diff --git a/src/specify_cli/workflows/steps/switch/__init__.py b/src/specify_cli/workflows/steps/switch/__init__.py index a63b283432..690df0f19a 100644 --- a/src/specify_cli/workflows/steps/switch/__init__.py +++ b/src/specify_cli/workflows/steps/switch/__init__.py @@ -42,6 +42,10 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: ) for case_key, case_steps in cases.items(): if str(case_key) == str_value: + if not isinstance(case_steps, list): + return self._non_list_branch_failure( + config, f"case {str(case_key)!r}", case_steps, value + ) return StepResult( status=StepStatus.COMPLETED, output={"matched_case": str(case_key), "expression_value": value}, @@ -50,12 +54,41 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: # Default fallback default_steps = config.get("default", []) + if default_steps is None: + default_steps = [] + elif not isinstance(default_steps, list): + return self._non_list_branch_failure( + config, "'default'", default_steps, value + ) return StepResult( status=StepStatus.COMPLETED, output={"matched_case": "__default__", "expression_value": value}, next_steps=default_steps, ) + @staticmethod + def _non_list_branch_failure( + config: dict[str, Any], branch_label: str, branch: Any, value: Any + ) -> StepResult: + """Fail the step for a non-list branch instead of crashing the run. + + ``validate`` rejects a non-list case/default branch, but the engine does + not auto-validate and feeds ``next_steps`` straight into + ``_execute_steps``, which iterates them as step mappings. A non-list + branch would be iterated element-wise (a dict yields its keys, a str its + characters) and crash the whole run with AttributeError on ``.get()``. + Fail this step loudly on an unvalidated run instead, mirroring the + non-mapping ``cases`` guard above. + """ + return StepResult( + status=StepStatus.FAILED, + output={"matched_case": None, "expression_value": value}, + error=( + f"Switch step {config.get('id', '?')!r}: {branch_label} must be " + f"a list of steps, got {type(branch).__name__}." + ), + ) + def validate(self, config: dict[str, Any]) -> list[str]: errors = super().validate(config) if "expression" not in config: diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 81aec96d3e..6a9a9eed58 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -2066,6 +2066,69 @@ def test_validate_missing_condition(self): errors = step.validate({"id": "test", "then": []}) assert any("missing 'condition'" in e for e in errors) + @pytest.mark.parametrize("bad_branch", [{"id": "x"}, "oops", 5]) + def test_execute_non_list_then_fails_loudly(self, bad_branch): + """A non-list ``then`` must fail the step, not crash the run. + + ``validate`` rejects a non-list ``then``, but the engine does not + auto-validate (see ``WorkflowEngine.load_workflow``) and feeds + ``next_steps`` straight into ``_execute_steps``, which iterates them as + step mappings. Before the guard, a non-list ``then`` (a single mapping + or scalar authoring mistake) was iterated element-wise and raised + AttributeError on ``.get()``, taking down the whole run. Mirrors the + switch/fan-out non-list handling. + """ + from specify_cli.workflows.steps.if_then import IfThenStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = IfThenStep() + ctx = StepContext(inputs={}) + result = step.execute( + {"id": "branch", "condition": "true", "then": bad_branch}, ctx + ) + assert result.status == StepStatus.FAILED + assert "'then' must be a list of steps" in (result.error or "") + assert result.next_steps == [] + + @pytest.mark.parametrize("bad_branch", [{"id": "x"}, "oops", 5]) + def test_execute_non_list_else_fails_loudly(self, bad_branch): + """A non-list ``else`` selected at runtime must fail the step, not crash. + + Same asymmetry as ``then``: the ``else`` branch is only reached when the + condition is false, so a non-list ``else`` reaches ``next_steps`` and + would crash the engine's step iteration on an unvalidated run. + """ + from specify_cli.workflows.steps.if_then import IfThenStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = IfThenStep() + ctx = StepContext(inputs={}) + result = step.execute( + {"id": "branch", "condition": "false", "then": [], "else": bad_branch}, + ctx, + ) + assert result.status == StepStatus.FAILED + assert "'else' must be a list of steps" in (result.error or "") + assert result.next_steps == [] + + def test_execute_none_else_stays_empty(self): + """An explicit ``else: null`` selected at runtime stays an empty branch. + + ``validate`` deliberately accepts ``else: None``; the execute guard must + normalize it to an empty branch (COMPLETED) rather than failing a + validator-approved workflow when the condition is false. + """ + from specify_cli.workflows.steps.if_then import IfThenStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = IfThenStep() + ctx = StepContext(inputs={}) + result = step.execute( + {"id": "branch", "condition": "false", "then": [], "else": None}, ctx + ) + assert result.status == StepStatus.COMPLETED + assert result.next_steps == [] + @pytest.mark.parametrize("bad_else", [False, 0, "", {}, 42]) def test_validate_rejects_non_list_else(self, bad_else): """A non-list 'else' must be rejected even when it is falsy. @@ -2199,6 +2262,91 @@ def test_execute_non_dict_cases_fails_loudly(self): # expression is still evaluated, so its value is surfaced for context. assert result.output["expression_value"] == "approve" + @pytest.mark.parametrize("bad_branch", [{"id": "x"}, "oops", 5]) + def test_execute_non_list_matched_case_fails_loudly(self, bad_branch): + """A matched case with a non-list body must fail the step, not crash. + + ``validate`` rejects a non-list case body, but the engine does not + auto-validate (see ``WorkflowEngine.load_workflow``) and feeds the + selected branch straight into ``_execute_steps``, which iterates it as + step mappings. A non-list body (a single mapping or scalar authoring + mistake) would be iterated element-wise and raise AttributeError on + ``.get()``, taking down the whole run. Mirrors the non-mapping + ``cases`` guard. + """ + 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"}}}) + result = step.execute( + { + "id": "route", + "expression": "{{ steps.review.output.choice }}", + "cases": {"approve": bad_branch}, + }, + ctx, + ) + assert result.status == StepStatus.FAILED + assert "case 'approve' must be a list of steps" in (result.error or "") + assert result.next_steps == [] + # expression is still evaluated, so its value is surfaced for context. + assert result.output["expression_value"] == "approve" + + @pytest.mark.parametrize("bad_branch", [{"id": "x"}, "oops", 5]) + def test_execute_non_list_default_fails_loudly(self, bad_branch): + """A non-list ``default`` reached at runtime must fail, not crash. + + Same asymmetry as the case body: ``default`` is only selected when no + case matches, so a non-list ``default`` reaches ``next_steps`` and would + crash the engine's step iteration on an unvalidated run. + """ + from specify_cli.workflows.steps.switch import SwitchStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = SwitchStep() + ctx = StepContext(steps={"review": {"output": {"choice": "other"}}}) + result = step.execute( + { + "id": "route", + "expression": "{{ steps.review.output.choice }}", + "cases": {"approve": [{"id": "plan", "command": "speckit.plan"}]}, + "default": bad_branch, + }, + ctx, + ) + assert result.status == StepStatus.FAILED + assert "'default' must be a list of steps" in (result.error or "") + assert result.next_steps == [] + # expression is still evaluated, so its value is surfaced for context. + assert result.output["expression_value"] == "other" + + @pytest.mark.parametrize("ok_default", [None, [], [{"id": "x", "command": "/y"}]]) + def test_execute_none_default_stays_empty(self, ok_default): + """An explicit ``default: null`` or a list default stays valid. + + ``validate`` deliberately accepts ``default: None``; the execute guard + must normalize it to an empty branch (COMPLETED) rather than failing a + validator-approved workflow. + """ + from specify_cli.workflows.steps.switch import SwitchStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = SwitchStep() + ctx = StepContext(steps={"review": {"output": {"choice": "other"}}}) + result = step.execute( + { + "id": "route", + "expression": "{{ steps.review.output.choice }}", + "cases": {"approve": [{"id": "plan", "command": "speckit.plan"}]}, + "default": ok_default, + }, + ctx, + ) + assert result.status == StepStatus.COMPLETED + assert result.output["matched_case"] == "__default__" + assert result.next_steps == (ok_default or []) + def test_validate_missing_expression(self): from specify_cli.workflows.steps.switch import SwitchStep