diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py index 11c9185..f102116 100644 --- a/packages/tangle-cli/src/tangle_cli/__init__.py +++ b/packages/tangle-cli/src/tangle_cli/__init__.py @@ -14,6 +14,6 @@ try: __version__ = metadata_version("tangle-cli") except PackageNotFoundError: - __version__ = "0.1.12" + __version__ = "0.1.13" __all__ = ["TangleDynamicDiscoveryClient", "__version__"] diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_validation.py b/packages/tangle-cli/src/tangle_cli/pipeline_validation.py index d2fc128..3d45595 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_validation.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_validation.py @@ -8,7 +8,9 @@ from __future__ import annotations +import difflib import json +import os from functools import lru_cache from importlib import resources from typing import Any, Iterable, Mapping @@ -20,7 +22,15 @@ PIPELINE_GRAPH_PATH = "implementation.graph" TASKS_PATH = f"{PIPELINE_GRAPH_PATH}.tasks" +# Escape hatch for repositories that still carry undeclared task arguments. +# Set to a truthy value to downgrade the undeclared-argument check to a no-op. +# Follows the ``TANGLE_*`` boolean env-var convention used elsewhere in the CLI +# (see ``tangle_verbose_enabled`` / ``TANGLE_TRUSTED_HYDRATION_ALLOW_ALL``). +ALLOW_UNDECLARED_TASK_ARGUMENTS_ENV = "TANGLE_ALLOW_UNDECLARED_TASK_ARGUMENTS" +_TRUTHY_ENV_VALUES = ("1", "true", "yes", "on") + __all__ = [ + "ALLOW_UNDECLARED_TASK_ARGUMENTS_ENV", "PipelineValidationError", "collect_pipeline_spec_errors", "load_pipeline_schema", @@ -186,6 +196,138 @@ def _get_component_spec(task: Mapping[str, Any]) -> Mapping[str, Any] | None: return None +def _undeclared_arguments_allowed() -> bool: + """Return True when the undeclared-argument check is disabled by env var.""" + + value = os.environ.get(ALLOW_UNDECLARED_TASK_ARGUMENTS_ENV, "") + return value.strip().lower() in _TRUTHY_ENV_VALUES + + +def _name_tokens(name: str) -> list[str]: + """Split an input name into lowercase alphanumeric tokens. + + Component authors mix ``snake_case``, ``kebab-case`` and spaces for the + same concept, so tokenizing lets the hint survive separator drift. + """ + + token = "" + tokens: list[str] = [] + for char in name: + if char.isalnum(): + token += char.lower() + elif token: + tokens.append(token) + token = "" + if token: + tokens.append(token) + return tokens + + +def _contains_tokens(haystack: list[str], needle: list[str]) -> bool: + """Return True when needle appears as a contiguous run inside haystack.""" + + if not needle or len(needle) > len(haystack): + return False + return any( + haystack[start : start + len(needle)] == needle + for start in range(len(haystack) - len(needle) + 1) + ) + + +def _nearest_declared_input(name: str, declared_names: Iterable[str]) -> str | None: + """Return the declared input ``name`` most plausibly meant, if any. + + Three deterministic passes over sorted candidates, strongest first: + + 1. separator-insensitive equality (``bq_table`` vs ``bq-table``); + 2. ``difflib`` near-match, which catches ordinary typos; + 3. token containment, which catches qualification renames such as + ``snapshot_date`` -> ``merchant_match_reference_snapshot_date``. + + Semantic renames with no lexical overlap (``wait`` -> ``timeout``) return + None on purpose: a wrong hint is worse than none. + """ + + candidates = sorted(declared_names) + tokens = _name_tokens(name) + if not tokens: + return None + + for candidate in candidates: + if _name_tokens(candidate) == tokens: + return candidate + + close = difflib.get_close_matches(name, candidates, n=1, cutoff=0.8) + if close: + return close[0] + + containing = [ + candidate + for candidate in candidates + if _contains_tokens(_name_tokens(candidate), tokens) + or _contains_tokens(tokens, _name_tokens(candidate)) + ] + if containing: + return min(containing, key=lambda candidate: (len(_name_tokens(candidate)), candidate)) + return None + + +def _suggest_declared_input(name: str, declared_names: Iterable[str]) -> str: + """Return a ``Did you mean ...`` clause, or an empty string when unsure. + + Real-world undeclared arguments are overwhelmingly typos or renames, so a + nearest-match hint is far more actionable than the declared-input list + alone. + """ + + nearest = _nearest_declared_input(name, declared_names) + return f" Did you mean '{nearest}'?" if nearest else "" + + +def _validate_undeclared_arguments( + full_task_name: str, + component_spec: Mapping[str, Any], + component_inputs: Any, + task_arguments: Mapping[str, Any], + declared_names: set[str], +) -> list[str]: + """Return errors for task arguments the component does not declare. + + Deliberately fails OPEN in the cases where the answer is unknowable rather + than wrong: + + * the caller only invokes this when ``_get_component_spec`` resolved a spec, + so unhydrated / digest-pinned / URL-only component refs are skipped; + * a malformed (non-list) ``inputs`` field is skipped, because the declared + set cannot be trusted. + + An empty-but-well-formed ``inputs`` list is NOT skipped: a component that + declares no inputs genuinely accepts no arguments. + + ``TaskSpec`` keeps ``annotations``, ``executionOptions`` and ``isEnabled`` + as siblings of ``arguments`` (see the vendored pipeline schema), so there + are no reserved/metadata keys inside ``arguments`` to allowlist. + """ + + if _undeclared_arguments_allowed(): + return [] + if component_inputs is not None and not isinstance(component_inputs, list): + return [] + + undeclared = sorted(str(name) for name in task_arguments if str(name) not in declared_names) + if not undeclared: + return [] + + component_name = component_spec.get("name") or "" + declared_display = sorted(declared_names) + return [ + f"Task '{full_task_name}': argument '{name}' is not a declared input of " + f"component '{component_name}'.{_suggest_declared_input(name, declared_names)} " + f"Declared inputs: {declared_display}" + for name in undeclared + ] + + def _is_input_required(input_spec: Mapping[str, Any]) -> bool: """Return True when a component input is non-optional and lacks a default.""" @@ -283,7 +425,8 @@ def _validate_task_inputs( ) -> list[str]: """Return component-input wiring errors for one task. - Required component inputs must be present in task arguments. When declared + Required component inputs must be present in task arguments, and supplied + arguments must correspond to a declared component input. When declared inputs are explicitly supplied, graph/task output references are checked regardless of whether the input is required. Nested graph component specs are validated recursively. @@ -309,13 +452,27 @@ def _validate_task_inputs( if not component_spec: return errors - component_inputs = component_spec.get("inputs", []) - if not isinstance(component_inputs, list): - component_inputs = [] + raw_component_inputs = component_spec.get("inputs", []) + component_inputs = raw_component_inputs if isinstance(raw_component_inputs, list) else [] task_arguments = task_spec.get("arguments", {}) or {} if not isinstance(task_arguments, Mapping): task_arguments = {} + declared_names = { + str(spec["name"]) + for spec in component_inputs + if isinstance(spec, Mapping) and spec.get("name") + } + errors.extend( + _validate_undeclared_arguments( + full_task_name, + component_spec, + raw_component_inputs, + task_arguments, + declared_names, + ) + ) + for input_spec in component_inputs: if not isinstance(input_spec, Mapping): continue @@ -344,12 +501,7 @@ def _validate_task_inputs( implementation = component_spec.get("implementation", {}) nested_graph = implementation.get("graph") if isinstance(implementation, Mapping) else None if isinstance(nested_graph, Mapping): - subgraph_inputs = { - str(inp.get("name")) - for inp in component_inputs - if isinstance(inp, Mapping) and inp.get("name") - } - errors.extend(_validate_graph_inputs(nested_graph, subgraph_inputs, f"{full_task_name} > ")) + errors.extend(_validate_graph_inputs(nested_graph, declared_names, f"{full_task_name} > ")) return errors @@ -389,7 +541,10 @@ def _validate_graph_inputs( def validate_component_inputs(pipeline_spec: Mapping[str, Any]) -> list[str]: - """Return required-input and reference-wiring errors for a pipeline. + """Return input-wiring errors for a pipeline. + + Covers missing required inputs, undeclared task arguments, and dangling + graph-input / task-output references. Validation uses embedded component specs when present. If the pipeline has no object-shaped implementation graph, this component-input pass returns no diff --git a/pyproject.toml b/pyproject.toml index d30d701..9c05dcf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tangle-cli" -version = "0.1.12" +version = "0.1.13" description = "CLI for Tangle, the open-source ML pipeline orchestration platform" readme = "README.md" authors = [ diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 345fc66..5bcb1a1 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -183,7 +183,7 @@ def test_tangle_cli_wheel_supports_expert_no_deps_import_path_without_tangle_api requires_dist = [line for line in metadata.splitlines() if line.startswith("Requires-Dist: ")] assert not any(name.startswith("tangle_api/") for name in names) assert "tangle_cli/openapi/openapi.json" not in names - assert "Version: 0.1.12" in metadata + assert "Version: 0.1.13" in metadata assert "Requires-Dist: tangle-api==0.1.1" in requires_dist assert not any("extra == 'native'" in line for line in requires_dist) assert "Provides-Extra: native" in metadata diff --git a/tests/test_pipeline_runs_cli.py b/tests/test_pipeline_runs_cli.py index 165a266..ce86fb5 100644 --- a/tests/test_pipeline_runs_cli.py +++ b/tests/test_pipeline_runs_cli.py @@ -550,8 +550,9 @@ def test_pipeline_runs_submit_dry_run_prints_sanitized_payload(monkeypatch, tmp_ "componentRef": { "name": "text-component", "text": ( - "name: Text Component\n_source_dir: /tmp/private\nimplementation:\n" - " container:\n image: busybox\n" + "name: Text Component\n_source_dir: /tmp/private\n" + "inputs:\n- {name: config, type: JsonObject}\n" + "implementation:\n container:\n image: busybox\n" ), } } diff --git a/tests/test_pipelines_cli.py b/tests/test_pipelines_cli.py index c9f28f7..e026ab0 100644 --- a/tests/test_pipelines_cli.py +++ b/tests/test_pipelines_cli.py @@ -11,6 +11,7 @@ from tangle_cli import cli from tangle_cli.pipeline_hydrator import PipelineHydrator +from tangle_cli.pipeline_validation import ALLOW_UNDECLARED_TASK_ARGUMENTS_ENV from tangle_cli.pipelines import ( _dependency_edges, collect_pipeline_spec_errors, @@ -493,6 +494,215 @@ def test_component_input_validation_rejects_bad_graph_input_ref(): ] +def test_component_input_validation_rejects_undeclared_argument(): + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec(inputs=[{"name": "query", "type": "String"}]) + }, + "arguments": {"query": "shoes", "totally_made_up_arg": "1"}, + } + ) + + assert validate_component_inputs(pipeline) == [ + ( + "Task 'task': argument 'totally_made_up_arg' is not a declared input of " + "component 'Component'. Declared inputs: ['query']" + ) + ] + + +def test_component_input_validation_suggests_nearest_declared_input(): + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec( + inputs=[{"name": "timeout", "type": "Integer", "default": "1000"}] + ) + }, + "arguments": {"timeouts": "1440"}, + } + ) + + assert validate_component_inputs(pipeline) == [ + ( + "Task 'task': argument 'timeouts' is not a declared input of " + "component 'Component'. Did you mean 'timeout'? Declared inputs: ['timeout']" + ) + ] + + +def test_component_input_validation_suggests_qualified_rename(): + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec( + inputs=[ + { + "name": "merchant_match_reference_snapshot_date", + "type": "String", + "optional": True, + } + ] + ) + }, + "arguments": {"snapshot_date": "2026-09-12"}, + } + ) + + assert validate_component_inputs(pipeline) == [ + ( + "Task 'task': argument 'snapshot_date' is not a declared input of " + "component 'Component'. Did you mean " + "'merchant_match_reference_snapshot_date'? Declared inputs: " + "['merchant_match_reference_snapshot_date']" + ) + ] + + +def test_component_input_validation_suggests_across_separator_drift(): + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec(inputs=[{"name": "bq-table", "type": "String"}]) + }, + "arguments": {"bq_table": "t"}, + } + ) + + errors = validate_component_inputs(pipeline) + + assert "Did you mean 'bq-table'?" in errors[0] + + +def test_component_input_validation_omits_hint_for_semantic_rename(): + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec( + inputs=[{"name": "timeout", "type": "Integer", "default": "1000"}] + ) + }, + "arguments": {"wait": "1440"}, + } + ) + + assert validate_component_inputs(pipeline) == [ + ( + "Task 'task': argument 'wait' is not a declared input of " + "component 'Component'. Declared inputs: ['timeout']" + ) + ] + + +def test_component_input_validation_accepts_fully_declared_arguments(): + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec( + inputs=[ + {"name": "query", "type": "String"}, + {"name": "timeout", "type": "Integer", "default": "1000"}, + ] + ) + }, + "arguments": {"query": "shoes", "timeout": "1440"}, + } + ) + + assert validate_component_inputs(pipeline) == [] + + +@pytest.mark.parametrize( + "component_ref", + [ + {"url": "resolve://components.resolve.yaml#Component"}, + {"name": "Component", "digest": "879a8ddf"}, + {"text": "not: [valid"}, + ], + ids=["url-only", "digest-pinned", "unparsable-text"], +) +def test_component_input_validation_skips_unresolvable_component_spec(component_ref): + pipeline = _single_task_pipeline( + { + "componentRef": component_ref, + "arguments": {"totally_made_up_arg": "1"}, + } + ) + + assert validate_component_inputs(pipeline) == [] + + +def test_component_input_validation_skips_undeclared_check_on_malformed_inputs(): + spec = _component_spec() + spec["inputs"] = "not-a-list" + pipeline = _single_task_pipeline( + {"componentRef": {"spec": spec}, "arguments": {"anything": "1"}} + ) + + assert validate_component_inputs(pipeline) == [] + + +def test_component_input_validation_rejects_arguments_to_input_less_component(): + pipeline = _single_task_pipeline( + {"componentRef": {"spec": _component_spec()}, "arguments": {"anything": "1"}} + ) + + assert validate_component_inputs(pipeline) == [ + ( + "Task 'task': argument 'anything' is not a declared input of " + "component 'Component'. Declared inputs: []" + ) + ] + + +def test_component_input_validation_flags_undeclared_arguments_in_nested_graphs(): + inner = _component_spec( + inputs=[{"name": "query", "type": "String"}], + implementation={ + "graph": { + "tasks": { + "inner": { + "componentRef": { + "spec": _component_spec( + inputs=[{"name": "query", "type": "String"}] + ) + }, + "arguments": { + "query": {"graphInput": {"inputName": "query"}}, + "snapshot_date": "2026-09-12", + }, + } + } + } + }, + ) + pipeline = _single_task_pipeline( + {"componentRef": {"spec": inner}, "arguments": {"query": "shoes"}} + ) + + assert validate_component_inputs(pipeline) == [ + ( + "Task 'task > inner': argument 'snapshot_date' is not a declared input of " + "component 'Component'. Declared inputs: ['query']" + ) + ] + + +def test_component_input_validation_undeclared_check_can_be_disabled(monkeypatch): + monkeypatch.setenv(ALLOW_UNDECLARED_TASK_ARGUMENTS_ENV, "1") + pipeline = _single_task_pipeline( + { + "componentRef": { + "spec": _component_spec(inputs=[{"name": "query", "type": "String"}]) + }, + "arguments": {"query": "shoes", "totally_made_up_arg": "1"}, + } + ) + + assert validate_component_inputs(pipeline) == [] + + def test_collect_pipeline_errors_combines_shape_schema_and_input_wiring(): pipeline = _single_task_pipeline( { diff --git a/uv.lock b/uv.lock index c495089..cf7bae0 100644 --- a/uv.lock +++ b/uv.lock @@ -2083,7 +2083,7 @@ requires-dist = [{ name = "pydantic", specifier = ">=2.0" }] [[package]] name = "tangle-cli" -version = "0.1.12" +version = "0.1.13" source = { editable = "." } dependencies = [ { name = "cloud-pipelines" },