From 7381c4033790d109a57f13e9fdc29a7fe1a8d049 Mon Sep 17 00:00:00 2001 From: Volv G Date: Tue, 29 Sep 2026 18:18:17 -0700 Subject: [PATCH] Add `pipelines dehydrate` with verified, portable component refs (v0.1.28) Add a first-class `tangle sdk pipelines dehydrate PIPELINE -o OUTPUT [--mode auto|digest|name|url|file] [--components-dir DIR]` command backed by the shared PipelineDehydrator, and fix correctness bugs it exposed. - A published digest is emitted only when its own spec semantically matches the inline component, never merely because it exists. - Explicit DIGEST/NAME write a per-output resolve config with an ordered [verified primary, local copy] fallback; NAME pins the inspected owner. - A `fallback_on_error` resolve-config marker lets a generated primary fall through to its local copy; unmarked configs are unchanged. - Every mode fully dehydrates nested graphs. - Extracted subgraph files are content-addressed, fixing silent cross-output overwrites from per-run counter names. - Per-call extraction state resets, so reused instances cannot leak one output's bundle into another. --- README.md | 40 ++ .../tangle-cli/src/tangle_cli/__init__.py | 2 +- .../src/tangle_cli/pipeline_dehydrator.py | 322 +++++++++++--- .../src/tangle_cli/pipeline_hydrator.py | 21 +- .../tangle-cli/src/tangle_cli/pipelines.py | 101 +++++ .../src/tangle_cli/pipelines_cli.py | 92 +++- pyproject.toml | 2 +- tests/test_dehydrator_portable_refs.py | 418 ++++++++++++++++++ tests/test_dehydrator_subgraph_addressing.py | 163 +++++++ tests/test_packaging.py | 2 +- tests/test_pipeline_dehydrator.py | 48 +- tests/test_pipeline_runs_cli.py | 3 +- tests/test_pipelines_dehydrate_cli.py | 341 ++++++++++++++ uv.lock | 2 +- 14 files changed, 1488 insertions(+), 69 deletions(-) create mode 100644 tests/test_dehydrator_portable_refs.py create mode 100644 tests/test_dehydrator_subgraph_addressing.py create mode 100644 tests/test_pipelines_dehydrate_cli.py diff --git a/README.md b/README.md index 9af821c..50db428 100644 --- a/README.md +++ b/README.md @@ -168,6 +168,7 @@ uv run tangle sdk pipelines validate pipeline.yaml uv run tangle sdk pipelines diagram pipeline.yaml uv run tangle sdk pipelines layout pipeline.yaml --recursive uv run tangle sdk pipelines hydrate pipeline.yaml --output hydrated.yaml +uv run tangle sdk pipelines dehydrate hydrated.yaml --output pipeline.yaml uv run tangle sdk components generate from-python path/to/component.py --image python:3.12 uv run tangle sdk components bump-version path/to/component.yaml ``` @@ -488,10 +489,25 @@ Local pipeline commands live under `sdk pipelines`: ```bash uv run tangle sdk pipelines validate pipeline.yaml uv run tangle sdk pipelines hydrate pipeline.yaml --output hydrated.yaml +uv run tangle sdk pipelines dehydrate hydrated.yaml --output pipeline.yaml uv run tangle sdk pipelines diagram pipeline.yaml uv run tangle sdk pipelines layout pipeline.yaml --recursive ``` +`pipelines dehydrate` is the inverse of `hydrate`. It replaces inline component specs with references, using the same noninteractive dehydrator as `pipeline-runs export --dehydrate`, and a single `--mode` applies to every component: + +| `--mode` | Each inline component becomes | +| --- | --- | +| `auto` (default) | its canonical URL; else a published digest whose own spec matches; else a local file | +| `digest` | a verified digest with a local fallback, via `.components.yaml` | +| `name` | an owner-pinned published name with a local fallback, via `.components.yaml` | +| `url` | its canonical URL (a component without one falls back to `digest`) | +| `file` | always a local file | + +Every mode fully dehydrates nested graphs. `auto`, `url` and `file` extract each inline subgraph to a local file under `subgraphs/` beside the output; a subgraph has no canonical URL of its own, so under `url` it too becomes a portable local file. Subgraph filenames are `-.yaml`, addressing the subgraph exactly as written. As a result, identical subgraphs share one file, a subgraph that differs only in an annotation gets its own file, and several outputs dehydrated into one directory cannot overwrite each other's subgraphs. `digest` and `name` extract subgraphs into the components directory, behind the same resolve config. The leaves inside a subgraph follow the chosen mode as usual. + +`--output` is required. The input, `--output` and `--components-dir` must all be local paths: URIs such as `gs://`, `https://` or `file://` are refused before anything is written. A file whose YAML is not a mapping, including an empty or `null` document, is refused too, rather than being written out as `{}`. Extracted files go to `components/` beside the output unless `--components-dir` is given. Verification uses `--base-url`/`--token`/`--auth-header`/`--header`/`--config`, and the client is created only when a lookup is needed. An unreachable or unauthorized library degrades to local copies, but a malformed `--header` or `--auth-header` exits with an error instead of silently shipping local-only output. Verification uses the client's normal timeouts and retries, so an unreachable library can delay each fallback. See [Resolve-config fallback entries](#resolve-config-fallback-entries) for what `digest` and `name` emit and guarantee. + Pipeline run API/submit commands live under `sdk pipeline-runs`: ```bash @@ -1137,6 +1153,30 @@ Resolvers receive the hydrator instance, the reference value, a display path, an Downstream-only features such as Docker/from-container materialization or cloud storage can be added by registering new resolvers while the OSS default remains explicit about unsupported kinds. +#### Resolve-config fallback entries + +A resolve-config fragment may be an ordered list; the first entry that resolves wins. An entry that *raises* normally fails the hydration. An entry marked `fallback_on_error: true` is instead treated as a miss when a later entry exists, so the next one is tried. That particular warning names only the entry index and exception type; other hydration warnings are unchanged and may include exception text. Unmarked entries, single-entry fragments and the last entry always propagate their errors, so hand-authored configs behave exactly as before. + +```yaml +l-501c7f02ā€¦: + - digest: 501c7f02ā€¦ + fallback_on_error: true + - local: ./components/l-501c7f02ā€¦.yaml +``` + +The dehydrator's explicit `digest` and `name` choices emit exactly this shape into `.components.yaml` and reference it as `resolve://./.components.yaml#`. A fragment identifies the whole entry list, so the same component reached through different verified primaries gets separate fragments that share one local copy. + +- **The primary is emitted only when verified.** A digest's *own* published spec must be semantically equal to the inline one; the digests themselves are never compared, because a published digest usually addresses the component's source text. For `name`, the component's digest is inspected (`ComponentInspector.inspect_by_digest`). Its spec must match, and it must have a published name and a non-symbolic `published_by`. The published name is emitted with that `publisher` pinned, so another author's same-named component cannot resolve. Otherwise the fragment is the local copy alone. +- **`name` pins the author, not the version.** Hydration resolves that owner's latest candidate, which may be a newer version or the `[Official]`-prefixed name. +- **Deprecation successors are followed, by design.** Hydration follows a digest's successor even when the successor's content differs. This applies to `digest`, `name` and `auto` alike. The local copy covers *availability*, a library that is unreachable or no longer has the component; it does not pin content or version. +- **A local copy is always written**, content-addressed under the components directory, so the output stays usable when the library is unreachable or the component is later unpublished. +- **Nested graphs are fully dehydrated** into the same manifest. +- **A local output file and components directory are required.** Without them there is nowhere portable to write, so dehydration raises `ResolveManifestUnavailableError` instead of writing into the working directory. + +`auto` also verifies content: a published digest whose own spec is a different component is treated as a library miss, and the component is extracted to a file. + +Each dehydration call starts with fresh extraction state. When one dehydrator instance is reused for several outputs, each output writes its own local copies and never references another output's bundle. + ### Pipeline run hooks `packages/tangle-cli/src/tangle_cli/pipeline_runs.py` defines `PipelineRunHooks`, passed into `PipelineRunManager`. Subclass it to customize submit/load/wait/log behavior: diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py index d866dea..b8a2fd4 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.27" + __version__ = "0.1.28" __all__ = ["TangleDynamicDiscoveryClient", "__version__"] diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_dehydrator.py b/packages/tangle-cli/src/tangle_cli/pipeline_dehydrator.py index 8e8a3fc..1d74d6b 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_dehydrator.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_dehydrator.py @@ -11,11 +11,12 @@ from __future__ import annotations import copy +import hashlib import json import os import re import textwrap -from collections.abc import Mapping +from collections.abc import Callable, Mapping from dataclasses import dataclass from pathlib import Path from typing import Any @@ -50,6 +51,15 @@ def _truncate_utf8(text: str, budget: int) -> str: return encoded[:budget].decode("utf-8", "ignore") +class ResolveManifestUnavailableError(RuntimeError): + """Raised when a portable resolve ref is required but has no path context. + + ``DIGEST``/``NAME`` always materialize a local fallback beside the output, + so an in-memory dehydration with no output path cannot produce a portable + reference. Failing is better than writing a surprising file into the cwd. + """ + + class ComponentFilenameError(RuntimeError): """Base class for problems naming an extracted component file.""" @@ -127,11 +137,13 @@ def __init__( component_extension: str | None = None, *, base_url: str | None = None, + client_factory: Callable[[], Any] | None = None, uri_readers: Mapping[str, UriReader] | None = None, uri_writers: Mapping[str, UriWriter] | None = None, ) -> None: super().__init__( client=client, + client_factory=client_factory, logger=logger, base_url=base_url or DEFAULT_API_URL, ) @@ -150,6 +162,10 @@ def __init__( self.interactive = interactive self._saved_components: dict[str, Path | str] = {} self._component_filenames: dict[str, str] = {} + # fragment id -> ordered resolve-config entries, written as one sidecar + self._resolve_manifest: dict[str, list[dict[str, Any]]] = {} + # content address -> written subgraph file; kept apart from components + self._saved_subgraphs: dict[str, Path | str] = {} self._current_reference_file: Path | str | None = output_file self._io = PipelineHydrator( enable_resolution=False, @@ -247,33 +263,205 @@ def _auto_dehydrate_choice( canonical_url: str | None, resolved_digest: str, name: str, - _spec: dict[str, Any], + spec: dict[str, Any], path: str, - ) -> str: - """Determine Auto mode outcome: ``url``, ``digest``, or ``file``.""" + ) -> tuple[str, str | None]: + """Auto outcome ``url``/``digest``/``file``, plus the digest to emit. + + A digest is emitted only when its own published spec is this component. + """ self.log.info(f" Auto: '{name}' at {path} (digest: {resolved_digest[:16]}...)") if canonical_url: self.log.info(" Auto: has canonical URL -> url ref") - return "url" - if not resolved_digest or resolved_digest == "unknown": - self.log.info(" Auto: no digest -> file") - return "file" + return "url", None + current = self._verified_digest(resolved_digest, spec, path) + if current is None: + return "file", None + self.log.info(f" Auto: digest {current[:16]} found in library -> digest ref") + return "digest", current + + def _content_digest(self, spec: Any) -> str: + """Normalized-content digest, computed the way the hydrator sees specs. + + Never a locator: a published digest often addresses the component's + source TEXT while an inline spec hashes its sorted content, so equal + components legitimately carry different digests. + """ + return utils.compute_spec_digest(self._io.normalize_component_spec(spec)) + + def _verified_digest(self, digest: str, spec: Mapping[str, Any], path: str) -> str | None: + """``digest`` itself when its OWN published spec is this component. + + The digest is verified, not its current successor: following a later + deprecation is accepted product behavior, even when the successor's + content differs, so hydration is left to do that as it always has. + """ + if not digest or digest == "unknown": + return None + # Client creation can exit on missing credentials; both it and the + # lookup degrade to local-only rather than failing the dehydration. try: client = self._get_client() except (Exception, SystemExit): - self.log.info(" Auto: no API client available -> file") - return "file" + client = None if client is None: - self.log.info(" Auto: no API client provided -> file") - return "file" + self.log.info(f" No component library client for {path} -> local only") + return None try: - client.get_component_spec(resolved_digest) - self.log.info(f" Auto: digest {resolved_digest[:16]} found in library -> digest ref") - return "digest" + published = client.get_component_spec(digest) except Exception: - self.log.info(f" Auto: digest {resolved_digest[:16]} not in library -> file") - return "file" + self.log.info(f" Digest {digest[:16]} not in library -> local only") + return None + if self._content_digest(published) != self._content_digest(spec): + self.log.info(f" Digest {digest[:16]} resolves elsewhere -> local only") + return None + return digest + + def _verified_digest_primary( + self, digest: str, spec: Mapping[str, Any], path: str + ) -> dict[str, Any] | None: + verified = self._verified_digest(digest, spec, path) + return {"digest": verified} if verified else None + + def _verified_name_primary( + self, digest: str, spec: Mapping[str, Any], path: str + ) -> dict[str, Any] | None: + """``{name, publisher}`` from the inspected publication of ``digest``. + + This pins the AUTHOR, not the version: hydration resolves that owner's + latest candidate. The inspected spec must equal the inline one, and the + emitted name is the published one rather than any inline placeholder. + """ + from .authenticated_identity import is_symbolic_me + from .component_inspector import ComponentInspector + + if not digest or digest == "unknown": + return None + try: + client = self._get_client() + except (Exception, SystemExit): + client = None + if client is None: + self.log.info(f" No component library client for {path} -> local only") + return None + try: + inspected = ComponentInspector(client=client, logger=self.log).inspect_by_digest( + digest, full_spec=True + ) + except Exception: + self.log.info(f" Inspection failed for {path} -> local only") + return None + published = inspected.get("spec") + if ( + inspected.get("status") != "success" + or published is None + or self._content_digest(published) != self._content_digest(spec) + ): + self.log.info(f" No matching publication for {path} -> local only") + return None + name, owner = inspected.get("name"), inspected.get("published_by") + if not isinstance(name, str) or not name.strip(): + self.log.info(f" Publication for {path} has no name -> local only") + return None + if not isinstance(owner, str) or not owner.strip() or is_symbolic_me(owner): + self.log.info(f" Untrustworthy owner for {path} -> local only") + return None + return {"name": name, "publisher": owner} + + def _manifest_path(self) -> Path: + """``.components.yaml`` beside a LOCAL output file.""" + output = self.output_file + if output is None or not self._is_local_destination(output): + raise ResolveManifestUnavailableError( + "digest/name dehydration writes a resolve config and a local " + "fallback beside the output, so it needs a local output_file; " + "none was given." + ) + if not self._is_local_destination(self.components_dir): + raise ResolveManifestUnavailableError( + "digest/name dehydration needs a local components_dir for its " + "fallback copies." + ) + text = str(output) + path = Path(text[7:] if text.startswith("file://") else text) + return (path.parent / f"{path.stem}.components.yaml").resolve() + + def _portable_resolve_ref( + self, name: str, spec: dict[str, Any], primary: dict[str, Any] | None + ) -> dict[str, str]: + """Materialize ``spec`` locally and reference it through the manifest. + + The fragment is an ordered list, so the local copy is used only when + the primary fails; a single mapping would instead invoke the + resolver's version comparison and could prefer the local copy. + """ + manifest = self._manifest_path() + self._save_component_to_file(name, spec) + saved = Path(str(self._saved_components[utils.compute_spec_digest(spec)])) + # The fragment addresses the whole entry list: the content (via the + # content-addressed file) AND the primary. Equal specs reached through + # different verified locators are distinct fragments, not a conflict. + fragment = saved.stem + if primary: + policy = json.dumps(primary, sort_keys=True).encode("utf-8") + fragment = f"{fragment}-{hashlib.sha256(policy).hexdigest()[:12]}" + local = os.path.relpath(saved, manifest.parent).replace(os.sep, "/") + # The marker lets the hydrator fall through to the local copy when the + # primary raises (404, offline); the local entry stays unmarked. + marked = [{**primary, "fallback_on_error": True}] if primary else [] + self._resolve_manifest[fragment] = [*marked, {"local": f"./{local}"}] + ref_text = str(self._current_reference_file or self.output_file) + ref_dir = Path(ref_text[7:] if ref_text.startswith("file://") else ref_text).parent.resolve() + rel = os.path.relpath(manifest, ref_dir).replace(os.sep, "/") + return {"url": f"resolve://./{rel}#{fragment}"} + + def _write_resolve_manifest(self) -> None: + if self._resolve_manifest: + manifest = self._manifest_path() + ordered = {k: self._resolve_manifest[k] for k in sorted(self._resolve_manifest)} + self._write_text(manifest, utils.dump_yaml(ordered), kind="resolve config") + + def _portable_primary( + self, choice: str, name: str, digest: str, spec: Mapping[str, Any], path: str + ) -> dict[str, Any] | None: + if choice == DehydrateChoice.NAME: + return self._verified_name_primary(digest, spec, path) + return self._verified_digest_primary(digest, spec, path) + + def _extract_subgraphs_portable(self, data: dict[str, Any], choice: str) -> None: + """Replace every inline graph boundary with a manifest-backed ref. + + Deepest first, so a boundary is written only after its own nested + boundaries are already refs. Each extracted graph lives in the + components bundle, and its filename addresses the dehydrated content + that is actually written. + """ + bundle_file = self._join_destination(self.components_dir, "_") + root_reference = self._current_reference_file + for depth, path in _build_subgraph_processing_queue(data): + if depth == 0: + continue + result = _get_subgraph_by_path(data, path) + if not result: + continue + component_ref, spec = result + spec_name = str(spec.get("name", "subgraph")) + try: + self._current_reference_file = bundle_file + written = utils.traverse_pipeline_tasks( + copy.deepcopy(spec), spec_name, self._process_task + ) + digest = component_ref.get("digest") or utils.compute_spec_digest(spec) + primary = self._portable_primary(choice, spec_name, digest, spec, path) + # The boundary ref lives in its parent: the root for depth 1, + # otherwise another extracted graph in the bundle. + self._current_reference_file = root_reference if depth == 1 else bundle_file + new_ref = self._portable_resolve_ref(spec_name, written, primary) + finally: + self._current_reference_file = root_reference + component_ref.clear() + component_ref.update(new_ref) def _prompt_choice(self, name: str, digest: str, canonical_url: str | None, path: str) -> str: self.log.info(f"\nšŸ“¦ Found componentRef at: {path}") @@ -352,23 +540,33 @@ def _process_task( new_task = {k: v for k, v in task_data.items() if k != "componentRef"} if choice == DehydrateChoice.AUTO: - effective = self._auto_dehydrate_choice(canonical_url, resolved_digest, name, spec, path) + effective, current = self._auto_dehydrate_choice( + canonical_url, resolved_digest, name, spec, path + ) if effective == "url": new_task["componentRef"] = {"url": canonical_url} self.log.info(" ā†’ Auto: Replaced with componentRef.url") elif effective == "digest": - new_task["componentRef"] = {"digest": resolved_digest} + new_task["componentRef"] = {"digest": current} self.log.info(" ā†’ Auto: Replaced with componentRef.digest (found in library)") else: file_url = self._save_component_to_file(name, spec) new_task["componentRef"] = {"url": file_url} self.log.info(" ā†’ Auto: Extracted to file (no URL, not in library or no client)") elif choice == DehydrateChoice.DIGEST: - new_task["componentRef"] = {"digest": resolved_digest} - self.log.info(" ā†’ Replaced with componentRef.digest") + primary = self._verified_digest_primary(resolved_digest, spec, path) + new_task["componentRef"] = self._portable_resolve_ref(name, spec, primary) + self.log.info( + " ā†’ Replaced with componentRef.url (resolve config: " + f"{'digest + local' if primary else 'local only'})" + ) elif choice == DehydrateChoice.NAME: - new_task["componentRef"] = {"name": name} - self.log.info(" ā†’ Replaced with componentRef.name") + primary = self._verified_name_primary(resolved_digest, spec, path) + new_task["componentRef"] = self._portable_resolve_ref(name, spec, primary) + self.log.info( + " ā†’ Replaced with componentRef.url (resolve config: " + f"{'name + local' if primary else 'local only'})" + ) elif choice == DehydrateChoice.URL and canonical_url: new_task["componentRef"] = {"url": canonical_url} self.log.info(" ā†’ Replaced with componentRef.url") @@ -483,12 +681,19 @@ def _subgraph_destination(self, filename: str) -> Path | str: return self._join_destination(subgraph_dir, filename) def _extract_subgraphs_to_files(self, data: dict[str, Any]) -> dict[str, Any]: - """Extract subgraph specs to YAML files and replace them with URL refs.""" + """Extract inline subgraph specs to content-addressed YAML files. - queue = _build_subgraph_processing_queue(data) - subgraph_counter = 0 + A filename addresses the fully written subgraph, so identical subgraphs + share one file and two outputs in one directory can only ever collide on + identical bytes. A per-run counter let a later output overwrite an + earlier output's subgraph with different content. Deepest first, so a + parent's address covers its children's content-addressed names. + """ - for depth, path in queue: + # Relative URLs inside a subgraph depend only on its directory, so they + # are computed against a placeholder there before the address is known. + placeholder = self._subgraph_destination("_") + for depth, path in _build_subgraph_processing_queue(data): if depth == 0: continue @@ -496,34 +701,35 @@ def _extract_subgraphs_to_files(self, data: dict[str, Any]) -> dict[str, Any]: if not result: continue component_ref, spec = result - - spec_name = spec.get("name", "subgraph") - filename = f"{self._safe_filename(str(spec_name), 'subgraph')}_{subgraph_counter}{self.component_extension}" - subgraph_counter += 1 - destination = self._subgraph_destination(filename) + spec_name = str(spec.get("name", "subgraph")) original_ref = self._current_reference_file - self._current_reference_file = destination + self._current_reference_file = placeholder try: - spec_to_write = utils.traverse_pipeline_tasks(copy.deepcopy(spec), str(spec_name), self._process_task) + spec_to_write = utils.traverse_pipeline_tasks(copy.deepcopy(spec), spec_name, self._process_task) finally: self._current_reference_file = original_ref - - if self._is_local_destination(destination): - destination_text = str(destination) - if destination_text.startswith("file://"): - destination_text = destination_text[7:] - destination_path = Path(destination_text) - self._relativize_file_urls(spec_to_write, destination_path.parent) - self._write_text(destination_path, utils.dump_yaml(spec_to_write) + "\n", kind="subgraph") - component_url = f"file://{destination_path.resolve()}" - else: + local = self._is_local_destination(placeholder) + if local: + placeholder_text = str(placeholder) + placeholder_path = Path(placeholder_text[7:] if placeholder_text.startswith("file://") else placeholder_text) + self._relativize_file_urls(spec_to_write, placeholder_path.parent) + + address = utils.compute_spec_digest(spec_to_write) + destination = self._saved_subgraphs.get(address) + if destination is None: + filename = self._component_filename(spec_name, address) + destination = self._subgraph_destination(filename) + if local: + text = str(destination) + destination = Path(text[7:] if text.startswith("file://") else text).resolve() + # Always (re)written once per run: a file already on disk under + # this name is not trusted to still hold this content. self._write_text(destination, utils.dump_yaml(spec_to_write) + "\n", kind="subgraph") - component_url = str(destination) - - self.log.info(f" šŸ“¦ Extracted subgraph '{spec_name}' -> {filename}") + self._saved_subgraphs[address] = destination + self.log.info(f" šŸ“¦ Extracted subgraph '{spec_name}' -> {filename}") component_ref.clear() - component_ref["url"] = component_url + component_ref["url"] = f"file://{destination}" if local else str(destination) if self.output_file and self._is_local_destination(self.output_file): output_file_text = str(self.output_file) @@ -538,11 +744,27 @@ def dehydrate(self, data: dict[str, Any]) -> dict[str, Any]: """Return a dehydrated copy of *data* according to configured choices.""" working = copy.deepcopy(data) - if self.remembered_choices.get("") == DehydrateChoice.AUTO: + # One output per call. ``dehydrate_file``/``export_to_jinja2`` retarget + # output_file and components_dir between calls, so extraction caches + # from an earlier output would point this one at the old bundle. + self._saved_components = {} + self._component_filenames = {} + self._resolve_manifest = {} + self._saved_subgraphs = {} + self._current_reference_file = self.output_file + default_choice = self.remembered_choices.get("") + # Every choice that promises a dehydrated document must replace nested + # graph boundaries too; otherwise they keep their inline spec. A nested + # graph has no canonical URL of its own, so URL extracts it to a file. + if default_choice in (DehydrateChoice.AUTO, DehydrateChoice.FILE, DehydrateChoice.URL): self._extract_subgraphs_to_files(working) + elif default_choice in (DehydrateChoice.DIGEST, DehydrateChoice.NAME): + self._extract_subgraphs_portable(working, default_choice) pipeline_name = working.get("name", "pipeline") - return utils.traverse_pipeline_tasks(working, str(pipeline_name), self._process_task) + result = utils.traverse_pipeline_tasks(working, str(pipeline_name), self._process_task) + self._write_resolve_manifest() + return result def export_to_jinja2( self, diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_hydrator.py b/packages/tangle-cli/src/tangle_cli/pipeline_hydrator.py index 4646ed5..c2a1d5d 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_hydrator.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_hydrator.py @@ -943,12 +943,31 @@ def _resolve_from_config( path: str, base_dir: Path | None = None, ) -> tuple[str, dict[str, Any]] | None: - """Resolve a component from a parsed resolve config.""" + """Resolve a component from a parsed resolve config. + + An entry marked ``fallback_on_error: true`` that raises is treated as a + miss when a later entry exists, so a generated ``[primary, local]`` + list stays usable offline. Unmarked, single and last entries still + propagate their errors. + """ entries = config if isinstance(config, list) else [config] for i, entry in enumerate(entries): if not isinstance(entry, dict): self.log.warn(f" āš ļø Resolve config entry {i} is not a dict, skipping") continue + if entry.get("fallback_on_error") is True and i < len(entries) - 1: + try: + result = self._try_resolve_entry(entry, path, base_dir) + except Exception as exc: + # Type only: the message may carry a URL or response body. + self.log.warn( + f" āš ļø Resolve config entry {i} failed at {path} " + f"({type(exc).__name__}); trying the next entry" + ) + continue + if result is not None: + return result + continue result = self._try_resolve_entry(entry, path, base_dir) if result is not None: return result diff --git a/packages/tangle-cli/src/tangle_cli/pipelines.py b/packages/tangle-cli/src/tangle_cli/pipelines.py index 0c6eb05..b5d8ba2 100644 --- a/packages/tangle-cli/src/tangle_cli/pipelines.py +++ b/packages/tangle-cli/src/tangle_cli/pipelines.py @@ -59,6 +59,17 @@ class LayoutResult: graphs_positioned: int +@dataclass(frozen=True) +class DehydrateResult: + """Summary of a dehydrate operation.""" + + output_path: Path + mode: str + components_dir: Path | None + extracted_components: int + resolve_config: Path | None + + @dataclass(frozen=True) class HydrateResult: """Summary of a hydrate operation.""" @@ -245,6 +256,96 @@ def hydrate_pipeline_file( ) +DEHYDRATE_MODES = ("auto", "digest", "name", "url", "file") + + +def _local_path(value: str | Path, what: str) -> Path: + """This command is local-path-only; refuse URIs before ``Path`` mangles them. + + ``Path("gs://b/x")`` collapses to ``gs:/b/x``, a relative LOCAL path, so the + check must run on the raw string. The rule matches the engine's own URI + detection (``://``), so a Windows ``C:\\dir`` path is still local. The value + itself is not echoed: a URI may carry credentials. + """ + if isinstance(value, str) and "://" in value: + scheme = value.split("://", 1)[0] + raise PipelineValidationError(f"{what} must be a local path; {scheme}:// URIs are not supported") + return Path(value) + + +def dehydrate_pipeline_file( + pipeline_path: str | Path, + *, + output: str | Path, + mode: str = "auto", + components_dir: str | Path | None = None, + client: Any | None = None, + logger: Any | None = None, + base_url: str | None = None, +) -> DehydrateResult: + """Dehydrate a local pipeline YAML with the shared ``PipelineDehydrator``. + + Noninteractive: ``mode`` applies to every component. Input, output and + components directory must all be local paths. Without ``client``, one is + created lazily -- against ``base_url`` when given -- only if a lookup needs it. + """ + + from .pipeline_dehydrator import ( + ComponentFilenameError, + DehydrateChoice, + PipelineDehydrator, + ResolveManifestUnavailableError, + ) + + choices = { + "auto": DehydrateChoice.AUTO, + "digest": DehydrateChoice.DIGEST, + "name": DehydrateChoice.NAME, + "url": DehydrateChoice.URL, + "file": DehydrateChoice.FILE, + } + if mode not in choices: + raise PipelineValidationError( + f"Unknown dehydrate mode {mode!r}; expected one of {', '.join(DEHYDRATE_MODES)}" + ) + source = _local_path(pipeline_path, "pipeline path") + output_path = _local_path(output, "--output") + bundle = _local_path(components_dir, "--components-dir") if components_dir is not None else None + # Strict on purpose: the dehydrator's own loader turns null/false/0/[]/"" into + # an empty mapping and would write ``{}``. + data = load_pipeline_file(source) + + def create_client() -> Any: + from .client import TangleApiClient + + return TangleApiClient(base_url=base_url, logger=logger) + + dehydrator = PipelineDehydrator( + {"": choices[mode]}, + components_dir=bundle, + output_file=output_path, + client=client, + client_factory=create_client if client is None and base_url is not None else None, + logger=logger, + base_url=base_url, + ) + try: + dehydrator.write_file(dehydrator.dehydrate(data), output_path) + except (ComponentFilenameError, ResolveManifestUnavailableError) as exc: + raise PipelineValidationError(str(exc)) from exc + + # Report only what THIS run wrote; a sidecar or bundle from an earlier run + # with a different mode may still be on disk. + extracted = len(dehydrator._saved_components) + return DehydrateResult( + output_path=output_path, + mode=mode, + components_dir=Path(str(dehydrator.components_dir)) if extracted else None, + extracted_components=extracted, + resolve_config=dehydrator._manifest_path() if dehydrator._resolve_manifest else None, + ) + + # --------------------------------------------------------------------------- # Compile # --------------------------------------------------------------------------- diff --git a/packages/tangle-cli/src/tangle_cli/pipelines_cli.py b/packages/tangle-cli/src/tangle_cli/pipelines_cli.py index 533fb5f..a87cb18 100644 --- a/packages/tangle-cli/src/tangle_cli/pipelines_cli.py +++ b/packages/tangle-cli/src/tangle_cli/pipelines_cli.py @@ -3,7 +3,7 @@ from __future__ import annotations import pathlib -from typing import Annotated, Any +from typing import Annotated, Any, Literal from cyclopts import App, Parameter @@ -26,6 +26,7 @@ from .pipelines import ( PipelineValidationError, compile_pipeline_file, + dehydrate_pipeline_file, generate_mermaid, hydrate_pipeline_file, layout_pipeline_file, @@ -231,6 +232,95 @@ def pipelines_hydrate( ) +@app.command(name="dehydrate") +def pipelines_dehydrate( + # ``str``, not ``Path``: Path("gs://b/x") collapses to the LOCAL path + # gs:/b/x, which would hide a URI from the local-path-only check. + pipeline_path: str, + *, + output: Annotated[ + str | None, + Parameter( + name="--output", + alias="-o", + help="Local output path for the dehydrated pipeline YAML (required).", + ), + ] = None, + mode: Annotated[ + Literal["auto", "digest", "name", "url", "file"] | None, + Parameter( + name="--mode", + help=( + "How to replace inline component specs. auto: canonical URL, else a " + "verified published digest, else a local file. digest/name: a " + "verified digest or owner-pinned name with a local fallback, via a " + ".components.yaml resolve config. url: canonical URL. " + "file: always a local file. Default: auto." + ), + ), + ] = None, + components_dir: Annotated[ + str | None, + Parameter( + name="--components-dir", + help="Where extracted component files go. Defaults to components/ beside the output.", + ), + ] = None, + base_url: BaseUrlOption = None, + token: TokenOption = None, + auth_header: AuthHeaderOption = None, + header: HeaderOption = None, + config: ConfigOption = None, + log_type: LogTypeOption = "console", +) -> None: + """Dehydrate inline component specs in a local pipeline YAML file.""" + + config_values = load_config_or_exit(config) + resolved_output = output or _optional_str(config_values.get("output")) + if resolved_output is None: + raise SystemExit("--output is required: dehydration writes files beside the output.") + config_base_url = _optional_str(config_values.get("base_url")) + resolved_base_url = base_url if base_url is not None else config_base_url + include_env_credentials = not (base_url is None and config_base_url is not None) + resolved_token = token if token is not None else _optional_str(config_values.get("token")) + resolved_auth_header = ( + auth_header if auth_header is not None else _optional_str(config_values.get("auth_header")) + ) + resolved_mode = mode if mode is not None else (_optional_str(config_values.get("mode")) or "auto") + + logger, finalize_logs = logger_for_log_type(log_type) + try: + result = dehydrate_pipeline_file( + pipeline_path, + output=resolved_output, + mode=resolved_mode, + components_dir=components_dir or _optional_str(config_values.get("components_dir")), + base_url=resolved_base_url, + logger=logger, + # Lazy: a network or auth failure during verification degrades to a + # local copy, while a malformed --header/--auth-header still exits. + client=LazyTangleApiClient( + command_name="pipeline dehydration with published-component verification", + base_url=resolved_base_url, + token=resolved_token, + auth_header=resolved_auth_header, + header=_header_entries(header, config_values), + include_env_credentials=include_env_credentials, + logger=logger, + ), + ) + except PipelineValidationError as exc: + raise SystemExit(str(exc)) from exc + finally: + finalize_logs() + + print(f"Dehydrated {pipeline_path} -> {result.output_path} (mode: {result.mode}).") + if result.resolve_config is not None: + print(f" Resolve config: {result.resolve_config}") + if result.components_dir is not None: + print(f" Components: {result.extracted_components} file(s) in {result.components_dir}") + + @app.command(name="compile") def pipelines_compile( pipeline_path: pathlib.Path, diff --git a/pyproject.toml b/pyproject.toml index 1a8c40e..d457bfb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tangle-cli" -version = "0.1.27" +version = "0.1.28" description = "CLI for Tangle, the open-source ML pipeline orchestration platform" readme = "README.md" authors = [ diff --git a/tests/test_dehydrator_portable_refs.py b/tests/test_dehydrator_portable_refs.py new file mode 100644 index 0000000..71e63ba --- /dev/null +++ b/tests/test_dehydrator_portable_refs.py @@ -0,0 +1,418 @@ +"""Digest/name dehydration emits portable, semantically verified references. + +A published digest is trusted only when it resolves to the component actually +in hand, and explicit DIGEST/NAME always ship a local copy behind an ordered +resolve config, so the output stays usable when the library is unreachable. +""" + +from __future__ import annotations + +import copy +from pathlib import Path +from typing import Any + +import pytest +import yaml + +from tangle_cli.models import ComponentInfo +from tangle_cli.pipeline_dehydrator import ( + DehydrateChoice, + PipelineDehydrator, + ResolveManifestUnavailableError, +) +from tangle_cli.pipeline_hydrator import PipelineHydrator +from tangle_cli.utils import compute_spec_digest, compute_text_digest, dump_yaml + +LEAF = {"name": "L", "implementation": {"container": {"image": "i", "command": ["c"]}}} +OTHER = {"name": "L", "implementation": {"container": {"image": "other", "command": ["x"]}}} +NAMELESS = {"implementation": LEAF["implementation"]} +INNER = {"name": "Inner", "implementation": {"graph": {"tasks": {"leaf": {"componentRef": {"spec": LEAF}}}}}} +SPEC_DIGEST = compute_spec_digest(LEAF) +TEXT_DIGEST = compute_text_digest(dump_yaml(LEAF)) + + +class Library: + """Published digests resolve to ``specs``; ``error`` simulates an outage.""" + + def __init__(self, specs=None, infos=(), error: Exception | None = None, successors=None) -> None: + self.specs = specs or {} + self.infos = list(infos) + self.error = error + self.successors = successors or {} + + def resolve_digest(self, digest: str) -> str: + return self.successors.get(digest, digest) + + def get_component_spec(self, digest: str) -> dict[str, Any]: + if self.error is not None: + raise self.error + if digest not in self.specs: + raise KeyError(digest) + return copy.deepcopy(self.specs[digest]) + + def find_existing_components(self, *args: Any, **kwargs: Any) -> list[ComponentInfo]: + if self.error is not None: + raise self.error + return list(self.infos) + + +def _pipeline(ref: dict[str, Any], task: str = "t") -> dict[str, Any]: + return {"name": "O", "implementation": {"graph": {"tasks": {task: {"componentRef": ref}}}}} + + +def _dehydrate(tmp_path: Path, choice: str, document: dict[str, Any], client: Any) -> dict[str, Any]: + return PipelineDehydrator({"": choice}, output_file=tmp_path / "out.yaml", client=client).dehydrate( + document + ) + + +def _ref(result: dict[str, Any], task: str = "t") -> dict[str, Any]: + return result["implementation"]["graph"]["tasks"][task]["componentRef"] + + +def _fragment(tmp_path: Path, result: dict[str, Any], task: str = "t") -> list[dict[str, Any]]: + url = _ref(result, task)["url"] + assert url.startswith("resolve://./out.components.yaml#") + manifest = yaml.safe_load((tmp_path / "out.components.yaml").read_text(encoding="utf-8")) + return manifest[url.split("#", 1)[1]] + + +def _leaf_digests(document: dict[str, Any]) -> list[str]: + found: list[str] = [] + for task in document["implementation"]["graph"]["tasks"].values(): + spec = task["componentRef"]["spec"] + if "graph" in spec["implementation"]: + found += _leaf_digests(spec) + else: + found.append(compute_spec_digest({k: v for k, v in spec.items() if not k.startswith("_")})) + return found + + +# --------------------------------------------------------------------------- AUTO + + +@pytest.mark.parametrize( + ("label", "ref", "published", "expected"), + [ + ("carried-digest-matches", {"digest": TEXT_DIGEST, "spec": LEAF}, {TEXT_DIGEST: LEAF}, "digest"), + ("carried-digest-is-stale", {"digest": TEXT_DIGEST, "spec": LEAF}, {TEXT_DIGEST: OTHER}, "file"), + ("spec-digest-indexed", {"spec": LEAF}, {SPEC_DIGEST: LEAF}, "digest"), + ("text-digest-indexed-only", {"spec": LEAF}, {TEXT_DIGEST: LEAF}, "file"), + ], +) +def test_auto_trusts_a_digest_only_when_it_resolves_to_this_component( + tmp_path: Path, label: str, ref: dict[str, Any], published: dict[str, Any], expected: str +) -> None: + """Existence in the library is not enough: a stale or foreign digest that + happens to be published must not replace the spec it no longer describes.""" + result = _dehydrate(tmp_path, DehydrateChoice.AUTO, _pipeline(ref), Library(published)) + + emitted = _ref(result) + assert ("digest" in emitted) == (expected == "digest"), label + assert ("url" in emitted and emitted["url"].startswith("file://")) == (expected == "file"), label + + +# ------------------------------------------------------------------------- DIGEST + + +def test_a_verified_digest_is_primary_with_a_marked_local_fallback(tmp_path: Path) -> None: + ref = {"digest": TEXT_DIGEST, "spec": LEAF} + + result = _dehydrate(tmp_path, DehydrateChoice.DIGEST, _pipeline(ref), Library({TEXT_DIGEST: LEAF})) + + fragment = _fragment(tmp_path, result) + local = fragment[-1]["local"] + assert fragment == [{"digest": TEXT_DIGEST, "fallback_on_error": True}, {"local": local}] + assert yaml.safe_load((tmp_path / local).read_text(encoding="utf-8")) == LEAF + + +@pytest.mark.parametrize( + ("label", "library"), + [ + ("stale", Library({TEXT_DIGEST: OTHER})), + ("unpublished", Library({})), + ("offline", Library(error=ConnectionError("unreachable"))), + ], +) +def test_an_unverifiable_digest_ships_the_local_copy_alone( + tmp_path: Path, label: str, library: Library +) -> None: + """DIGEST never emits a digest it could not verify, and never fails for it.""" + ref = {"digest": TEXT_DIGEST, "spec": LEAF} + + result = _dehydrate(tmp_path, DehydrateChoice.DIGEST, _pipeline(ref), library) + + fragment = _fragment(tmp_path, result) + assert len(fragment) == 1 and set(fragment[0]) == {"local"}, label + + +def test_digest_dehydration_without_an_output_path_writes_nothing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """With nowhere portable to go it raises, instead of writing into the cwd.""" + monkeypatch.chdir(tmp_path) + + with pytest.raises(ResolveManifestUnavailableError): + PipelineDehydrator({"": DehydrateChoice.DIGEST}, client=Library()).dehydrate( + _pipeline({"spec": LEAF}) + ) + + assert list(tmp_path.iterdir()) == [] + + +@pytest.mark.parametrize("failure", ["raises", "none"]) +@pytest.mark.parametrize("choice", [DehydrateChoice.DIGEST, DehydrateChoice.NAME]) +def test_no_usable_client_degrades_to_the_local_copy(tmp_path: Path, choice: str, failure: str) -> None: + """Client creation may exit on missing credentials or yield nothing; the + dehydration still succeeds, shipping only the local copy.""" + + class NoClient(PipelineDehydrator): + def _get_client(self): + if failure == "raises": + raise SystemExit("no credentials") + return None + + result = NoClient({"": choice}, output_file=tmp_path / "out.yaml").dehydrate( + _pipeline({"digest": SPEC_DIGEST, "spec": LEAF}) + ) + + fragment = _fragment(tmp_path, result) + assert len(fragment) == 1 and set(fragment[0]) == {"local"} + + +@pytest.mark.parametrize("primary_fails", [False, True], ids=["primary-ok", "primary-fails"]) +def test_a_nested_graph_is_fully_dehydrated_and_rehydrates_to_the_original( + tmp_path: Path, primary_fails: bool +) -> None: + document = _pipeline({"spec": INNER}, task="sub") + out = tmp_path / "out.yaml" + result = _dehydrate(tmp_path, DehydrateChoice.DIGEST, document, Library({SPEC_DIGEST: LEAF})) + out.write_text(dump_yaml(result), encoding="utf-8") + + # Fully dehydrated: every ref at every level is a pure locator. + written = [yaml.safe_load(out.read_text(encoding="utf-8"))] + written += [yaml.safe_load(p.read_text(encoding="utf-8")) for p in (tmp_path / "components").glob("*.yaml")] + for doc in written: + for task in doc.get("implementation", {}).get("graph", {}).get("tasks", {}).values(): + assert set(task["componentRef"]) == {"url"} + library = Library({SPEC_DIGEST: LEAF}, error=ConnectionError("down") if primary_fails else None) + hydrated = PipelineHydrator(client=library).hydrate_file(out).data + + assert _leaf_digests(hydrated) == _leaf_digests(document) + + +# --------------------------------------------------------------------------- NAME + + +def _real_client(rows: list[dict[str, Any]], specs: dict[str, dict[str, Any]]) -> Any: + """The REAL client, stubbed only at its HTTP seams. + + Owner discovery goes through ``ComponentInspector.inspect_by_digest`` on + this client, so the production request/row shape is what is exercised. + """ + from tangle_cli.client import TangleApiClient + from tangle_cli.models import ComponentSpec + + client = TangleApiClient(base_url="https://review.invalid", include_env_credentials=False) + client._published_component_rows = lambda **kw: [ + copy.deepcopy(r) for r in rows if kw.get("digest") in (None, r["digest"]) + ] + client.get_component_spec = lambda digest: ComponentSpec.from_dict(copy.deepcopy(specs[digest])) + return client + + +def _row(owner: Any, digest: str = SPEC_DIGEST, name: str = "L") -> dict[str, Any]: + return {"name": name, "digest": digest, "published_by": owner} + + +def test_a_name_reference_is_pinned_to_its_inspected_owner(tmp_path: Path) -> None: + """NAME pins the AUTHOR, not the version: hydration resolves that owner's + latest candidate, and the local copy is an availability fallback only.""" + client = _real_client([_row("user-123")], {SPEC_DIGEST: LEAF}) + + result = _dehydrate(tmp_path, DehydrateChoice.NAME, _pipeline({"spec": LEAF}), client) + + fragment = _fragment(tmp_path, result) + assert fragment[0] == {"name": "L", "publisher": "user-123", "fallback_on_error": True} + assert set(fragment[1]) == {"local"} + + +@pytest.mark.parametrize( + ("label", "rows", "specs", "spec"), + [ + ("unpublished", [], {SPEC_DIGEST: LEAF}, LEAF), + ("symbolic-me", [_row("me")], {SPEC_DIGEST: LEAF}, LEAF), + ("empty-owner", [_row("")], {SPEC_DIGEST: LEAF}, LEAF), + # The carried digest is published, but as different content. + ("different-content", [_row("user-1")], {SPEC_DIGEST: OTHER}, LEAF), + # Content matches, so only the no-name rule can stop the lookup. + ("nameless", [_row("user-1", compute_spec_digest(NAMELESS))], {compute_spec_digest(NAMELESS): NAMELESS}, NAMELESS), + ], +) +def test_name_is_emitted_only_for_an_inspected_matching_publication( + tmp_path: Path, label: str, rows: list[dict[str, Any]], specs: dict[str, Any], spec: dict[str, Any] +) -> None: + """Without a trustworthy inspected owner a name lookup would be global, + so a same-named component by any other author could resolve instead.""" + client = _real_client(rows, specs) + + result = _dehydrate(tmp_path, DehydrateChoice.NAME, _pipeline({"spec": spec}), client) + + fragment = _fragment(tmp_path, result) + assert len(fragment) == 1 and set(fragment[0]) == {"local"}, label + + +# ------------------------------------------------- fragment identity and reuse + + +@pytest.mark.parametrize( + ("label", "second", "library"), + [ + ("two-verified-locators", {"spec": LEAF}, Library({TEXT_DIGEST: LEAF, SPEC_DIGEST: LEAF})), + ("verified-and-unpublished", {"digest": "f" * 64, "spec": LEAF}, Library({TEXT_DIGEST: LEAF})), + ], +) +def test_equal_specs_with_different_primaries_get_distinct_fragments( + tmp_path: Path, label: str, second: dict[str, Any], library: Library +) -> None: + """The fragment addresses the whole entry list, not just the content.""" + document = { + "name": "O", + "implementation": { + "graph": { + "tasks": { + "a": {"componentRef": {"digest": TEXT_DIGEST, "spec": LEAF}}, + "b": {"componentRef": second}, + } + } + }, + } + + result = _dehydrate(tmp_path, DehydrateChoice.DIGEST, document, library) + + a, b = _fragment(tmp_path, result, "a"), _fragment(tmp_path, result, "b") + assert a != b, label + assert a[-1] == b[-1], "one content-addressed local copy serves both" + + +@pytest.mark.parametrize("entry", ["dehydrate_file", "retargeted_attributes"]) +@pytest.mark.parametrize("choice", [DehydrateChoice.DIGEST, DehydrateChoice.AUTO]) +def test_a_reused_dehydrator_gives_every_output_its_own_bundle( + tmp_path: Path, choice: str, entry: str +) -> None: + """Caches from an earlier output must not point a later one at the old + bundle -- that output would ship no local copy of its own. + + Every output shares one component (a stale cache would be hit) and has one + of its own (a leaked fragment could not hide behind a shared key). + """ + first = tmp_path / "out0" / "out.yaml" + dehydrator = PipelineDehydrator({"": choice}, output_file=first, client=Library({})) + for index in range(2): + own = {"name": f"Own{index}", "implementation": {"container": {"image": f"own{index}"}}} + document = { + "name": "O", + "implementation": { + "graph": {"tasks": {"shared": {"componentRef": {"spec": LEAF}}, "own": {"componentRef": {"spec": own}}}} + }, + } + out_dir = tmp_path / f"out{index}" + out_dir.mkdir(exist_ok=True) + out = out_dir / "out.yaml" + if entry == "dehydrate_file": + (out_dir / "in.yaml").write_text(dump_yaml(document), encoding="utf-8") + dehydrator.dehydrate_file(out_dir / "in.yaml", out) + else: + dehydrator.output_file = out + dehydrator.components_dir = out_dir / "components" + out.write_text(dump_yaml(dehydrator.dehydrate(document)), encoding="utf-8") + + text = out.read_text(encoding="utf-8") + assert "../" not in text, "a ref escaped into another output's bundle" + assert len(list((out_dir / "components").glob("*.yaml"))) == 2, "a local copy is missing" + if choice == DehydrateChoice.DIGEST: + manifest = yaml.safe_load((out_dir / "out.components.yaml").read_text(encoding="utf-8")) + assert len(manifest) == 2, "fragments leaked from the earlier output" + + +# ------------------------------------------------ deprecation successors + + +@pytest.mark.parametrize("choice", [DehydrateChoice.DIGEST, DehydrateChoice.AUTO]) +def test_export_verifies_the_carried_digest_itself_not_its_successor( + tmp_path: Path, choice: str +) -> None: + """Accepted behavior, pinned so it is not "fixed" back: the digest whose + OWN spec matches is emitted, even if it already has a different successor.""" + successor = "5" * 64 + library = Library({SPEC_DIGEST: LEAF, successor: OTHER}, successors={SPEC_DIGEST: successor}) + + result = _dehydrate(tmp_path, choice, _pipeline({"digest": SPEC_DIGEST, "spec": LEAF}), library) + + emitted = _ref(result) if choice == DehydrateChoice.AUTO else _fragment(tmp_path, result)[0] + assert emitted["digest"] == SPEC_DIGEST + + +@pytest.mark.parametrize("choice", [DehydrateChoice.DIGEST, DehydrateChoice.AUTO]) +def test_hydration_follows_a_later_successor_by_design(tmp_path: Path, choice: str) -> None: + """Deprecated successors may change content; that is accepted product + behavior for both choices. The local copy covers availability, not drift.""" + out = tmp_path / "out.yaml" + result = _dehydrate(tmp_path, choice, _pipeline({"spec": LEAF}), Library({SPEC_DIGEST: LEAF})) + out.write_text(dump_yaml(result), encoding="utf-8") + successor = "7" * 64 + later = Library({SPEC_DIGEST: LEAF, successor: OTHER}, successors={SPEC_DIGEST: successor}) + + hydrated = PipelineHydrator(client=later).hydrate_file(out).data + + assert _leaf_digests(hydrated) == [compute_spec_digest(OTHER)] + + +# ---------------------------------------------------- hydrator: marked fallback + + +def _write_config(tmp_path: Path, config: Any) -> Path: + (tmp_path / "c.yaml").write_text(dump_yaml(LEAF), encoding="utf-8") + (tmp_path / "r.yaml").write_text(dump_yaml({"frag": config}), encoding="utf-8") + pipeline = tmp_path / "p.yaml" + pipeline.write_text(dump_yaml(_pipeline({"url": "resolve://./r.yaml#frag"})), encoding="utf-8") + return pipeline + + +@pytest.mark.parametrize( + ("label", "primary", "error"), + [ + ("digest-404", {"digest": SPEC_DIGEST}, None), + ("digest-network", {"digest": SPEC_DIGEST}, ConnectionError("down")), + ("name-network", {"name": "L", "publisher": "user-1"}, ConnectionError("down")), + ], +) +def test_a_marked_primary_that_raises_falls_through_to_the_local_copy( + tmp_path: Path, label: str, primary: dict[str, Any], error: Exception | None +) -> None: + pipeline = _write_config(tmp_path, [{**primary, "fallback_on_error": True}, {"local": "./c.yaml"}]) + + hydrated = PipelineHydrator(client=Library({}, error=error)).hydrate_file(pipeline).data + + assert _leaf_digests(hydrated) == [SPEC_DIGEST], label + + +@pytest.mark.parametrize( + ("label", "config"), + [ + ("unmarked-multi-entry", [{"digest": SPEC_DIGEST}, {"local": "./c.yaml"}]), + ( + "marked-last-entry", + [{"digest": SPEC_DIGEST, "fallback_on_error": True}, {"digest": SPEC_DIGEST, "fallback_on_error": True}], + ), + ("marked-single-entry", {"digest": SPEC_DIGEST, "fallback_on_error": True}), + ], +) +def test_errors_still_propagate_outside_the_generated_fallback_contract( + tmp_path: Path, label: str, config: Any +) -> None: + """Hand-authored, single and last entries keep today's loud failure.""" + pipeline = _write_config(tmp_path, config) + + with pytest.raises(ConnectionError): + PipelineHydrator(client=Library(error=ConnectionError("down"))).hydrate_file(pipeline) diff --git a/tests/test_dehydrator_subgraph_addressing.py b/tests/test_dehydrator_subgraph_addressing.py new file mode 100644 index 0000000..4f1ba3b --- /dev/null +++ b/tests/test_dehydrator_subgraph_addressing.py @@ -0,0 +1,163 @@ +"""Extracted subgraph files are addressed by their written content. + +A per-run counter (``_0.yaml``) let a second pipeline dehydrated into the +same directory overwrite the first one's subgraph with different content, so +the first output silently rehydrated to the second's pipeline. +""" + +from __future__ import annotations + +import shutil +from pathlib import Path +from typing import Any + +import pytest +import yaml + +from tangle_cli.pipeline_dehydrator import DehydrateChoice, PipelineDehydrator +from tangle_cli.pipeline_hydrator import PipelineHydrator +from tangle_cli.pipelines import dehydrate_pipeline_file +from tangle_cli.utils import dump_yaml + +MODES = ["auto", "file", "url"] + + +class NoLibrary: + """Nothing is published, so every leaf ends up local.""" + + def resolve_digest(self, digest: str) -> str: + return digest + + def get_component_spec(self, digest: str) -> Any: + raise KeyError(digest) + + def list_published_component_infos(self, *args: Any, **kwargs: Any) -> list[Any]: + return [] + + +def _leaf(image: str) -> dict[str, Any]: + return {"name": "L", "implementation": {"container": {"image": image}}} + + +def _graph(name: str, tasks: dict[str, Any], annotations: dict[str, str] | None = None) -> dict[str, Any]: + spec: dict[str, Any] = {"name": name, "implementation": {"graph": {"tasks": tasks}}} + if annotations: + spec["metadata"] = {"annotations": annotations} + return spec + + +def _ref(spec: dict[str, Any]) -> dict[str, Any]: + return {"componentRef": {"spec": spec}} + + +def _dehydrate(directory: Path, document: dict[str, Any], mode: str, out: str = "out.yaml") -> Path: + source = directory / f"{out}.in" + source.write_text(dump_yaml(document), encoding="utf-8") + dehydrate_pipeline_file(source, output=directory / out, mode=mode, client=NoLibrary()) + return directory / out + + +def _rehydrate(output: Path) -> dict[str, Any]: + return PipelineHydrator(client=NoLibrary()).hydrate_file(output).data + + +def _images(document: dict[str, Any]) -> list[str]: + found: list[str] = [] + for task in document["implementation"]["graph"]["tasks"].values(): + spec = task["componentRef"]["spec"] + if "graph" in spec["implementation"]: + found += _images(spec) + else: + found.append(spec["implementation"]["container"]["image"]) + return found + + +def _subgraph_files(directory: Path) -> list[str]: + return sorted(p.name for p in (directory / "subgraphs").glob("*")) + + +@pytest.mark.parametrize("mode", MODES) +def test_identical_subgraphs_share_one_file_and_reference(tmp_path: Path, mode: str) -> None: + judge = _graph("Judge", {"leaf": _ref(_leaf("i"))}) + out = _dehydrate(tmp_path, _graph("O", {"a": _ref(judge), "b": _ref(judge)}), mode) + + tasks = yaml.safe_load(out.read_text(encoding="utf-8"))["implementation"]["graph"]["tasks"] + assert len(_subgraph_files(tmp_path)) == 1 + assert tasks["a"]["componentRef"] == tasks["b"]["componentRef"] + + +@pytest.mark.parametrize("mode", MODES) +def test_a_provenance_only_difference_keeps_two_files(tmp_path: Path, mode: str) -> None: + """Like two generated sources of one Judge spec: different documents.""" + tasks = {"leaf": _ref(_leaf("i"))} + first = _graph("Judge", tasks, {"component_yaml_path": "judge-47d488e5.yaml"}) + second = _graph("Judge", tasks, {"component_yaml_path": "judge-f89931a0.yaml"}) + + _dehydrate(tmp_path, _graph("O", {"a": _ref(first), "b": _ref(second)}), mode) + + assert len(_subgraph_files(tmp_path)) == 2 + + +@pytest.mark.parametrize("mode", MODES) +def test_a_second_output_in_the_same_directory_cannot_mutate_the_first(tmp_path: Path, mode: str) -> None: + def pipeline(image: str) -> dict[str, Any]: + return _graph("O", {"s": _ref(_graph("Judge", {"leaf": _ref(_leaf(image))}))}) + + a = _dehydrate(tmp_path, pipeline("image-A"), mode, "a.yaml") + b = _dehydrate(tmp_path, pipeline("image-B"), mode, "b.yaml") + + assert _images(_rehydrate(a)) == ["image-A"] + assert _images(_rehydrate(b)) == ["image-B"] + + +@pytest.mark.parametrize("mode", MODES) +def test_deep_subgraphs_are_merkle_addressed_stable_and_relocatable(tmp_path: Path, mode: str) -> None: + """A deep leaf change renames every enclosing subgraph; identical input + reproduces identical names; the bundle rehydrates after relocation.""" + + def pipeline(image: str) -> dict[str, Any]: + inner = _graph("Inner", {"leaf": _ref(_leaf(image))}) + return _graph("O", {"mid": _ref(_graph("Mid", {"inner": _ref(inner)}))}) + + names = {} + for label, image in (("first", "x"), ("again", "x"), ("changed", "y")): + directory = tmp_path / label + directory.mkdir() + _dehydrate(directory, pipeline(image), mode) + names[label] = _subgraph_files(directory) + + assert names["first"] == names["again"] + assert len(names["first"]) == 2 + assert not set(names["first"]) & set(names["changed"]) + + moved = tmp_path / "moved" + shutil.copytree(tmp_path / "first", moved) + shutil.rmtree(tmp_path / "first") + assert _images(_rehydrate(moved / "out.yaml")) == ["x"] + + +def test_a_reused_instance_keeps_no_stale_subgraph_map_and_honors_the_extension(tmp_path: Path) -> None: + """Each output writes its own subgraph files, with the configured extension.""" + dehydrator = PipelineDehydrator({"": DehydrateChoice.FILE}, component_extension=".yml") + judge = _graph("Judge", {"leaf": _ref(_leaf("i"))}) + for index in range(2): + directory = tmp_path / f"out{index}" + directory.mkdir() + (directory / "in.yaml").write_text(dump_yaml(_graph("O", {"s": _ref(judge)})), encoding="utf-8") + dehydrator.dehydrate_file(directory / "in.yaml", directory / "out.yaml") + + (only,) = _subgraph_files(directory) + assert only.endswith(".yml") + assert "../" not in (directory / "out.yaml").read_text(encoding="utf-8") + + +def test_a_tampered_file_under_a_content_addressed_name_is_rewritten(tmp_path: Path) -> None: + """A name on disk is not trusted to still hold the content it addresses.""" + document = _graph("O", {"s": _ref(_graph("Judge", {"leaf": _ref(_leaf("i"))}))}) + out = _dehydrate(tmp_path, document, "file") + (name,) = _subgraph_files(tmp_path) + (tmp_path / "subgraphs" / name).write_text("name: tampered\n", encoding="utf-8") + + _dehydrate(tmp_path, document, "file") + + assert _images(_rehydrate(out)) == ["i"] diff --git a/tests/test_packaging.py b/tests/test_packaging.py index ad022be..fb1333a 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.27" in metadata + assert "Version: 0.1.28" 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_dehydrator.py b/tests/test_pipeline_dehydrator.py index e3187c5..57af762 100644 --- a/tests/test_pipeline_dehydrator.py +++ b/tests/test_pipeline_dehydrator.py @@ -68,7 +68,13 @@ def _extract(tmp_path: Path, tasks: dict[str, Any], **kwargs: Any) -> dict[str, class FakeClient: - def __init__(self, found: set[str]) -> None: + """A library whose published digests resolve to real specs. + + It returns the genuine spec rather than a placeholder: AUTO now checks + that a digest resolves to the component in hand, not merely that it exists. + """ + + def __init__(self, found: dict[str, dict[str, Any]]) -> None: self.found = found self.calls: list[str] = [] @@ -76,17 +82,32 @@ def get_component_spec(self, digest: str) -> dict[str, Any]: self.calls.append(digest) if digest not in self.found: raise KeyError(digest) - return {"name": "found"} + return self.found[digest] + + def resolve_digest(self, digest: str) -> str: + return digest + + def find_existing_components(self, *args: Any, **kwargs: Any) -> list[Any]: + return [] + + def list_published_component_infos(self, *args: Any, **kwargs: Any) -> list[Any]: + return [] def test_pipeline_dehydrator_replaces_refs_by_explicit_choice(tmp_path: Path) -> None: data = _pipeline({"task": _task("Leaf Component", "digest-1", canonical_url="https://example.test/leaf.yaml")}) - digest_result = PipelineDehydrator({"": DehydrateChoice.DIGEST}, output_file=tmp_path / "out.yaml").dehydrate(data) - assert digest_result["implementation"]["graph"]["tasks"]["task"]["componentRef"] == {"digest": "digest-1"} - - name_result = PipelineDehydrator({"": DehydrateChoice.NAME}, output_file=tmp_path / "out.yaml").dehydrate(data) - assert name_result["implementation"]["graph"]["tasks"]["task"]["componentRef"] == {"name": "Leaf Component"} + # DIGEST and NAME always emit a portable resolve ref; with nothing + # published the manifest fragment is the local copy alone. + for choice in (DehydrateChoice.DIGEST, DehydrateChoice.NAME): + result = PipelineDehydrator( + {"": choice}, output_file=tmp_path / "out.yaml", client=FakeClient({}) + ).dehydrate(data) + ref = result["implementation"]["graph"]["tasks"]["task"]["componentRef"] + assert ref["url"].startswith("resolve://./out.components.yaml#") + fragment = ref["url"].split("#", 1)[1] + manifest = yaml.safe_load((tmp_path / "out.components.yaml").read_text(encoding="utf-8")) + assert manifest[fragment] == [{"local": f"./components/{fragment}.yaml"}] url_result = PipelineDehydrator({"": DehydrateChoice.URL}, output_file=tmp_path / "out.yaml").dehydrate(data) assert url_result["implementation"]["graph"]["tasks"]["task"]["componentRef"] == { @@ -182,7 +203,7 @@ def fail_get_client(_self): def test_pipeline_dehydrator_auto_lazily_creates_client_for_library_lookup(tmp_path: Path) -> None: """Auto mode creates a default client only when a library lookup is needed.""" - client = FakeClient({"digest-found"}) + client = FakeClient({"digest-found": _leaf_spec("Published")}) class LazyDehydrator(PipelineDehydrator): def _get_client(self): @@ -228,7 +249,7 @@ def test_pipeline_dehydrator_auto_uses_url_digest_then_file(tmp_path: Path) -> N "local": _task("Local Only", "digest-missing"), } ) - client = FakeClient({"digest-found"}) + client = FakeClient({"digest-found": _leaf_spec("Published")}) result = PipelineDehydrator( {"": DehydrateChoice.AUTO}, @@ -255,11 +276,14 @@ def test_pipeline_dehydrator_auto_extracts_subgraphs_and_rewrites_relative_urls( result = PipelineDehydrator({"": DehydrateChoice.AUTO}, output_file=tmp_path / "out.yaml").dehydrate(data) - nested_ref = result["implementation"]["graph"]["tasks"]["nested"]["componentRef"] - assert nested_ref == {"url": "file://./subgraphs/nested_subgraph_0.yaml"} + from tangle_cli.utils import compute_spec_digest - subgraph_file = tmp_path / "subgraphs" / "nested_subgraph_0.yaml" + nested_ref = result["implementation"]["graph"]["tasks"]["nested"]["componentRef"] + (subgraph_file,) = (tmp_path / "subgraphs").glob("*.yaml") subgraph = yaml.safe_load(subgraph_file.read_text(encoding="utf-8")) + # The name addresses the subgraph as actually written, not a run counter. + assert subgraph_file.name == f"nested_subgraph-{compute_spec_digest(subgraph)}.yaml" + assert nested_ref == {"url": f"file://./subgraphs/{subgraph_file.name}"} extracted = _extracted_name("Inner Leaf") assert subgraph["implementation"]["graph"]["tasks"]["leaf"]["componentRef"] == { "url": f"file://./../components/{extracted}" diff --git a/tests/test_pipeline_runs_cli.py b/tests/test_pipeline_runs_cli.py index f6f854a..926fa60 100644 --- a/tests/test_pipeline_runs_cli.py +++ b/tests/test_pipeline_runs_cli.py @@ -969,7 +969,8 @@ def get_run_pipeline_spec(self, run_id: str) -> Any: def get_component_spec(self, digest: str) -> dict[str, Any]: self.spec_lookups.append(digest) - return {"name": "published"} + # The real published spec: AUTO verifies what a digest resolves to. + return {"name": "Step", "version": "1.0.0"} app = cli.build_app() fake_client = ExportClient() diff --git a/tests/test_pipelines_dehydrate_cli.py b/tests/test_pipelines_dehydrate_cli.py new file mode 100644 index 0000000..dd60b56 --- /dev/null +++ b/tests/test_pipelines_dehydrate_cli.py @@ -0,0 +1,341 @@ +"""`tangle sdk pipelines dehydrate` drives the shared dehydrator end to end. + +The API client is the REAL ``TangleApiClient`` stubbed only at its HTTP seams, +so publication verification and inspector-backed NAME exercise production +row/spec shapes rather than pre-enriched fixtures. +""" + +from __future__ import annotations + +import copy +from pathlib import Path +from typing import Any + +import pytest +import requests +import yaml + +from tangle_cli import cli, pipelines_cli +from tangle_cli.utils import compute_spec_digest, dump_yaml + +LEAF = {"name": "L", "implementation": {"container": {"image": "i", "command": ["c"]}}} +DIGEST = compute_spec_digest(LEAF) + + +def _write_pipeline(tmp_path: Path) -> Path: + source = tmp_path / "in.yaml" + source.write_text( + dump_yaml({"name": "O", "implementation": {"graph": {"tasks": {"t": {"componentRef": {"spec": LEAF}}}}}}), + encoding="utf-8", + ) + return source + + +@pytest.fixture +def library(monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]: + """Configure what the stubbed real client publishes; records CLI plumbing.""" + from tangle_cli.client import TangleApiClient + from tangle_cli.models import ComponentSpec + + state: dict[str, Any] = {"specs": {}, "rows": [], "error": None, "kwargs": None} + + def factory(**kwargs: Any) -> Any: + state["kwargs"] = kwargs + client = TangleApiClient(base_url="https://review.invalid", include_env_credentials=False) + + def get_component_spec(digest: str) -> Any: + if state["error"] is not None: + raise state["error"] + if digest not in state["specs"]: + raise KeyError(digest) + return ComponentSpec.from_dict(copy.deepcopy(state["specs"][digest])) + + client.get_component_spec = get_component_spec + client._published_component_rows = lambda **kw: [ + dict(row) for row in state["rows"] if kw.get("digest") in (None, row["digest"]) + ] + return client + + monkeypatch.setattr(pipelines_cli, "LazyTangleApiClient", factory) + return state + + +def _run(args: list[str]) -> None: + cli.build_app()(["sdk", "pipelines", "dehydrate", *args]) + + +def _succeed(args: list[str]) -> None: + try: + _run(args) + except SystemExit as exc: + assert exc.code in (0, None), exc.code + + +def _task_ref(output: Path) -> dict[str, Any]: + return yaml.safe_load(output.read_text(encoding="utf-8"))["implementation"]["graph"]["tasks"]["t"]["componentRef"] + + +def _fragment(output: Path) -> list[dict[str, Any]]: + url = _task_ref(output)["url"] + manifest = yaml.safe_load(output.with_name(f"{output.stem}.components.yaml").read_text(encoding="utf-8")) + return manifest[url.split("#", 1)[1]] + + +def test_default_auto_emits_a_verified_published_digest(tmp_path: Path, library: dict[str, Any], capsys) -> None: + library["specs"][DIGEST] = LEAF + output = tmp_path / "out.yaml" + + _succeed([str(_write_pipeline(tmp_path)), "-o", str(output), "--base-url", "https://x.test", "--token", "t"]) + + assert _task_ref(output) == {"digest": DIGEST} + assert "(mode: auto)" in capsys.readouterr().out + assert library["kwargs"]["base_url"] == "https://x.test" + assert library["kwargs"]["token"] == "t" + + +def test_default_auto_extracts_an_unpublished_component(tmp_path: Path, library: dict[str, Any], capsys) -> None: + output = tmp_path / "out.yaml" + + _succeed([str(_write_pipeline(tmp_path)), "-o", str(output)]) + + assert _task_ref(output)["url"].startswith("file://./components/") + assert "Components: 1 file(s)" in capsys.readouterr().out + + +def test_digest_mode_writes_a_portable_resolve_config(tmp_path: Path, library: dict[str, Any], capsys) -> None: + library["specs"][DIGEST] = LEAF + output = tmp_path / "out.yaml" + + _succeed([str(_write_pipeline(tmp_path)), "-o", str(output), "--mode", "digest"]) + + fragment = _fragment(output) + assert fragment[0] == {"digest": DIGEST, "fallback_on_error": True} + assert (tmp_path / fragment[1]["local"]).is_file() + assert "Resolve config:" in capsys.readouterr().out + + +def test_name_mode_pins_the_inspected_owner(tmp_path: Path, library: dict[str, Any]) -> None: + library["specs"][DIGEST] = LEAF + library["rows"].append({"name": "L", "digest": DIGEST, "published_by": "user-123"}) + output = tmp_path / "out.yaml" + + _succeed([str(_write_pipeline(tmp_path)), "-o", str(output), "--mode", "name"]) + + assert _fragment(output)[0] == {"name": "L", "publisher": "user-123", "fallback_on_error": True} + + +def test_an_unreachable_library_degrades_to_the_local_copy(tmp_path: Path, library: dict[str, Any]) -> None: + library["error"] = requests.ConnectionError("unreachable") + output = tmp_path / "out.yaml" + + _succeed([str(_write_pipeline(tmp_path)), "-o", str(output), "--mode", "digest"]) + + fragment = _fragment(output) + assert len(fragment) == 1 and set(fragment[0]) == {"local"} + + +def test_components_dir_override_is_honored(tmp_path: Path, library: dict[str, Any]) -> None: + output = tmp_path / "out.yaml" + + _succeed([str(_write_pipeline(tmp_path)), "-o", str(output), "--mode", "file", "--components-dir", str(tmp_path / "bundle")]) + + assert list((tmp_path / "bundle").glob("*.yaml")) + assert not (tmp_path / "components").exists() + + +@pytest.mark.parametrize( + ("label", "args"), + [ + ("unknown-mode", ["--mode", "bogus"]), + ("missing-output", None), + ("missing-input", ["--missing-input"]), + ], +) +def test_invalid_invocations_exit_non_zero_without_writing( + tmp_path: Path, library: dict[str, Any], label: str, args: list[str] | None +) -> None: + source = _write_pipeline(tmp_path) + output = tmp_path / "out.yaml" + if args is None: + argv = [str(source)] + elif args == ["--missing-input"]: + argv = [str(tmp_path / "nope.yaml"), "-o", str(output)] + else: + argv = [str(source), "-o", str(output), *args] + + with pytest.raises(SystemExit) as excinfo: + _run(argv) + + assert excinfo.value.code not in (0, None), label + assert not output.exists(), label + + +def test_a_malformed_header_fails_loudly_instead_of_degrading( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Misconfiguration is not an outage: it must not silently ship local-only.""" + monkeypatch.setenv("TANGLE_API_URL", "https://review.invalid") + output = tmp_path / "out.yaml" + + with pytest.raises(SystemExit) as excinfo: + _run([str(_write_pipeline(tmp_path)), "-o", str(output), "--mode", "digest", "--header", "no-colon"]) + + assert excinfo.value.code not in (0, None) + assert "header" in str(excinfo.value.code).lower() + + +# ------------------------------------------------------------ nested graphs + + +def _nested(leaf: dict[str, Any]) -> dict[str, Any]: + inner = {"name": "Inner", "implementation": {"graph": {"tasks": {"leaf": {"componentRef": {"spec": leaf}}}}}} + return {"name": "O", "implementation": {"graph": {"tasks": {"sub": {"componentRef": {"spec": inner}}}}}} + + +def _leaf_digests(document: dict[str, Any]) -> list[str]: + found: list[str] = [] + for task in document["implementation"]["graph"]["tasks"].values(): + spec = task["componentRef"]["spec"] + if "graph" in spec["implementation"]: + found += _leaf_digests(spec) + else: + found.append(compute_spec_digest({k: v for k, v in spec.items() if not k.startswith("_")})) + return found + + +@pytest.mark.parametrize("mode", ["file", "url"]) +def test_a_nested_graph_is_fully_dehydrated_portable_and_rehydrates( + tmp_path: Path, library: dict[str, Any], mode: str +) -> None: + """A public dehydrate must never return a partially hydrated document. + + The whole bundle is relocated before rehydrating, offline, so every ref + written -- root, extracted subgraph and resolve config -- must be relative. + """ + import shutil + + from tangle_cli.pipeline_hydrator import PipelineHydrator + from tangle_cli.schema_validation import validate_dehydrated_pipeline + + work = tmp_path / "work" + work.mkdir() + document = _nested(LEAF) + (work / "in.yaml").write_text(dump_yaml(document), encoding="utf-8") + _succeed([str(work / "in.yaml"), "-o", str(work / "out.yaml"), "--mode", mode]) + + written = [p for p in work.rglob("*.yaml") if p.name not in {"in.yaml", "out.components.yaml"}] + for path in written: + tasks = yaml.safe_load(path.read_text(encoding="utf-8")).get("implementation", {}).get("graph", {}).get("tasks", {}) + for task in tasks.values(): + assert "spec" not in task["componentRef"], f"inline spec left in {path.name}" + validate_dehydrated_pipeline(yaml.safe_load((work / "out.yaml").read_text(encoding="utf-8"))) + + moved = tmp_path / "moved" + shutil.copytree(work, moved) + shutil.rmtree(work) + + class Offline: + def resolve_digest(self, digest: str) -> str: + return digest + + def get_component_spec(self, digest: str) -> Any: + raise requests.ConnectionError("offline") + + hydrated = PipelineHydrator(client=Offline()).hydrate_file(moved / "out.yaml").data + assert _leaf_digests(hydrated) == _leaf_digests(document) + + +@pytest.mark.parametrize(("mode", "kept_url"), [("url", True), ("file", False)]) +def test_leaf_mode_behavior_holds_inside_an_extracted_subgraph( + tmp_path: Path, library: dict[str, Any], mode: str, kept_url: bool +) -> None: + """Extracting the boundary must not change how the leaves inside it are + replaced: url keeps a canonical URL, file always writes a local file.""" + canonical = "https://example.test/leaf.yaml" + leaf = {**LEAF, "metadata": {"annotations": {"canonical_location": canonical}}} + (tmp_path / "in.yaml").write_text(dump_yaml(_nested(leaf)), encoding="utf-8") + + _succeed([str(tmp_path / "in.yaml"), "-o", str(tmp_path / "out.yaml"), "--mode", mode]) + + (subgraph,) = (tmp_path / "subgraphs").glob("*.yaml") + leaf_ref = yaml.safe_load(subgraph.read_text(encoding="utf-8"))["implementation"]["graph"]["tasks"]["leaf"]["componentRef"] + if kept_url: + assert leaf_ref == {"url": canonical} + else: + assert leaf_ref["url"].startswith("file://") and leaf_ref["url"] != canonical + + +# ---------------------------------------------------------- input boundaries + + +@pytest.mark.parametrize("text", ["null", "false", "0", "''", "[]", "just a string"]) +def test_a_non_mapping_pipeline_is_refused_without_writing( + tmp_path: Path, library: dict[str, Any], monkeypatch: pytest.MonkeyPatch, text: str +) -> None: + """Falsey YAML must not be read as an empty pipeline and written as ``{}``.""" + monkeypatch.chdir(tmp_path) + (tmp_path / "in.yaml").write_text(text + "\n", encoding="utf-8") + + with pytest.raises(SystemExit) as excinfo: + _run(["in.yaml", "-o", "out.yaml"]) + + assert excinfo.value.code not in (0, None) + assert sorted(p.name for p in tmp_path.iterdir()) == ["in.yaml"] + + +@pytest.mark.parametrize("uri", ["gs://bucket/x.yaml", "https://example.test/x.yaml", "file:///tmp/x.yaml"]) +@pytest.mark.parametrize("position", ["input", "output", "components-dir"]) +def test_a_uri_is_refused_before_path_can_mangle_it_into_a_local_write( + tmp_path: Path, library: dict[str, Any], monkeypatch: pytest.MonkeyPatch, uri: str, position: str +) -> None: + """Path("gs://b/x") is the LOCAL path gs:/b/x; the command must refuse the + URI instead of silently writing a gs:/... tree under the cwd.""" + monkeypatch.chdir(tmp_path) + _write_pipeline(tmp_path) + argv = { + "input": [uri, "-o", "out.yaml"], + "output": ["in.yaml", "-o", uri], + "components-dir": ["in.yaml", "-o", "out.yaml", "--mode", "file", "--components-dir", uri], + }[position] + + with pytest.raises(SystemExit) as excinfo: + _run(argv) + + assert excinfo.value.code not in (0, None) + assert "local path" in str(excinfo.value.code) + assert sorted(p.name for p in tmp_path.iterdir()) == ["in.yaml"] + + +@pytest.mark.parametrize("explicit", [True, False]) +def test_the_standalone_helper_verifies_against_an_explicit_base_url( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, explicit: bool +) -> None: + """With no client injected, the lazily created client must use the + requested endpoint, not the ambient TANGLE_API_URL.""" + from tangle_cli.pipelines import dehydrate_pipeline_file + + monkeypatch.setenv("TANGLE_API_URL", "https://ambient.invalid") + hosts: list[str] = [] + + def record(self: Any, method: str, url: str, **kwargs: Any) -> Any: + hosts.append(url.split("/")[2]) + # A real, non-retryable 404: a ConnectionError would be retried with + # backoff until the client's deadline (about a minute). + response = requests.Response() + response.status_code = 404 + response.url = url + response.headers["Content-Type"] = "application/json" + response._content = b'{"detail": "not found"}' + return response + + monkeypatch.setattr(requests.Session, "request", record) + + dehydrate_pipeline_file( + _write_pipeline(tmp_path), + output=tmp_path / "out.yaml", + mode="digest", + base_url="https://requested.invalid" if explicit else None, + ) + + assert hosts, "verification never reached the client" + assert set(hosts) == {"requested.invalid" if explicit else "ambient.invalid"} diff --git a/uv.lock b/uv.lock index 6b51d98..7d562b1 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.27" +version = "0.1.28" source = { editable = "." } dependencies = [ { name = "cloud-pipelines" },