diff --git a/src/specify_cli/workflows/steps/command/__init__.py b/src/specify_cli/workflows/steps/command/__init__.py index 891b9da4e7..7a6d893ed0 100644 --- a/src/specify_cli/workflows/steps/command/__init__.py +++ b/src/specify_cli/workflows/steps/command/__init__.py @@ -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 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: # 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 @@ def validate(self, config: dict[str, Any]) -> list[str]: 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 diff --git a/tests/test_workflows.py b/tests/test_workflows.py index d7cff20f6d..02a5a8e7de 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -925,6 +925,33 @@ def test_validate_missing_command(self): 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