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
38 changes: 37 additions & 1 deletion src/specify_cli/workflows/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -193,6 +193,20 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]:
f"Must be 'string', 'number', or 'boolean'."
)

# ``enum`` must be a list. Checked here — not only via the
# ``_coerce_input`` call below — because that call is reached only
# when a ``default`` is present, and the ``integration: auto`` case
# strips ``enum`` before coercing; a scalar/string ``enum`` on an
# input with no default (or the auto-integration default) would
# otherwise slip through here and then crash ``_resolve_inputs`` with
# a raw ``TypeError`` at run time. ``None`` means "no enum".
enum_values = input_def.get("enum")
if enum_values is not None and not isinstance(enum_values, list):
errors.append(
f"Input {input_name!r} has invalid 'enum': must be a list, "
f"got {type(enum_values).__name__}."
)

# Validate the default eagerly so authoring mistakes (e.g. a
# default not in the declared enum, or a non-numeric default for
# a number input) surface at install/validation time instead of
Expand All @@ -201,7 +215,13 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]:
# enum-membership check is exempted for that exact case — the
# declared type is still enforced (e.g. ``type: number`` paired
# with ``default: "auto"`` is still rejected).
if "default" in input_def:
# A malformed (non-list) ``enum`` is already reported above; skip
# the default check when it is, so ``_coerce_input`` doesn't re-raise
# the same enum error re-framed as an "invalid default" (a confusing
# duplicate). The type/coercion of the default is still meaningfully
# checkable, but reporting one clear enum error is better than two.
enum_is_valid = enum_values is None or isinstance(enum_values, list)
if "default" in input_def and enum_is_valid:
default_value = input_def["default"]
is_auto_integration = (
input_name == "integration" and default_value == "auto"
Expand Down Expand Up @@ -1418,6 +1438,22 @@ def _coerce_input(
input_type = input_def.get("type", "string")
enum_values = input_def.get("enum")

# ``enum`` must be a list. A scalar (``enum: 5``, ``enum: true``) makes
# the ``value not in enum_values`` membership test below raise a raw
# ``TypeError`` ("argument of type 'int' is not ... iterable"), which
# escapes ``validate_workflow``'s ``except ValueError`` and breaks its
# "return errors, never raise" contract — and crashes ``_resolve_inputs``
# outright at run time. A bare string is just as wrong: ``value in "abc"``
# is a silent substring/character test, not enum membership. Require a
# list so both forms fail fast with a clear message. ``None`` means "no
# enum" and is left alone.
if enum_values is not None and not isinstance(enum_values, list):
msg = (
f"Input {name!r} has invalid 'enum': must be a list, got "
f"{type(enum_values).__name__}."
)
raise ValueError(msg)

if input_type == "number":
# Reject bools explicitly: ``bool`` is a subclass of ``int`` so
# ``float(True)`` succeeds and would silently coerce a YAML
Expand Down
100 changes: 100 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -3950,6 +3950,106 @@ def test_validate_workflow_rejects_non_string_default_for_string_type(self):
errors = validate_workflow(definition)
assert any("invalid default" in e for e in errors), errors

def test_validate_workflow_rejects_scalar_enum(self):
"""A non-list ``enum`` (``enum: 5``) must be reported as a validation
error, not crash ``validate_workflow`` with a raw ``TypeError`` from the
``value not in enum_values`` membership test."""
from specify_cli.workflows.engine import WorkflowDefinition, validate_workflow

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "scalar-enum"
name: "Scalar Enum"
version: "1.0.0"
inputs:
scope:
type: string
enum: 5
steps:
- id: noop
type: gate
message: "noop"
options: [approve]
""")
errors = validate_workflow(definition)
assert any("invalid 'enum'" in e and "must be a list" in e for e in errors), errors

def test_validate_workflow_rejects_string_enum(self):
"""A bare-string ``enum`` (``enum: abc``) must be rejected too — otherwise
``value in "abc"`` is a silent substring/character test, not enum
membership."""
from specify_cli.workflows.engine import WorkflowDefinition, validate_workflow

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "string-enum"
name: "String Enum"
version: "1.0.0"
inputs:
scope:
type: string
default: "a"
enum: "abc"
steps:
- id: noop
type: gate
message: "noop"
options: [approve]
""")
errors = validate_workflow(definition)
assert any("invalid 'enum'" in e and "must be a list" in e for e in errors), errors
# The malformed enum is reported once — not re-framed a second time as an
# "invalid default" by the default-coercion path.
assert sum("invalid 'enum'" in e for e in errors) == 1, errors

def test_resolve_inputs_rejects_scalar_enum_at_runtime(self, project_dir):
"""A non-list ``enum`` must raise a clean ``ValueError`` (not a raw
``TypeError``) when a provided value is coerced at run time, since
``execute`` does not auto-validate the definition first."""
from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "runtime-scalar-enum"
name: "Runtime Scalar Enum"
version: "1.0.0"
inputs:
scope:
type: string
enum: 5
""")
engine = WorkflowEngine(project_dir)
with pytest.raises(ValueError, match="invalid 'enum'"):
engine._resolve_inputs(definition, {"scope": "bar"})

def test_validate_workflow_accepts_list_enum(self):
"""A well-formed list ``enum`` must still validate cleanly and continue to
check the default against enum membership."""
from specify_cli.workflows.engine import WorkflowDefinition, validate_workflow

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "list-enum"
name: "List Enum"
version: "1.0.0"
inputs:
scope:
type: string
default: "full"
enum: ["full", "backend-only"]
steps:
- id: noop
type: gate
message: "noop"
options: [approve]
""")
errors = validate_workflow(definition)
assert not any("enum" in e for e in errors), errors

def test_while_loop_condition_reads_latest_iteration(self, project_dir):
"""Regression: while-loop condition must see updated step output
from the most recent iteration, not stale iteration-0 data.
Expand Down