Skip to content
Open
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
23 changes: 23 additions & 0 deletions src/specify_cli/workflows/steps/if_then/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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},
Expand Down
33 changes: 33 additions & 0 deletions src/specify_cli/workflows/steps/switch/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Comment thread
mnriem marked this conversation as resolved.
return self._non_list_branch_failure(
config, f"case {str(case_key)!r}", case_steps, value
)
Comment thread
mnriem marked this conversation as resolved.
Comment on lines +45 to +48
Comment on lines +45 to +48
return StepResult(
status=StepStatus.COMPLETED,
output={"matched_case": str(case_key), "expression_value": value},
Expand All @@ -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
)
Comment on lines +57 to +62
Comment on lines +59 to +62
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:
Expand Down
148 changes: 148 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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

Expand Down