From d8854ca79a4eb5d5d75a9d2b3fc14d3f7187f238 Mon Sep 17 00:00:00 2001 From: Noor-ul-ain001 Date: Tue, 14 Jul 2026 21:25:20 +0500 Subject: [PATCH] fix(workflows): fail while/do-while steps on non-list steps instead of crashing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `WhileStep.validate()` and `DoWhileStep.validate()` already reject a non-list `steps` body, but the engine's `execute()` path does not auto-validate (see `WorkflowEngine.load_workflow`, whose docstring notes the definition is "not yet validated"). On an unvalidated run the body is returned as `next_steps`, and the engine feeds it straight into `_execute_steps`, which iterates it as step mappings. A non-list `steps` — a single mapping or scalar authoring mistake — was iterated element-wise (a dict yields its string keys, a str its characters) and raised `AttributeError` on `.get()`, taking down the whole run; the engine invokes `step_impl.execute()` with no surrounding try/except. Guard both `execute` paths to return a FAILED StepResult naming the type error instead, mirroring the if/switch non-list-branch and fan-out non-list `items` handling. The do-while body always dispatches on the first call, so its guard is unconditional; the while body only dispatches when the condition is truthy, so its guard fires only then — a false condition leaves a non-list `steps` benign and the step completes, unchanged. The condition/expression is still evaluated first, so its result is surfaced in the step output for downstream context. Co-authored-by: Claude Opus 4.8 (1M context) --- .../workflows/steps/do_while/__init__.py | 24 +++++++ .../workflows/steps/while_loop/__init__.py | 26 ++++++++ tests/test_workflows.py | 65 +++++++++++++++++++ 3 files changed, 115 insertions(+) diff --git a/src/specify_cli/workflows/steps/do_while/__init__.py b/src/specify_cli/workflows/steps/do_while/__init__.py index f69a682140..ca6047a57a 100644 --- a/src/specify_cli/workflows/steps/do_while/__init__.py +++ b/src/specify_cli/workflows/steps/do_while/__init__.py @@ -27,6 +27,30 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: nested_steps = config.get("steps", []) condition = config.get("condition", "false") + # The engine does not auto-validate step config (see + # ``WorkflowEngine.load_workflow``) and feeds ``next_steps`` straight + # into ``_execute_steps``, which iterates them as step mappings. A + # non-list ``steps`` (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 + # ``steps``; fail this step loudly on an unvalidated run instead, + # mirroring the if/switch/fan-out steps. The body always runs on the + # first call, so unlike the while step this guard is unconditional. + if not isinstance(nested_steps, list): + return StepResult( + status=StepStatus.FAILED, + output={ + "condition": condition, + "max_iterations": max_iterations, + "loop_type": "do-while", + }, + error=( + f"Do-while step {config.get('id', '?')!r}: 'steps' must be " + f"a list of steps, got {type(nested_steps).__name__}." + ), + ) + # Always execute body at least once; the engine layer evaluates # `condition` after each iteration to decide whether to loop. return StepResult( diff --git a/src/specify_cli/workflows/steps/while_loop/__init__.py b/src/specify_cli/workflows/steps/while_loop/__init__.py index ea272543b6..e2dbb19305 100644 --- a/src/specify_cli/workflows/steps/while_loop/__init__.py +++ b/src/specify_cli/workflows/steps/while_loop/__init__.py @@ -26,6 +26,32 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: nested_steps = config.get("steps", []) result = evaluate_condition(condition, context) + + # The engine does not auto-validate step config (see + # ``WorkflowEngine.load_workflow``) and feeds ``next_steps`` straight + # into ``_execute_steps``, which iterates them as step mappings. A + # non-list ``steps`` (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 + # ``steps``; fail this step loudly on an unvalidated run instead, + # mirroring the if/switch/fan-out steps. The guard fires only when the + # body would actually be dispatched (condition truthy). The condition is + # still evaluated first, so its result is surfaced for downstream context. + if result and not isinstance(nested_steps, list): + return StepResult( + status=StepStatus.FAILED, + output={ + "condition_result": True, + "max_iterations": max_iterations, + "loop_type": "while", + }, + error=( + f"While step {config.get('id', '?')!r}: 'steps' must be a " + f"list of steps, got {type(nested_steps).__name__}." + ), + ) + if result: return StepResult( status=StepStatus.COMPLETED, diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 8fefad740e..a766283119 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -2260,6 +2260,47 @@ def test_execute_condition_false(self): assert result.output["condition_result"] is False assert result.next_steps == [] + @pytest.mark.parametrize("bad_steps", [{"id": "x"}, "oops", 5]) + def test_execute_non_list_steps_fails_loudly(self, bad_steps): + """A non-list ``steps`` reached at runtime must fail the step, not crash. + + ``validate`` rejects a non-list ``steps``, 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. The while body only dispatches when the condition is + truthy, so a non-list ``steps`` reaches ``next_steps`` and would crash + the engine's step iteration on an unvalidated run. Mirrors the + if/switch/fan-out non-list handling. + """ + from specify_cli.workflows.steps.while_loop import WhileStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = WhileStep() + ctx = StepContext(inputs={}) + result = step.execute( + {"id": "retry", "condition": "true", "steps": bad_steps}, ctx + ) + assert result.status == StepStatus.FAILED + assert "'steps' must be a list of steps" in (result.error or "") + assert result.next_steps == [] + + @pytest.mark.parametrize("bad_steps", [{"id": "x"}, "oops", 5]) + def test_execute_non_list_steps_ok_when_condition_false(self, bad_steps): + """A false condition never dispatches the body, so a non-list ``steps`` + stays benign — the step completes without touching ``next_steps``. + """ + from specify_cli.workflows.steps.while_loop import WhileStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = WhileStep() + ctx = StepContext(inputs={}) + result = step.execute( + {"id": "retry", "condition": "false", "steps": bad_steps}, ctx + ) + assert result.status == StepStatus.COMPLETED + assert result.output["condition_result"] is False + assert result.next_steps == [] + def test_validate_missing_fields(self): from specify_cli.workflows.steps.while_loop import WhileStep @@ -2350,6 +2391,30 @@ def test_execute_empty_steps(self): assert result.next_steps == [] assert result.status.value == "completed" + @pytest.mark.parametrize("bad_steps", [{"id": "x"}, "oops", 5]) + def test_execute_non_list_steps_fails_loudly(self, bad_steps): + """A non-list ``steps`` must fail the step, not crash the run. + + ``validate`` rejects a non-list ``steps``, 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. The do-while body always dispatches on the first call + regardless of condition, so a non-list ``steps`` always reaches + ``next_steps`` and would crash the engine's step iteration on an + unvalidated run. Mirrors the if/switch/fan-out non-list handling. + """ + from specify_cli.workflows.steps.do_while import DoWhileStep + from specify_cli.workflows.base import StepContext, StepStatus + + step = DoWhileStep() + ctx = StepContext(inputs={}) + result = step.execute( + {"id": "cycle", "condition": "false", "steps": bad_steps}, ctx + ) + assert result.status == StepStatus.FAILED + assert "'steps' must be a list of steps" in (result.error or "") + assert result.next_steps == [] + def test_validate_missing_fields(self): from specify_cli.workflows.steps.do_while import DoWhileStep