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
24 changes: 24 additions & 0 deletions src/specify_cli/workflows/steps/do_while/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
26 changes: 26 additions & 0 deletions src/specify_cli/workflows/steps/while_loop/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
65 changes: 65 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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

Expand Down
Loading