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
17 changes: 14 additions & 3 deletions src/specify_cli/workflows/steps/switch/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,8 @@ class SwitchStep(StepBase):
"""Multi-branch dispatch on an expression.

Evaluates ``expression:`` once, matches against ``cases:`` keys
(exact match, string-coerced). Falls through to ``default:`` if
(exact match; the resolved value is string-coerced and stripped of
surrounding whitespace first). Falls through to ``default:`` if
no case matches.
"""

Expand All @@ -22,8 +23,18 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
expression = config.get("expression", "")
value = evaluate_expression(expression, context)

# String-coerce for matching
str_value = str(value) if value is not None else ""
# String-coerce for matching, stripping surrounding whitespace first.
# The value a switch dispatches on is most often captured command
# output, and a ``shell`` step stores ``proc.stdout`` verbatim, so
# ``run: echo approve`` resolves to ``"approve\n"`` and matches no
# ``approve:`` case -- the switch silently falls through to ``default:``
# while still reporting COMPLETED. A workflow cannot strip it itself:
# the registered filters are default/join/map/contains/from_json, there
# is no ``trim``. ``evaluate_condition`` and ``InitStep._resolve_bool``
# already strip before matching a resolved string against declared
# literals, and case keys are exactly such literals. ``expression_value``
# below still reports the raw value, so nothing downstream loses it.
str_value = str(value).strip() if value is not None else ""

cases = config.get("cases", {})
if not isinstance(cases, dict):
Expand Down
49 changes: 49 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -3128,6 +3128,55 @@ def test_validate_accepts_missing_else(self):
class TestSwitchStep:
"""Test the switch step type."""

def test_execute_matches_case_ignoring_surrounding_whitespace(self):
"""A shell step's stdout keeps its trailing newline; the case must match.

`ShellStep` stores `proc.stdout` verbatim, so `run: echo approve`
resolves to "approve" plus a newline. Unstripped, that matched no
`approve:` case and the switch silently fell through to `default:`
while still reporting COMPLETED. There is no `trim` filter, so a
workflow author cannot strip it themselves.
"""
from specify_cli.workflows.steps.switch import SwitchStep
from specify_cli.workflows.base import StepContext, StepStatus

config = {
"id": "route",
"expression": "{{ steps.check.output.stdout }}",
"cases": {
"approve": [{"id": "approved", "type": "command", "command": "echo"}],
"reject": [{"id": "rejected", "type": "command", "command": "echo"}],
},
"default": [{"id": "fallback", "type": "command", "command": "echo"}],
}
for raw in ("approve\n", "approve\r\n", " approve ", "approve"):
ctx = StepContext(steps={"check": {"output": {"stdout": raw}}})
result = SwitchStep().execute(config, ctx)
assert result.status == StepStatus.COMPLETED
assert result.output["matched_case"] == "approve", repr(raw)
assert [s["id"] for s in result.next_steps] == ["approved"], repr(raw)
# The raw value is still reported unchanged.
assert result.output["expression_value"] == raw

def test_execute_still_falls_through_for_a_genuine_mismatch(self):
"""Stripping must not make unrelated values match."""
from specify_cli.workflows.steps.switch import SwitchStep
from specify_cli.workflows.base import StepContext

config = {
"id": "route",
"expression": "{{ steps.check.output.stdout }}",
"cases": {
"approve": [{"id": "approved", "type": "command", "command": "echo"}]
},
"default": [{"id": "fallback", "type": "command", "command": "echo"}],
}
ctx = StepContext(steps={"check": {"output": {"stdout": "approve-later\n"}}})
result = SwitchStep().execute(config, ctx)

assert result.output["matched_case"] == "__default__"
assert [s["id"] for s in result.next_steps] == ["fallback"]

def test_execute_matches_case(self):
from specify_cli.workflows.steps.switch import SwitchStep
from specify_cli.workflows.base import StepContext
Expand Down