diff --git a/src/specify_cli/workflows/steps/command/__init__.py b/src/specify_cli/workflows/steps/command/__init__.py index 891b9da4e7..dd44932b14 100644 --- a/src/specify_cli/workflows/steps/command/__init__.py +++ b/src/specify_cli/workflows/steps/command/__init__.py @@ -31,6 +31,11 @@ class CommandStep(StepBase): def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: command = config.get("command", "") input_data = config.get("input", {}) + # Defense in depth: validate() rejects a non-mapping input, but the + # engine does not auto-validate before execute(), so coerce defensively + # rather than crash on input_data.items() below. + if not isinstance(input_data, dict): + input_data = {} # Resolve expressions in input resolved_input: dict[str, Any] = {} @@ -50,7 +55,7 @@ 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: + if isinstance(step_options, dict) and step_options: options.update(step_options) # Attempt CLI dispatch @@ -155,4 +160,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 c2ec4acb4e..ded7d4c678 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -741,6 +741,24 @@ 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 + + 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}}) == [] + # defense in depth: execute() coerces a non-mapping input instead of crashing + result = step.execute({"id": "c", "command": "echo", "input": None}, StepContext()) + assert result.status is not None + def test_step_override_integration(self): from unittest.mock import patch from specify_cli.workflows.steps.command import CommandStep