fix(workflows): validate command step input/options are mappings (#3262)

* fix(workflows): validate command step input/options are mappings

CommandStep.validate() only checked for 'command'; execute() then does input.items() and options.update(step_options). A non-mapping input:/options: (e.g. a YAML list or scalar) raised AttributeError at run time, bypassing the per-step FAILED/continue-on-error contract -- unlike the sibling steps (switch 'cases', fan-out 'step') which type-check their config fields in validate(). Add the same checks, plus a defense-in-depth coercion in execute() since the engine does not auto-validate before running a step.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* docs: fix code-comment typo in CommandStep.validate

The explanatory comment said options.update(options) but execute() does
options.update(step_options). Comment-only change; no behavior change.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

* fix(workflows): command step FAILS on malformed input/options instead of coercing

execute() previously coerced a non-mapping 'input' to {} and silently ignored a
non-mapping 'options', then dispatched the command anyway. For a workflow that
skipped validation (the engine does not auto-validate before execute()), that
let an explicitly malformed step run with empty args and report COMPLETED —
masking the config error and defeating the per-step FAILED / continue_on_error
semantics this change is meant to provide.

Both now return a FAILED StepResult with the same contract error validate()
reports (never crashing on .items()/.update()). Valid mapping configs are
unaffected. Strengthened the execute() test to assert FAILED + the exact
'must be a mapping' error for input and options (fails before: the result
carried the downstream dispatch error, not the shape error).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ali jawwad
2026-07-14 01:54:59 +05:00
committed by GitHub
parent fc1a3fd76c
commit 52c1acf8ba
2 changed files with 65 additions and 2 deletions

View File

@@ -31,6 +31,20 @@ class CommandStep(StepBase):
def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
command = config.get("command", "")
input_data = config.get("input", {})
# validate() rejects a non-mapping input, but the engine does not
# auto-validate before execute(); a workflow that skipped validation can
# still reach here. Fail the step with the same contract error rather
# than silently coercing to {} and dispatching with empty args — that
# would change the command's meaning, hide the config error, and report
# COMPLETED, defeating the per-step FAILED / continue_on_error behavior.
if not isinstance(input_data, dict):
return StepResult(
status=StepStatus.FAILED,
error=(
f"Command step {config.get('id', '?')!r}: 'input' must be a "
f"mapping, got {type(input_data).__name__}."
),
)
# Resolve expressions in input
resolved_input: dict[str, Any] = {}
@@ -50,8 +64,18 @@ class CommandStep(StepBase):
# Merge options (workflow defaults ← step overrides)
options = dict(context.default_options)
step_options = config.get("options", {})
if step_options:
options.update(step_options)
# Same rationale as 'input': a malformed options fails the step rather
# than being silently ignored (which would let an invalid step run and
# apparently complete).
if not isinstance(step_options, dict):
return StepResult(
status=StepStatus.FAILED,
error=(
f"Command step {config.get('id', '?')!r}: 'options' must be a "
f"mapping, got {type(step_options).__name__}."
),
)
options.update(step_options)
# Attempt CLI dispatch
args_str = str(resolved_input.get("args", ""))
@@ -155,4 +179,16 @@ class CommandStep(StepBase):
errors.append(
f"Command step {config.get('id', '?')!r} is missing 'command' field."
)
# execute() iterates input.items() and options.update(step_options); a
# non-mapping here would raise at run time. Validate the shape like the
# sibling steps (switch 'cases', fan-out 'step') so it is reported, not
# crashed on.
if "input" in config and not isinstance(config["input"], dict):
errors.append(
f"Command step {config.get('id', '?')!r}: 'input' must be a mapping."
)
if "options" in config and not isinstance(config["options"], dict):
errors.append(
f"Command step {config.get('id', '?')!r}: 'options' must be a mapping."
)
return errors

View File

@@ -953,6 +953,33 @@ class TestCommandStep:
errors = step.validate({"id": "test"})
assert any("missing 'command'" in e for e in errors)
def test_validate_rejects_non_mapping_input_and_options(self):
from specify_cli.workflows.steps.command import CommandStep
from specify_cli.workflows.base import StepContext, StepStatus
step = CommandStep()
# execute() does input.items() / options.update(); a non-mapping must be
# reported by validate(), not crash at run time (like switch 'cases').
for bad in (None, "args", ["a", "b"], 5):
errs = step.validate({"id": "c", "command": "/x", "input": bad})
assert any("'input' must be a mapping" in e for e in errs), bad
errs = step.validate({"id": "c", "command": "/x", "options": 42})
assert any("'options' must be a mapping" in e for e in errs)
# a valid mapping config is still accepted
assert step.validate({"id": "c", "command": "/x", "input": {"args": "y"}, "options": {"k": 1}}) == []
# execute() has no auto-validation guarantee (the engine may skip
# validate), so a non-mapping input/options FAILS the step with the same
# contract error — it does not silently coerce to empty and report
# COMPLETED (which would defeat continue_on_error).
res_in = step.execute({"id": "c", "command": "echo", "input": None}, StepContext())
assert res_in.status is StepStatus.FAILED
assert "'input' must be a mapping" in (res_in.error or "")
res_opt = step.execute(
{"id": "c", "command": "echo", "input": {}, "options": 42}, StepContext()
)
assert res_opt.status is StepStatus.FAILED
assert "'options' must be a mapping" in (res_opt.error or "")
def test_step_override_integration(self):
from unittest.mock import patch
from specify_cli.workflows.steps.command import CommandStep