diff --git a/src/orchestrator/__init__.py b/src/orchestrator/__init__.py index bcca685..4d13fc1 100644 --- a/src/orchestrator/__init__.py +++ b/src/orchestrator/__init__.py @@ -89,6 +89,11 @@ "create_pipeline_api": ".api", "create_advanced_pipeline_compiler": ".api", "create_pipeline_executor": ".api", + # --- Validation, with its findings --- + "validate_pipeline_file": ".validation.pipeline_report", + "validate_pipeline_text": ".validation.pipeline_report", + "PipelineValidation": ".validation.pipeline_report", + "Finding": ".validation.pipeline_report", # --- Error hierarchy --- "OrchestratorError": ".core.exceptions", "PipelineError": ".core.exceptions", diff --git a/src/orchestrator/cli.py b/src/orchestrator/cli.py index cc4ad2c..d155d56 100644 --- a/src/orchestrator/cli.py +++ b/src/orchestrator/cli.py @@ -367,46 +367,6 @@ def _failed_steps(results) -> list: return failed -def _reportable_issues(compiler): - """Findings a successful validation should still tell the user about. - - The compiler has always collected these; nothing displayed them. A - pipeline could report `✓ is valid` while carrying a warning saying a - reference could not be checked -- and then fail at run time on exactly - that reference (#465). A warning that only reaches the log stream is not - a warning: `orchestrator validate` prints to stdout, and a script - capturing stdout saw nothing at all. - - Informational findings are excluded: "tool is available" and "execution - order computed" are not things anyone needs told. - """ - report = getattr(compiler, "validation_report", None) - if report is None: - return [] - return [issue for issue in report.issues if issue.is_error or issue.is_warning] - - -def _issue_payload(issue): - """One finding as stable, machine-readable fields. - - Field names are part of the interface: anything reading this should not - have to parse the human message to learn which step or reference is at - fault. - """ - metadata = issue.metadata or {} - return { - "code": issue.code, - "severity": issue.severity.value, - "category": issue.category, - "step": metadata.get("step", issue.component), - "parameter_path": metadata.get("parameter_path", issue.path), - "referenced_step": metadata.get("referenced_step"), - "referenced_field": metadata.get("referenced_field"), - "message": issue.message, - "suggestions": list(issue.suggestions or []), - } - - @cli.command("validate") @click.argument("pipeline_file", type=click.Path(exists=True, dir_okay=False)) @click.option("--json", "as_json", is_flag=True, @@ -414,72 +374,61 @@ def _issue_payload(issue): @verbose_option def validate_pipeline(pipeline_file: str, verbose: bool, as_json: bool): """Compile a pipeline without running it, and report its task graph.""" - import asyncio import json as _json if verbose: setup_logging(verbose=True) - from .compiler.yaml_compiler import YAMLCompiler - - compiler = YAMLCompiler() - - async def _compile(): - with open(pipeline_file) as fh: - return await compiler.compile(fh.read(), {}) + from .validation.pipeline_report import validate_pipeline_file try: - pipeline = asyncio.run(_compile()) + result = validate_pipeline_file(pipeline_file) except KeyboardInterrupt: click.echo("Interrupted.", err=True) sys.exit(EXIT_INTERRUPTED) - except Exception as exc: + + if not result.valid: if as_json: click.echo(_json.dumps({ "pipeline_file": pipeline_file, "valid": False, - "error": f"{type(exc).__name__}: {exc}", - "findings": [_issue_payload(i) for i in _reportable_issues(compiler)], + "error": result.error, + "findings": [f.as_dict() for f in result.findings], }, indent=2)) else: - click.echo(f"{type(exc).__name__}: {exc}", err=True) + click.echo(result.error, err=True) sys.exit(EXIT_VALIDATION_ERROR) - tasks = getattr(pipeline, "tasks", {}) or {} - findings = _reportable_issues(compiler) + tasks = result.tasks + findings = result.findings if as_json: click.echo(_json.dumps({ "pipeline_file": pipeline_file, "valid": True, - "pipeline": getattr(pipeline, "id", None), - "tasks": { - task_id: list(getattr(task, "dependencies", []) or []) - for task_id, task in tasks.items() - }, - "findings": [_issue_payload(issue) for issue in findings], + "pipeline": result.pipeline_id, + "tasks": tasks, + "findings": [f.as_dict() for f in findings], }, indent=2)) sys.exit(EXIT_OK) - warnings = [issue for issue in findings if issue.is_warning] + warnings = result.warnings summary = f"✓ {pipeline_file} is valid" if warnings: summary += f" with {len(warnings)} warning{'s' if len(warnings) > 1 else ''}" click.echo(summary) - click.echo(f" pipeline: {getattr(pipeline, 'id', '')}") + click.echo(f" pipeline: {result.pipeline_id or ''}") click.echo(f" tasks: {len(tasks)}") - for task_id in tasks: - task = tasks[task_id] - deps = getattr(task, "dependencies", []) or [] + for task_id, deps in tasks.items(): suffix = f" <- {', '.join(deps)}" if deps else "" click.echo(f" - {task_id}{suffix}") for issue in findings: click.echo("") - where = issue.path or issue.component - click.echo(f"{issue.severity.value}[{issue.code}] {where}:") + where = issue.parameter_path or issue.step + click.echo(f"{issue.severity}[{issue.code}] {where}:") click.echo(f" {issue.message}") - for suggestion in issue.suggestions or []: + for suggestion in issue.suggestions: click.echo(f" suggestion: {suggestion}") sys.exit(EXIT_OK) diff --git a/src/orchestrator/core/step_fields.py b/src/orchestrator/core/step_fields.py index 76cd01f..61df6c9 100644 --- a/src/orchestrator/core/step_fields.py +++ b/src/orchestrator/core/step_fields.py @@ -75,3 +75,20 @@ "dependencies", # already read, with the `declared` origin "depends_on", }) + +#: The same distinction one level up. A pipeline's own `name` and +#: `description` are prose about the pipeline; nothing renders them either, so +#: `name: "{{ nosuch }}"` at the top of a document was failing validation for a +#: pipeline that runs. +#: +#: Deliberately short. `outputs` *is* rendered; `parameters` declares names +#: rather than using them; and `id` and `version` are schema-constrained -- +#: `version` must match `\d+\.\d+\.\d+`, so a template there is a real error +#: and calling the field inert would describe it wrongly. Only fields a real +#: run tolerates an unresolvable reference in are listed, which is what +#: `test_a_pipeline_with_templates_in_prose_still_runs` checks. +INERT_PIPELINE_FIELDS: FrozenSet[str] = frozenset({ + "name", + "description", + "metadata", +}) diff --git a/src/orchestrator/validation/data_flow_validator.py b/src/orchestrator/validation/data_flow_validator.py index fd3fbfd..9efdca5 100644 --- a/src/orchestrator/validation/data_flow_validator.py +++ b/src/orchestrator/validation/data_flow_validator.py @@ -615,6 +615,14 @@ def _suggest_similar_names(self, target: str, available: List[str]) -> List[str] """Suggest similar names for typos.""" suggestions = [] target_lower = target.lower() + + # Sorted, because callers pass names collected in a set and this + # function both emits in iteration order and truncates to three. Two + # identical `orchestrator validate` runs offered the same three + # suggestions in different orders, and on a longer candidate list + # would have offered *different* three. The tiers below still rank the + # result; this only makes ties break the same way every time. + available = sorted(available) # Exact match (case insensitive) for name in available: diff --git a/src/orchestrator/validation/pipeline_report.py b/src/orchestrator/validation/pipeline_report.py new file mode 100644 index 0000000..b35562d --- /dev/null +++ b/src/orchestrator/validation/pipeline_report.py @@ -0,0 +1,169 @@ +"""Validating a pipeline, with its findings, from Python. + +`orchestrator validate` has reported structured findings since #467. The +Python API had no equivalent: `PipelineAPI.validate_yaml` returns a bare +`bool`, so a caller embedding the orchestrator could learn *that* a pipeline +was rejected and nothing about why -- and could not see warnings at all, which +is where "this reference could not be checked" lives. + +The findings were not missing, only private: `cli._reportable_issues` and +`cli._issue_payload` built them for the JSON output and nothing else could +reach them. A second implementation for the API would be a second thing to +drift; this module is the one implementation, and the CLI formats what it +returns. + +The payload field names are interface. Anything consuming a finding should not +have to parse the human message to learn which step or reference is at fault. +""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple, Union + + +@dataclass(frozen=True) +class Finding: + """One validation finding, in stable machine-readable fields.""" + + code: str + severity: str + category: Optional[str] = None + step: Optional[str] = None + parameter_path: Optional[str] = None + referenced_step: Optional[str] = None + referenced_field: Optional[str] = None + message: str = "" + suggestions: Tuple[str, ...] = () + + @property + def is_error(self) -> bool: + return self.severity == "error" + + @property + def is_warning(self) -> bool: + return self.severity == "warning" + + def as_dict(self) -> Dict[str, Any]: + return { + "code": self.code, + "severity": self.severity, + "category": self.category, + "step": self.step, + "parameter_path": self.parameter_path, + "referenced_step": self.referenced_step, + "referenced_field": self.referenced_field, + "message": self.message, + "suggestions": list(self.suggestions), + } + + +@dataclass(frozen=True) +class PipelineValidation: + """The result of validating one pipeline document.""" + + valid: bool + findings: Tuple[Finding, ...] = () + pipeline_id: Optional[str] = None + #: Step id -> its dependencies, so a caller can check the graph the + #: compiler actually built rather than re-deriving it. + tasks: Dict[str, List[str]] = field(default_factory=dict) + #: The exception that stopped compilation, if one did. + error: Optional[str] = None + + @property + def errors(self) -> Tuple[Finding, ...]: + return tuple(f for f in self.findings if f.is_error) + + @property + def warnings(self) -> Tuple[Finding, ...]: + return tuple(f for f in self.findings if f.is_warning) + + def as_dict(self) -> Dict[str, Any]: + return { + "valid": self.valid, + "pipeline": self.pipeline_id, + "tasks": self.tasks, + "findings": [f.as_dict() for f in self.findings], + **({"error": self.error} if self.error is not None else {}), + } + + +def _findings_from(compiler) -> Tuple[Finding, ...]: + """Findings a successful validation should still tell the caller about. + + The compiler has always collected these; nothing displayed them. A + pipeline could report valid while carrying a warning saying a reference + could not be checked -- and then fail at run time on exactly that + reference (#465). + + Informational findings are excluded: "tool is available" and "execution + order computed" are not things anyone needs told. + """ + report = getattr(compiler, "validation_report", None) + if report is None: + return () + return tuple( + _finding(issue) + for issue in report.issues + if issue.is_error or issue.is_warning + ) + + +def _finding(issue) -> Finding: + metadata = issue.metadata or {} + severity = issue.severity + return Finding( + code=issue.code, + severity=getattr(severity, "value", severity), + category=issue.category, + step=metadata.get("step", issue.component), + parameter_path=metadata.get("parameter_path", issue.path), + referenced_step=metadata.get("referenced_step"), + referenced_field=metadata.get("referenced_field"), + message=issue.message, + suggestions=tuple(issue.suggestions or ()), + ) + + +async def validate_pipeline_text(yaml_text: str) -> PipelineValidation: + """Compile `yaml_text` without running it and report what was found. + + Compilation failure is a result, not an exception: a caller validating + user input wants the findings alongside the failure, and raising would + discard them. + """ + from ..compiler.yaml_compiler import YAMLCompiler + + compiler = YAMLCompiler() + try: + pipeline = await compiler.compile(yaml_text, {}) + except Exception as exc: + return PipelineValidation( + valid=False, + findings=_findings_from(compiler), + error=f"{type(exc).__name__}: {exc}", + ) + + tasks = getattr(pipeline, "tasks", {}) or {} + return PipelineValidation( + valid=True, + findings=_findings_from(compiler), + pipeline_id=getattr(pipeline, "id", None), + tasks={ + task_id: list(getattr(task, "dependencies", []) or []) + for task_id, task in tasks.items() + }, + ) + + +def validate_pipeline_file(path: Union[str, Path]) -> PipelineValidation: + """`validate_pipeline_text` for a file on disk, without an event loop. + + Provided because the common case -- checking a pipeline someone wrote -- + is synchronous, and requiring callers to build an event loop to learn + whether a file is valid is why the boolean survived so long. + """ + return asyncio.run(validate_pipeline_text(Path(path).read_text())) diff --git a/src/orchestrator/validation/template_validator.py b/src/orchestrator/validation/template_validator.py index 791b381..3f79671 100644 --- a/src/orchestrator/validation/template_validator.py +++ b/src/orchestrator/validation/template_validator.py @@ -19,6 +19,7 @@ from ..core.runtime_context import BARE_RUNTIME_NAMES, RUNTIME_NAMESPACE from ..core.loop_contracts import ALL_BINDINGS, LoopContract, contracts_for +from ..core.step_fields import INERT_PIPELINE_FIELDS, INERT_STEP_FIELDS from ..core.template_globals import ( ALL_LOOP_VARIABLES, DOLLAR_LOOP_VARIABLES, @@ -39,6 +40,17 @@ #: this is where a loop construct is looked for -- see `_validate_object_templates`. _STEP_PATH = re.compile(r"steps\[\d+\]$") +#: What a template in an inert field actually does. Validation used to report +#: `{{ b.result }}` in a step's `name:` as "references step results - will be +#: resolved at runtime", which is the opposite of true: nothing renders `name`, +#: so the braces reach the log verbatim. Worse, `{{ nosuch }}` in a +#: `description:` was a hard error, so a stray brace in prose rejected a +#: pipeline that runs correctly. +_INERT_TEMPLATE_MESSAGE = ( + "'{field}' is copied verbatim, so this template is never rendered -- " + "the braces appear literally in the output" +) + def _binding_set(value: Union[bool, FrozenSet[str], None]) -> FrozenSet[str]: """Normalise the loop-scope argument to a set of names. @@ -399,9 +411,16 @@ def _validate_variables( if loop_var in template ] - # Combine both sets of variables - all_var_names = set(var_names) | set(loop_var_matches) - + # Combine both sets of variables. + # + # Sorted, because findings are emitted in this order and a set of + # strings iterates by hash. Two identical `orchestrator validate` + # runs on the same file produced the same 44 findings in different + # orders, so any caller diffing runs, pinning output, or reporting + # "the first problem" saw noise. `PYTHONHASHSEED` differs per + # process, which is why the in-process check missed it. + all_var_names = sorted(set(var_names) | set(loop_var_matches)) + for var_name in all_var_names: used_variables.add(var_name) @@ -605,6 +624,7 @@ def _validate_object_templates( undefined_variables: Set, loop_bindings: FrozenSet[str] = frozenset(), loop_scope: Optional[Tuple[LoopContract, str, FrozenSet[str]]] = None, + inert_field: Optional[str] = None, ): """Recursively validate templates in an object. @@ -613,10 +633,29 @@ def _validate_object_templates( path reached inside it, because scope is a property of the field and not of the step: a `create_parallel_queue`'s `on` resolves before any item exists while the action list beside it runs per item. + + `inert_field` names the step field the runtime copies verbatim, if + this walk is inside one. Nothing substitutes into it, so a reference + there cannot be undefined and cannot be resolved later. The *field* is + carried rather than a flag so a nested value reports `metadata` -- the + field that is inert -- instead of whichever key it sits under. """ if isinstance(obj, str): # Check if this contains templates if '{{' in obj or '{%' in obj: + if inert_field: + warnings.append(TemplateValidationError( + template=obj, + error_type="inert_field_template", + message=_INERT_TEMPLATE_MESSAGE.format(field=inert_field), + context_path=path, + severity="warning", + suggestions=[ + "Move the reference to a field that is rendered " + "(parameters, action, location), or remove the braces" + ], + )) + return result = self.validate_template( obj, context, path, step_ids, loop_bindings ) @@ -631,7 +670,8 @@ def _validate_object_templates( # `action_loop` look like a second, separate loop, which replaced # the queue's scope with the action loop's and let `{{ item }}` # through in the `on` expression that generates the queue. - declared = contracts_for(obj) if _STEP_PATH.search(path) else () + is_step = bool(_STEP_PATH.search(path)) + declared = contracts_for(obj) if is_step else () if len(declared) > 1: # Which construct wins is decided by declaration order in # `loop_contracts`, and no engine agreed to that order. The @@ -659,6 +699,11 @@ def _validate_object_templates( for key, value in obj.items(): new_path = f"{path}.{key}" if path else key + inert_here = ( + (is_step and key in INERT_STEP_FIELDS) + or (not path and key in INERT_PIPELINE_FIELDS) + ) + child_inert = inert_field or (key if inert_here else None) child_bindings, child_scope = loop_bindings, loop_scope if loop_scope is not None: contract, prefix, enclosing = loop_scope @@ -668,7 +713,7 @@ def _validate_object_templates( self._validate_object_templates( value, context, step_ids, new_path, errors, warnings, used_variables, undefined_variables, - child_bindings, child_scope, + child_bindings, child_scope, child_inert, ) elif isinstance(obj, list): @@ -677,7 +722,7 @@ def _validate_object_templates( self._validate_object_templates( item, context, step_ids, new_path, errors, warnings, used_variables, undefined_variables, - loop_bindings, loop_scope, + loop_bindings, loop_scope, inert_field, ) def _is_step_result_reference(self, var_name: str, step_ids: List[str]) -> bool: diff --git a/tests/test_inert_field_diagnostics.py b/tests/test_inert_field_diagnostics.py new file mode 100644 index 0000000..11e6963 --- /dev/null +++ b/tests/test_inert_field_diagnostics.py @@ -0,0 +1,152 @@ +"""A template in a field nothing renders is not a runtime promise. + +`core.step_fields` established which parts of a step the runtime renders, and +#471 wired it into dependency inference so two inert strings stopped inventing +a cycle. Template validation was still reading those same fields as if they +resolved, and got both directions wrong: + + - id: a + name: "{{ b.result }}" # "will be resolved at runtime" -- it is not + description: "{{ nosuch }}" # a hard error, in prose nobody renders + +The first is a false promise: nothing substitutes into `name`, so the braces +reach the log verbatim and the reader is told the opposite. The second is a +false rejection: a stray brace in a description failed a pipeline that runs +correctly, which is the class of bug #465, #469 and #472 each removed +elsewhere. + +The run in `test_a_pipeline_with_templates_in_prose_still_runs` is the check +that this is a real property of the runtime and not a claim about it. +""" + +import asyncio +from pathlib import Path + +import pytest + +from orchestrator.core.step_fields import ( + INERT_STEP_FIELDS, + RENDERABLE_STEP_FIELDS, +) +from orchestrator.validation.template_validator import TemplateValidator +from tests.test_infrastructure import create_test_orchestrator + +pytestmark = [pytest.mark.contract] + +#: `id` names the step and `dependencies`/`depends_on` hold step ids. They are +#: inert for the same reason, but a template in one is a structural error +#: rather than prose, so they are not probed as free text here. +PROSE_FIELDS = sorted(INERT_STEP_FIELDS - {"id", "dependencies", "depends_on", "tool"}) + + +def _validate(step, context=None): + return TemplateValidator().validate_pipeline_templates( + {"id": "p", "steps": [step, {"id": "b", "parameters": {}}]}, + context or {}, + ) + + +@pytest.mark.parametrize("field", PROSE_FIELDS) +def test_an_undefined_name_in_an_inert_field_is_not_an_error(field): + """It cannot be undefined: it is never looked up.""" + result = _validate({"id": "a", field: "{{ nosuch_variable }}"}) + assert result.is_valid, [ + (e.error_type, e.context_path, e.message) for e in result.errors + ] + + +@pytest.mark.parametrize("field", PROSE_FIELDS) +def test_an_inert_field_still_warns(field): + """Silence would be wrong too -- the author wrote a template and will get + braces. The warning is how they find out before reading the output.""" + result = _validate({"id": "a", field: "{{ b.result }}"}) + kinds = [w.error_type for w in result.warnings] + assert "inert_field_template" in kinds, kinds + + +def test_the_warning_does_not_claim_the_value_arrives_later(): + """The old message said `will be resolved at runtime`, which is the one + thing that does not happen.""" + result = _validate({"id": "a", "name": "{{ b.result }}"}) + inert = [w for w in result.warnings if w.error_type == "inert_field_template"] + assert inert, [w.error_type for w in result.warnings] + assert "resolved at runtime" not in inert[0].message + assert "never rendered" in inert[0].message + + +def test_the_warning_names_the_inert_field_not_the_key_beneath_it(): + """`metadata.note` is inert because `metadata` is. Naming `note` would + send the reader looking for a rule about a key they invented.""" + result = _validate({"id": "a", "metadata": {"note": "{{ b.result }}"}}) + inert = [w for w in result.warnings if w.error_type == "inert_field_template"] + assert inert and "'metadata'" in inert[0].message, [w.message for w in inert] + assert inert[0].context_path == "steps[0].metadata.note", inert[0].context_path + + +@pytest.mark.parametrize("field", RENDERABLE_STEP_FIELDS) +def test_a_renderable_field_still_reports_an_undefined_name(field): + """The suppression must not spread. These fields do resolve, so a name + that is not there is still an error.""" + value = {"x": "{{ nosuch_variable }}"} if field == "parameters" else "{{ nosuch_variable }}" + result = _validate({"id": "a", field: value}) + assert not result.is_valid, f"{field} is rendered; an undefined name there is an error" + assert "undefined_variable" in [e.error_type for e in result.errors] + + +def test_a_step_result_reference_in_a_rendered_field_is_still_a_runtime_promise(): + result = _validate({"id": "a", "parameters": {"x": "{{ b.result }}"}}) + assert "runtime_variable" in [w.error_type for w in result.warnings] + + +@pytest.mark.e2e +def test_a_pipeline_with_templates_in_prose_still_runs(tmp_path): + """The evidence that these fields are inert, rather than the assertion. + + Every prose field carries a reference to a name that exists nowhere. If + any of them were rendered the run would fail or write the wrong thing; the + file that lands proves it did neither. + """ + pipeline = f""" +id: inert_prose +name: "{{{{ nosuch_pipeline_name }}}}" +description: "{{{{ nosuch_description }}}}" +metadata: + owner: "{{{{ nosuch_owner }}}}" +steps: + - id: write_it + name: "{{{{ nosuch_step_name }}}}" + description: "{{{{ nosuch_step_description }}}}" + metadata: + note: "{{{{ nosuch_metadata }}}}" + tool: filesystem + action: write + parameters: + path: "{tmp_path}/out.txt" + content: "written" +""" + asyncio.run(create_test_orchestrator().execute_yaml(pipeline, {})) + written = Path(tmp_path, "out.txt") + assert written.exists(), "the step did not run" + assert written.read_text() == "written" + + +@pytest.mark.e2e +def test_that_same_pipeline_validates(tmp_path): + """Both surfaces agree: what runs, validates.""" + result = TemplateValidator().validate_pipeline_templates( + { + "id": "inert_prose", + "name": "{{ nosuch_pipeline_name }}", + "steps": [{ + "id": "write_it", + "name": "{{ nosuch_step_name }}", + "description": "{{ nosuch_step_description }}", + "metadata": {"note": "{{ nosuch_metadata }}"}, + "tool": "filesystem", + "action": "write", + "parameters": {"path": str(tmp_path / "out.txt"), "content": "written"}, + }], + }, + {}, + ) + assert result.is_valid, [(e.error_type, e.context_path) for e in result.errors] diff --git a/tests/test_validation_api.py b/tests/test_validation_api.py new file mode 100644 index 0000000..67b405a --- /dev/null +++ b/tests/test_validation_api.py @@ -0,0 +1,182 @@ +"""Validation findings reach a Python caller, not only the terminal. + +`orchestrator validate --json` has emitted structured findings since #467. The +Python API had `PipelineAPI.validate_yaml`, which returns a bare `bool`: a +caller embedding the orchestrator could learn *that* a document was rejected +and nothing about why, and could not see warnings at all -- which is where +"this reference could not be checked" lives, the warning that precedes the +run-time failure in #465. + +The findings existed; they were private to `cli.py`. So the risk in exposing +them was building a *second* implementation that drifts from the one the CLI +prints, which is the class of bug #466 removed for dependencies. There is one +implementation, and `test_the_cli_and_the_api_report_the_same_findings` is +what holds it to that. +""" + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +from orchestrator.validation.pipeline_report import ( + Finding, + validate_pipeline_file, + validate_pipeline_text, +) + +pytestmark = [pytest.mark.contract] + +REPO = Path(__file__).resolve().parent.parent +SUPPORTED = REPO / "examples" / "supported" / "01_hello_filesystem.yaml" + +#: A document that validates *and* carries findings. Every assertion about +#: findings needs one: the hermetic example above produces none, so a suite +#: written only against it passes just as happily when warnings are dropped +#: on the floor -- which is what the first version of this file did. +WITH_FINDINGS = REPO / "examples" / "auto_tags_demo.yaml" + +#: The payload's field names are interface: a consumer must not have to parse +#: the human message to learn which step or reference is at fault. Asserted as +#: an exact set so removing or renaming one is a deliberate act. +FINDING_FIELDS = { + "code", "severity", "category", "step", "parameter_path", + "referenced_step", "referenced_field", "message", "suggestions", +} + + +def test_a_valid_pipeline_reports_its_graph(): + result = validate_pipeline_file(SUPPORTED) + assert result.valid, result.error + assert result.pipeline_id == "hello_filesystem" + assert result.tasks["read_back"] == ["write_greeting"], result.tasks + assert result.error is None + + +def test_a_broken_pipeline_is_a_result_not_an_exception(): + """A caller validating user input wants the findings alongside the + failure. Raising would discard them at the moment they are most useful.""" + import asyncio + + result = asyncio.run(validate_pipeline_text("id: broken\nsteps: [[[[")) + assert not result.valid + assert result.error, "the failure must say what happened" + + +def test_warnings_reach_a_python_caller(): + """The reported gap. `validate_yaml` returns a bool, so a warning saying + a reference could not be checked -- the one that precedes the run-time + failure in #465 -- was invisible to anything but the terminal.""" + result = validate_pipeline_file(WITH_FINDINGS) + assert result.valid, result.error + assert result.warnings, "this document warns; the API reported nothing" + assert any( + "will be resolved at runtime" in w.message for w in result.warnings + ), [w.message for w in result.warnings] + + +def test_a_warning_does_not_make_a_document_invalid(): + """Warnings inform; they do not block. Conflating them would reject + documents that run.""" + assert validate_pipeline_file(WITH_FINDINGS).valid + + +def test_findings_carry_stable_named_fields(): + result = validate_pipeline_file(WITH_FINDINGS) + assert result.findings, "nothing to check the field names of" + for finding in result.findings: + assert set(finding.as_dict()) == FINDING_FIELDS, finding.as_dict() + + +def test_errors_and_warnings_are_separable(): + """A caller that wants to block on errors and log warnings should not have + to string-match a severity out of a message.""" + findings = ( + Finding(code="a", severity="error", message="x"), + Finding(code="b", severity="warning", message="y"), + ) + from orchestrator.validation.pipeline_report import PipelineValidation + + result = PipelineValidation(valid=False, findings=findings) + assert [f.code for f in result.errors] == ["a"] + assert [f.code for f in result.warnings] == ["b"] + assert result.errors[0].is_error and result.warnings[0].is_warning + + +def test_the_same_document_yields_the_same_findings_in_the_same_order(): + """Order is part of the contract: a caller diffing two runs, or a test + pinning output, cannot use a set that reshuffles.""" + first = validate_pipeline_file(WITH_FINDINGS) + second = validate_pipeline_file(WITH_FINDINGS) + assert first.findings, "an empty list is trivially stable" + assert [f.as_dict() for f in first.findings] == [ + f.as_dict() for f in second.findings + ] + + +@pytest.mark.e2e +def test_findings_are_stable_across_processes(): + """The check above cannot catch this on its own. + + Findings were emitted while iterating sets, and `PYTHONHASHSEED` is fixed + for the life of a process -- so two calls in one interpreter agreed while + two `orchestrator validate` runs produced the same 44 findings in + different orders, and offered the same three "did you mean" suggestions + permuted. On a longer candidate list, truncating to three would have + offered *different* suggestions run to run. + """ + digests = {_cli_findings_digest() for _ in range(3)} + assert len(digests) == 1, "validate is not reproducible run to run" + + +def _cli_findings_digest() -> str: + import hashlib + + proc = subprocess.run( + [sys.executable, "-m", "orchestrator.cli", "validate", "--json", str(WITH_FINDINGS)], + cwd=str(REPO), env=_cli_env(), capture_output=True, text=True, timeout=300, + ) + assert proc.returncode == 0, proc.stdout[-800:] + proc.stderr[-800:] + findings = json.loads(proc.stdout)["findings"] + assert findings, "nothing to be stable about" + return hashlib.sha256(json.dumps(findings).encode()).hexdigest() + + +def _cli_env() -> dict: + env = dict(os.environ) + env["PYTHONPATH"] = str(REPO / "src") + os.pathsep + env.get("PYTHONPATH", "") + env["ORCHESTRATOR_AUTO_INSTALL"] = "0" + return env + + +@pytest.mark.e2e +def test_the_cli_and_the_api_report_the_same_findings(): + """The guard against a second implementation. + + The CLI's findings were private helpers in `cli.py`; exposing them to + Python by reimplementing would have produced two things to drift apart. + """ + proc = subprocess.run( + [sys.executable, "-m", "orchestrator.cli", "validate", "--json", str(WITH_FINDINGS)], + cwd=str(REPO), env=_cli_env(), capture_output=True, text=True, timeout=300, + ) + assert proc.returncode == 0, proc.stdout[-800:] + proc.stderr[-800:] + from_cli = json.loads(proc.stdout) + + from_api = validate_pipeline_file(WITH_FINDINGS) + assert from_api.findings, "comparing two empty lists proves nothing" + assert from_cli["valid"] == from_api.valid + assert from_cli["pipeline"] == from_api.pipeline_id + assert from_cli["tasks"] == from_api.tasks + assert from_cli["findings"] == [f.as_dict() for f in from_api.findings] + + +def test_the_package_exports_them(): + """A caller should not have to know which module they live in.""" + import orchestrator + + assert orchestrator.validate_pipeline_file is validate_pipeline_file + assert orchestrator.Finding is Finding