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
13 changes: 12 additions & 1 deletion src/specify_cli/workflows/expressions.py
Original file line number Diff line number Diff line change
Expand Up @@ -192,7 +192,18 @@ def _evaluate_simple_expression(expr: str, namespace: dict[str, Any]) -> Any:
filter_name = filter_expr.strip()
if filter_name == "default":
return _filter_default(value)
return value
# No recognized filter matched — an unknown filter name, or a known
# filter used in an unsupported form. Fail loudly rather than silently
# returning the unfiltered value: a passthrough turns a mis-typed or
# unsupported filter into a wrong result with no signal. Mirrors the
# strict `from_json` handling above.
leading_name = re.match(r"\w+", filter_expr)
name = leading_name.group(0) if leading_name else filter_expr
raise ValueError(
f"unknown filter '{name}': expected one of default('x'), "
"join('sep'), map('attr'), contains('s'), or from_json "
f"(got '| {filter_expr}')"
)

# Boolean operators — parse 'or' first (lower precedence) so that
# 'a or b and c' is evaluated as 'a or (b and c)'.
Expand Down
43 changes: 43 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -342,6 +342,49 @@ def test_filter_from_json_rejects_malformed_forms(self):
"{{ steps.emit.output.stdout | " + bad + " }}", ctx
)

def test_filter_unknown_name_raises(self):
# An unregistered filter name must fail loudly rather than silently
# returning the unfiltered value (which hides a typo / unsupported
# filter as a wrong result).
import pytest
from specify_cli.workflows.expressions import evaluate_expression
from specify_cli.workflows.base import StepContext

ctx = StepContext(inputs={"items": [1, 2, 3]})
with pytest.raises(ValueError, match="unknown filter 'length'"):
evaluate_expression("{{ inputs.items | length }}", ctx)

def test_filter_unknown_name_with_args_raises(self):
# The unknown-filter path must also catch the `name(arg)` form, which
# otherwise falls through the recognized-args branch silently.
import pytest
from specify_cli.workflows.expressions import evaluate_expression
from specify_cli.workflows.base import StepContext

ctx = StepContext(inputs={"text": "hello"})
with pytest.raises(ValueError, match="unknown filter 'upper'"):
evaluate_expression("{{ inputs.text | upper('x') }}", ctx)

def test_registered_filters_unaffected(self):
# Regression: the five registered filters keep working unchanged.
from specify_cli.workflows.expressions import evaluate_expression
from specify_cli.workflows.base import StepContext

ctx = StepContext(
inputs={"tags": ["a", "b", "c"], "text": "hello world", "missing": ""},
steps={"emit": {"output": {"stdout": '{"n": 1}'}}},
)
assert (
evaluate_expression("{{ inputs.missing | default('fb') }}", ctx) == "fb"
)
assert evaluate_expression("{{ inputs.tags | join(', ') }}", ctx) == "a, b, c"
assert (
evaluate_expression("{{ inputs.text | contains('world') }}", ctx) is True
)
assert evaluate_expression(
"{{ steps.emit.output.stdout | from_json }}", ctx
) == {"n": 1}

def test_condition_evaluation(self):
from specify_cli.workflows.expressions import evaluate_condition
from specify_cli.workflows.base import StepContext
Expand Down