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
12 changes: 12 additions & 0 deletions src/specify_cli/workflows/steps/do_while/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,18 @@ def validate(self, config: dict[str, Any]) -> list[str]:
f"Do-while step {config.get('id', '?')!r} is missing "
f"'condition' field."
)
elif not isinstance(config["condition"], str):
# The engine re-evaluates 'condition' via evaluate_condition() after
# each iteration; it returns a non-string as-is and takes bool() of
# it -- so a list/dict/number condition silently resolves to a
# truthiness (e.g. condition: [1, 2] is always truthy, looping to
# max_iterations) with no error. Reject non-strings at validation,
# mirroring the prompt/shell/command 'must be a string' checks.
# "true"/"false" and an expression like "{{ ... }}" stay valid.
errors.append(
f"Do-while step {config.get('id', '?')!r}: 'condition' must be a "
f"string, got {type(config['condition']).__name__}."
)
max_iter = config.get("max_iterations")
if max_iter is not None:
# bool is a subclass of int, so isinstance(True, int) is True and
Expand Down
12 changes: 12 additions & 0 deletions src/specify_cli/workflows/steps/if_then/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,18 @@ def validate(self, config: dict[str, Any]) -> list[str]:
errors.append(
f"If step {config.get('id', '?')!r} is missing 'condition' field."
)
elif not isinstance(config["condition"], str):
# execute() feeds 'condition' to evaluate_condition(), which returns
# a non-string as-is and takes bool() of it -- so a list/dict/number
# condition silently resolves to a truthiness (e.g. condition: [1, 2]
# is always True) with no error, branching wrongly on an authoring
# mistake. Reject non-strings at validation, mirroring the prompt/
# shell/command 'must be a string' checks. "true"/"false" and an
# expression like "{{ ... }}" are strings, so they stay valid.
errors.append(
f"If step {config.get('id', '?')!r}: 'condition' must be a "
f"string, got {type(config['condition']).__name__}."
)
if "then" not in config:
errors.append(
f"If step {config.get('id', '?')!r} is missing 'then' field."
Expand Down
12 changes: 12 additions & 0 deletions src/specify_cli/workflows/steps/while_loop/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,18 @@ def validate(self, config: dict[str, Any]) -> list[str]:
f"While step {config.get('id', '?')!r} is missing "
f"'condition' field."
)
elif not isinstance(config["condition"], str):
# execute() feeds 'condition' to evaluate_condition(), which returns
# a non-string as-is and takes bool() of it -- so a list/dict/number
# condition silently resolves to a truthiness (e.g. condition: [1, 2]
# is always truthy, spinning the loop to max_iterations) with no
# error. Reject non-strings at validation, mirroring the prompt/
# shell/command 'must be a string' checks. "true"/"false" and an
# expression like "{{ ... }}" are strings, so they stay valid.
errors.append(
f"While step {config.get('id', '?')!r}: 'condition' must be a "
f"string, got {type(config['condition']).__name__}."
)
max_iter = config.get("max_iterations")
if max_iter is not None:
# bool is a subclass of int, so isinstance(True, int) is True and
Expand Down
35 changes: 35 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -2490,6 +2490,25 @@ def test_validate_missing_condition(self):
errors = step.validate({"id": "test", "then": []})
assert any("missing 'condition'" in e for e in errors)

@pytest.mark.parametrize("bad", [["a", "b"], {"k": "v"}, 5, True])
def test_validate_rejects_non_string_condition(self, bad):
# A non-string condition is returned as-is by evaluate_condition and
# bool()-coerced, so it silently resolves to a truthiness (e.g. [1, 2]
# is always True) instead of erroring on the authoring mistake.
from specify_cli.workflows.steps.if_then import IfThenStep

step = IfThenStep()
errors = step.validate({"id": "test", "condition": bad, "then": []})
assert any("'condition' must be a string" in e for e in errors), bad

@pytest.mark.parametrize("good", ["true", "false", "{{ inputs.flag }}"])
def test_validate_accepts_string_condition(self, good):
from specify_cli.workflows.steps.if_then import IfThenStep

step = IfThenStep()
errors = step.validate({"id": "test", "condition": good, "then": []})
assert not any("'condition' must be a string" in e for e in errors), good

@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.
Expand Down Expand Up @@ -2880,6 +2899,14 @@ def test_validate_missing_fields(self):
assert any("missing 'condition'" in e for e in errors)
# max_iterations is optional (defaults to 10)

@pytest.mark.parametrize("bad", [["a", "b"], {"k": "v"}, 5, True])
def test_validate_rejects_non_string_condition(self, bad):
from specify_cli.workflows.steps.while_loop import WhileStep

step = WhileStep()
errors = step.validate({"id": "test", "condition": bad, "steps": []})
assert any("'condition' must be a string" in e for e in errors), bad

def test_validate_invalid_max_iterations(self):
from specify_cli.workflows.steps.while_loop import WhileStep

Expand Down Expand Up @@ -2994,6 +3021,14 @@ def test_validate_missing_fields(self):
assert any("missing 'condition'" in e for e in errors)
# max_iterations is optional (defaults to 10)

@pytest.mark.parametrize("bad", [["a", "b"], {"k": "v"}, 5, True])
def test_validate_rejects_non_string_condition(self, bad):
from specify_cli.workflows.steps.do_while import DoWhileStep

step = DoWhileStep()
errors = step.validate({"id": "test", "condition": bad, "steps": []})
assert any("'condition' must be a string" in e for e in errors), bad

def test_validate_steps_not_list(self):
from specify_cli.workflows.steps.do_while import DoWhileStep

Expand Down