mirror of
https://github.com/github/spec-kit.git
synced 2026-08-15 14:02:04 +02:00
fix(workflows): fail if/switch steps on non-list branch instead of crashing (#3515)
* fix(workflows): fail if/switch steps on non-list branch instead of crashing `IfThenStep.validate()` and `SwitchStep.validate()` already reject a non-list branch (`then`/`else`, and `case`/`default`), 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 selected branch is fed straight into `next_steps`, which `_execute_steps` iterates as step mappings. A non-list branch — 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 switch non-mapping `cases` and fan-out non-list `items` handling. The switch guard is factored into a shared `_non_list_branch_failure` helper covering both `case` and `default` branches. A missing `else`/`default` still defaults to an empty list (COMPLETED), unchanged; the guard fires only on an explicit non-list value. 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) <noreply@anthropic.com> * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> * test(workflows): cover switch non-list branch execute paths Copilot flagged the new switch branch guards as untested: coverage stopped at a non-mapping `cases` container. Add SwitchStep.execute tests for a matched case with a non-list body and a non-list default (dict/str/int), asserting FAILED, the branch-specific error, empty next_steps, and preserved expression_value. Also add explicit `default: null` / `else: null` normalization tests so the validator-approved empty-branch contract cannot regress. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
Copilot Autofix powered by AI
parent
fd101d531e
commit
459f483f57
@@ -22,10 +22,33 @@ class IfThenStep(StepBase):
|
||||
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},
|
||||
|
||||
@@ -42,6 +42,10 @@ class SwitchStep(StepBase):
|
||||
)
|
||||
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 @@ class SwitchStep(StepBase):
|
||||
|
||||
# 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:
|
||||
|
||||
@@ -2112,6 +2112,69 @@ class TestIfThenStep:
|
||||
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.
|
||||
@@ -2245,6 +2308,91 @@ class TestSwitchStep:
|
||||
# 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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user