diff --git a/README.md b/README.md index af9d42d..64d7f94 100644 --- a/README.md +++ b/README.md @@ -353,7 +353,36 @@ uv run tangle sdk published-components publish components/my-component.yaml --dr uv run tangle sdk published-components deprecate sha256:old --superseded-by sha256:new ``` -`publish` accepts `--image`, `--name`, `--description`, `--annotations` (JSON), `--dry-run`, `--published-by`, generic git metadata fields, generic API auth fields, `--log-type`, and `--config`. By default it scopes version checks and automatic old-version deprecation to the current authenticated user via `users_me()`; use `--published-by` to supply an explicit owner/publisher filter. Publishing fails closed if no owner can be determined. +`publish` accepts `--image`, `--name`, `--description`, `--annotations` (JSON), `--dry-run`, `--allow-downgrade`, `--published-by`, generic git metadata fields, generic API auth fields, `--log-type`, and `--config`. By default it scopes version checks and automatic old-version deprecation to the current authenticated user via `users_me()`; use `--published-by` to supply an explicit owner/publisher filter. Publishing fails closed if no owner can be determined. + +#### Monotonic publishing and result digests + +Publishing is monotonic against the highest **non-deprecated, owner-scoped** published version of the component (ordering comes from `compare_versions`, which zero-pads shorter versions so `1.0.1 > 1.0`): + +| Local vs latest published | Outcome | Notes | +| --- | --- | --- | +| nothing published (no non-deprecated owner-scoped version) | `proceed` | first publish | +| local strictly newer | `proceed` | publishes, then deprecates owner-scoped versions proven older | +| local equal | `skip` | no create/deprecate calls | +| local strictly older | `skip` | no-op; never publishes an older version and never deprecates a newer one | +| published version unreadable, or ambiguous tie at the latest version | `error` | fails closed; no create/deprecate calls | + +Every result carries the digest of the version it compared against: + +- `digest` — digest of a **newly created** publication (SUCCESS only, unchanged meaning). +- `latest_digest` — exact digest of the selected latest published version (set on PROCEED/SKIP, and carried through the SUCCESS/ERROR results that follow a version check). JSON output includes it as `latest_digest`. +- `ProcessingResult.resolved_digest` — the digest a caller should pin: `digest or latest_digest`, but deliberately `None` for any outcome other than SUCCESS/SKIP, so a failed publish never hands back a stale-but-plausible digest. + +The check **fails closed** (an `error`, with nothing published and nothing deprecated) whenever the published state cannot be read completely: + +- any non-deprecated owner-scoped candidate whose digest is missing, or whose spec/version cannot be fetched or parsed — an unreadable row could be newer than the local version, and must never be deprecated sight-unseen; +- two or more non-deprecated candidates tied at the selected latest version, where no exact digest can be chosen. The reason names the tied digests instead of guessing from API ordering. + +Deprecated components are never selected as "latest", and all digest lists in results/logs are sorted, so diagnostics do not depend on API response order. + +The published state is re-read immediately before create, and the same policy is re-applied to that fresh observation: a version that appeared concurrently since the first check can still turn the publish into a skip or an error. After a successful create, only digests **proven strictly older** in that final observation are deprecated — a row first seen after the publish decision is never deprecated on the strength of the earlier one. A race after the final read is not preventable client-side and needs a server-side conditional/CAS operation. + +**Contract change:** republishing an older version used to proceed (publishing the older spec and deprecating the newer one). It is now a skip. `--allow-downgrade` publishes the older spec but still never deprecates a strictly newer row; deprecate those explicitly with `published-components deprecate` if that is really intended. Deliberate downgrades must opt in with `--allow-downgrade` on the CLI, or `ComponentPublisher(allow_downgrade=True)` / `allow_downgrade=True` on the `publish_component_to_tangle` / `perform_version_check` wrappers. Republishing the same version is still a skip, as before. There is no separate OSS `publish-all` command. To publish multiple components, pass a YAML/JSON config list, or `_defaults` + `configs`, to the same `published-components publish` command; the command aggregates results and exits nonzero if any component errors. A top-level `_select` node can choose between such documents per environment (see [Environment-selected configs](#environment-selected-configs-_select)). diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py index a4243cf..f087eea 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.14" + __version__ = "0.1.15" __all__ = ["TangleDynamicDiscoveryClient", "__version__"] diff --git a/packages/tangle-cli/src/tangle_cli/authenticated_identity.py b/packages/tangle-cli/src/tangle_cli/authenticated_identity.py new file mode 100644 index 0000000..9e65edd --- /dev/null +++ b/packages/tangle-cli/src/tangle_cli/authenticated_identity.py @@ -0,0 +1,77 @@ +"""Resolution of the authenticated account, and the symbolic ``me`` publisher. + +Compilation is offline, so a compiler cannot write the author's account id into +a component entry. It writes the symbolic publisher :data:`ME` instead, and the +account is resolved at hydration time by whoever is actually authenticated. + +Two callers need the account: the publisher, which scopes its version check and +deprecation to the owner, and hydration of entries whose publisher is ``me``. +They must agree, so the parsing lives here once. + +They differ only in what an *unknown* account means, so that choice is left to +the caller: the publisher degrades, while resolution must fail closed -- see +:func:`require_authenticated_user_id`. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +#: Symbolic publisher meaning "whoever is authenticated at hydration time". +#: Matched exactly and case-sensitively, so a literal account id that happens to +#: differ in case is never mistaken for the sentinel. +ME = "me" + +__all__ = [ + "ME", + "IdentityUnavailableError", + "authenticated_user_id", + "is_symbolic_me", + "require_authenticated_user_id", +] + + +class IdentityUnavailableError(RuntimeError): + """The authenticated account could not be determined.""" + + +def is_symbolic_me(publisher: Any) -> bool: + """Whether ``publisher`` is the symbolic self-reference rather than an id.""" + return publisher == ME + + +def authenticated_user_id(client: Any) -> str | None: + """Return the current user id, or ``None`` if it cannot be read. + + An empty or missing id is as unusable as no answer at all. + """ + try: + user_info = client.users_me() + except Exception: + return None + if user_info is None: + return None + if isinstance(user_info, Mapping): + value = user_info.get("id") + else: + value = getattr(user_info, "id", None) + return str(value) if value else None + + +def require_authenticated_user_id(client: Any) -> str: + """Return the current user id, raising when it cannot be determined. + + For owner-scoped *resolution* an unknown account must never widen the + search: continuing unscoped is what would let a component published by + someone else under the same name be resolved as the author's own code. + """ + user_id = authenticated_user_id(client) + if not user_id: + raise IdentityUnavailableError( + "Cannot determine the authenticated account, so a component " + f"published by '{ME}' cannot be resolved. Refusing to fall back to " + "an unscoped lookup, which could resolve a component owned by " + "someone else. Re-authenticate and retry." + ) + return user_id diff --git a/packages/tangle-cli/src/tangle_cli/client.py b/packages/tangle-cli/src/tangle_cli/client.py index 2c15a9b..5f07831 100644 --- a/packages/tangle-cli/src/tangle_cli/client.py +++ b/packages/tangle-cli/src/tangle_cli/client.py @@ -952,12 +952,22 @@ def find_existing_components( search_digests.add(str(data["digest"])) publisher_filter = published_by_substring or published_by + # The API parameter is a SUBSTRING match. When the caller asked for an + # exact owner it is therefore only a prefilter, and accepting its + # results verbatim would let a superset owner id (``alice`` matching + # ``alice2``) satisfy an exact request. An owner-scoped lookup is an + # identity control, so exactness is enforced here on the returned rows + # and a row with no owner is never accepted. Explicit + # ``published_by_substring`` callers keep substring semantics. + exact_owner = published_by if published_by and not published_by_substring else None found: dict[str, ComponentInfo] = {} def add(info: ComponentInfo) -> None: key = info.digest or info.name if not key: return + if exact_owner is not None and info.published_by != exact_owner: + return found[key] = info if verbose: self.logger.info(f" Found existing component: {info.name} ({key[:16]}...)") diff --git a/packages/tangle-cli/src/tangle_cli/component_from_func.py b/packages/tangle-cli/src/tangle_cli/component_from_func.py index df246b8..7ea20f3 100644 --- a/packages/tangle-cli/src/tangle_cli/component_from_func.py +++ b/packages/tangle-cli/src/tangle_cli/component_from_func.py @@ -986,8 +986,13 @@ def _is_main_str(n: ast.expr) -> bool: # _strip_authoring_constructs). ``registered`` marks an op published separately # via its own gen_config.yaml; when that same op is baked (through its # local_from_python entry) the decorator + its authoring import must be stripped -# too, exactly like @task. -_AUTHORING_DECORATOR_NAMES = frozenset({"task", "pipeline", "subpipeline", "registered"}) +# too, exactly like @task. ``Publish`` likewise only records a publication +# declaration for the compiler; leaving it in the baked program would raise +# ``NameError`` at container startup, since its import is stripped with the +# rest of the authoring surface. +_AUTHORING_DECORATOR_NAMES = frozenset( + {"task", "pipeline", "subpipeline", "registered", "Publish"} +) # The python-pipeline authoring modules. ONLY imports of these modules (and # their submodules) are authoring-only and stripped from the baked source. We diff --git a/packages/tangle-cli/src/tangle_cli/component_publisher.py b/packages/tangle-cli/src/tangle_cli/component_publisher.py index 5db339d..24317f2 100644 --- a/packages/tangle-cli/src/tangle_cli/component_publisher.py +++ b/packages/tangle-cli/src/tangle_cli/component_publisher.py @@ -18,6 +18,7 @@ import tangle_cli.utils as utils +from .authenticated_identity import authenticated_user_id from .handler import TangleCliHandler from .logger import Logger @@ -34,9 +35,78 @@ class ProcessingOutcome(str, Enum): ERROR = "error" +@dataclass(frozen=True) +class _PublishedState: + """One observation of the owner-scoped published state of a component. + + ``versions`` maps digest → verified published version for non-deprecated + candidates. ``unreadable`` holds identifiers of non-deprecated candidates + whose version could not be determined; any entry there means the view is + incomplete and callers must fail closed rather than publish/deprecate. + Derived digest lists are sorted so diagnostics never depend on API order. + """ + + versions: Mapping[str, str] = field(default_factory=dict) + unreadable: tuple[str, ...] = () + deprecated_count: int = 0 + found_count: int = 0 + #: Dry-run ``TEST_LATEST_VERSION`` override. Synthetic: it has no backing + #: published component, so it never contributes a digest. + synthetic_latest_version: str | None = None + + @property + def latest_version(self) -> str | None: + """Highest verified published version, deterministic across API orders.""" + + if not self.versions: + return self.synthetic_latest_version + + latest: str | None = None + for version in sorted(set(self.versions.values())): + if latest is None or utils.compare_versions(version, latest) > 0: + latest = version + return latest + + @property + def latest_digests(self) -> tuple[str, ...]: + """Sorted digests whose version compares equal to :attr:`latest_version`.""" + + latest = self.latest_version + if latest is None: + return () + return tuple( + sorted( + digest + for digest, version in self.versions.items() + if utils.compare_versions(version, latest) == 0 + ) + ) + + def digests_older_than(self, local_version: str) -> tuple[str, ...]: + """Sorted digests proven strictly older than ``local_version``.""" + + return tuple( + sorted( + digest + for digest, version in self.versions.items() + if utils.compare_versions(version, local_version) < 0 + ) + ) + + @dataclass class ProcessingResult: - """Result for one component publish/deprecate processing step.""" + """Result for one component publish/deprecate processing step. + + ``digest`` is the digest of a *newly created* publication and is therefore + only populated on :attr:`ProcessingOutcome.SUCCESS`. ``latest_digest`` is + the exact digest of the selected latest non-deprecated owner-scoped + published version that the local component was compared against; it is + populated on PROCEED/SKIP (and carried through SUCCESS/ERROR results that + happen after a version check) whenever it can be resolved unambiguously. + Callers that want "the digest to pin, whatever happened" should use + :attr:`resolved_digest`. + """ outcome: ProcessingOutcome local_version: str | None = None @@ -44,7 +114,25 @@ class ProcessingResult: spec: Any = None reason: str | None = None digest: str | None = None + latest_digest: str | None = None response: Any = None + #: Verified digest → version snapshot the decision was made from. In-process + #: detail for callers/tests; deliberately not emitted by :meth:`to_dict`. + published_versions: Mapping[str, str] = field(default_factory=dict) + + @property + def resolved_digest(self) -> str | None: + """Digest a caller can pin: newly published digest, else latest published. + + Deliberately ``None`` for any outcome other than SUCCESS/SKIP: after an + ERROR (including a publish request that raised and may still have + landed server-side) the correct digest is unknown, so this must not + hand back the pre-existing one. + """ + + if self.outcome not in (ProcessingOutcome.SUCCESS, ProcessingOutcome.SKIP): + return None + return self.digest or self.latest_digest def to_dict(self) -> dict[str, Any]: payload: dict[str, Any] = { @@ -54,6 +142,7 @@ def to_dict(self) -> dict[str, Any]: "latest_version": self.latest_version, "reason": self.reason, "digest": self.digest, + "latest_digest": self.latest_digest, "response": _to_plain(self.response), } if self.spec is not None: @@ -139,6 +228,7 @@ def __init__( hooks: Sequence[ComponentPublishHook] | None = None, logger: Logger | None = None, base_url: str | None = None, + allow_downgrade: bool = False, ) -> None: """Initialize the ComponentPublisher. @@ -147,6 +237,11 @@ def __init__( ``client_factory`` is a downstream seam for lazily constructing a custom authenticated client; subclasses may also override :meth:`_get_client` for more control. + + ``allow_downgrade`` is an explicit opt-out from monotonic publishing. + By default a local version older than the latest published + owner-scoped version is a no-op SKIP; set it to ``True`` only for a + deliberate manual downgrade/republish. """ super().__init__( @@ -157,6 +252,7 @@ def __init__( base_url=base_url, ) self.published_by = published_by + self.allow_downgrade = allow_downgrade self.hooks = list(hooks or []) self.results: list[tuple[str, ProcessingResult]] = [] @@ -190,98 +286,218 @@ def component_digest(self, component: Any) -> str | None: digest = getattr(component, "digest", None) return str(digest) if digest else None + def component_is_deprecated(self, component: Any) -> bool: + """Return whether a published component is marked deprecated. + + Listing is already non-deprecated by default; this is a defensive + filter so deprecated entries can never be selected as "latest". + """ + + if isinstance(component, Mapping): + value = component.get("deprecated") + else: + value = getattr(component, "deprecated", None) + return bool(value) + def current_user_id(self, client: Any) -> str | None: """Return the current Tangle user id for owner-scoped lookups.""" - try: - user_info = client.users_me() - except Exception: - return None - if user_info is None: - return None - if isinstance(user_info, Mapping): - value = user_info.get("id") - else: - value = getattr(user_info, "id", None) - return str(value) if value else None + return authenticated_user_id(client) def perform_version_check(self, spec: Any) -> ProcessingResult: - """Perform owner-scoped version checking for a component. + """Perform owner-scoped, monotonic version checking for a component. If ``published_by`` is omitted, the current authenticated user is resolved via ``client.users_me().id``. Failure to determine an owner is an error so callers do not accidentally compare/deprecate components owned by others. + + Ordering is defined by :func:`tangle_cli.utils.compare_versions` + against the highest non-deprecated owner-scoped published version: + + * nothing published → PROCEED + * local strictly newer → PROCEED + * local equal → SKIP, carrying that version's exact digest + * local strictly older → SKIP (no-op), carrying the newer published + digest; never publishes an older version and never deprecates a + newer one. ``allow_downgrade=True`` opts out of this last rule. + + The check fails closed — ERROR, and callers must not publish or + deprecate anything — when the published state cannot be read + completely: + + * any non-deprecated owner-scoped candidate whose digest is missing or + whose spec/version cannot be fetched or parsed (an unreadable row + could be newer than the local version, and would later be deprecated + on a blind "deprecate everything" pass); + * several non-deprecated candidates tied at the selected latest + version, where no exact digest can be chosen. + + Diagnostics list digests in sorted order so results never depend on + API response ordering. """ local_version = spec.version self.log.info(f" Local version: {local_version}") - latest_version = None - if self.dry_run: test_version = os.environ.get("TEST_LATEST_VERSION") + state = _PublishedState() if test_version: - latest_version = test_version - self.log.info(f" Remote version (test): {latest_version}") - else: - client = self._get_client() - if client is None: - return ProcessingResult( - outcome=ProcessingOutcome.ERROR, - local_version=str(local_version), - latest_version=None, - reason="Failed to create API client", - ) + state = _PublishedState(synthetic_latest_version=test_version) + self.log.info(f" Remote version (test): {test_version}") + return self._evaluate_published_state(spec, state) - filter_by = self.published_by or self.current_user_id(client) - if not filter_by: - self.log.error( - "❌ Cannot determine current user — aborting to avoid deprecating components owned by others" - ) - return ProcessingResult( - outcome=ProcessingOutcome.ERROR, - local_version=str(local_version), - latest_version=None, - reason="Cannot determine current user for author filtering", - ) + client = self._get_client() + if client is None: + return ProcessingResult( + outcome=ProcessingOutcome.ERROR, + local_version=str(local_version), + latest_version=None, + reason="Failed to create API client", + ) - existing_components = client.find_existing_components( - spec.search_names, - verbose=False, - published_by=filter_by, + filter_by = self.owner_filter(client) + if not filter_by: + return ProcessingResult( + outcome=ProcessingOutcome.ERROR, + local_version=str(local_version), + latest_version=None, + reason="Cannot determine current user for author filtering", + ) + + state = self.collect_published_state(client, spec, filter_by, verbose=False) + return self._evaluate_published_state(spec, state) + + def owner_filter(self, client: Any) -> str | None: + """Resolve the owner scope for lookups/deprecation, logging on failure.""" + + filter_by = self.published_by or self.current_user_id(client) + if not filter_by: + self.log.error( + "❌ Cannot determine current user — aborting to avoid deprecating components owned by others" ) + return filter_by - if existing_components: - for component in existing_components: - digest = self.component_digest(component) - if not digest: - continue - try: - full_spec = client.get_component_spec(digest) - remote_version = full_spec.version if full_spec else None - if remote_version and ( - not latest_version or utils.compare_versions(remote_version, latest_version) > 0 - ): - latest_version = remote_version - except Exception as exc: - self.log.warn(f" Warning: Failed to get version for component {digest[:16]}: {exc}") - continue - - if latest_version: - self.log.info(f" Remote version: {latest_version}") + def collect_published_state( + self, + client: Any, + spec: Any, + filter_by: str, + *, + verbose: bool = False, + ) -> "_PublishedState": + """Read the owner-scoped published state for ``spec`` exactly once. + + Returns a verified digest → version snapshot plus the identifiers of + any non-deprecated candidate that could not be read. Callers decide + policy; this method only observes. + """ + + existing_components = client.find_existing_components( + spec.search_names, + verbose=verbose, + published_by=filter_by, + ) + + versions: dict[str, str] = {} + unreadable: list[str] = [] + deprecated_count = 0 + + for component in existing_components or []: + if self.component_is_deprecated(component): + deprecated_count += 1 + continue + digest = self.component_digest(component) + if not digest: + name = None + if isinstance(component, Mapping): + name = component.get("name") else: - self.log.info( - f" ℹ️ Found {len(existing_components)} component(s) but couldn't extract version" - ) + name = getattr(component, "name", None) + unreadable.append(f"") + continue + try: + full_spec = client.get_component_spec(digest) + remote_version = full_spec.version if full_spec else None + except Exception as exc: + self.log.warn(f" Warning: Failed to get version for component {digest[:16]}: {exc}") + unreadable.append(digest) + continue + if not remote_version: + unreadable.append(digest) + continue + versions[digest] = str(remote_version) + + return _PublishedState( + versions=versions, + unreadable=tuple(sorted(unreadable)), + deprecated_count=deprecated_count, + found_count=len(existing_components or []), + ) + + def _evaluate_published_state( + self, + spec: Any, + state: "_PublishedState", + *, + stage: str = "version check", + ) -> ProcessingResult: + """Apply the monotonic publish policy to an observed published state.""" + + local_version = spec.version + prefix = "" if stage == "version check" else f"({stage}) " + + if state.unreadable: + listed = ", ".join(state.unreadable) + reason = ( + f"Cannot read published version for {len(state.unreadable)} non-deprecated " + f"component(s) ({listed}); refusing to publish or deprecate without a complete view" + ) + self.log.error(f" ❌ {prefix}{reason}") + return ProcessingResult( + outcome=ProcessingOutcome.ERROR, + local_version=local_version, + latest_version=None, + spec=spec, + reason=reason, + published_versions=state.versions, + ) + + latest_version = state.latest_version + latest_digests = state.latest_digests + + if state.versions: + self.log.info(f" Remote version: {latest_version}") + elif state.found_count and state.deprecated_count == state.found_count: + self.log.info(f" ℹ️ Found {state.found_count} component(s), all deprecated") + + if len(latest_digests) > 1: + tied = ", ".join(latest_digests) + reason = ( + f"Ambiguous latest published version {latest_version}: " + f"{len(latest_digests)} non-deprecated components share it ({tied}); " + "cannot select an exact digest" + ) + self.log.error(f" ❌ {prefix}{reason}") + return ProcessingResult( + outcome=ProcessingOutcome.ERROR, + local_version=local_version, + latest_version=latest_version, + spec=spec, + reason=reason, + published_versions=state.versions, + ) + + latest_digest = latest_digests[0] if latest_digests else None - should_proceed = not latest_version or utils.compare_versions(local_version, latest_version) != 0 + comparison = 0 if latest_version is None else utils.compare_versions(local_version, latest_version) + should_proceed = latest_version is None or comparison > 0 or (comparison < 0 and self.allow_downgrade) if should_proceed: - is_older = latest_version is not None and utils.compare_versions(latest_version, local_version) > 0 - version_suffix = " (older)" if is_older else "" + version_suffix = " (older)" if comparison < 0 else "" self.log.info( - " ➡️ Version " + f" ➡️ {prefix}Version " + (f"{latest_version}{version_suffix}" if latest_version else "new") + f" → {local_version}" ) @@ -289,17 +505,31 @@ def perform_version_check(self, spec: Any) -> ProcessingResult: outcome=ProcessingOutcome.PROCEED, local_version=local_version, latest_version=latest_version, + latest_digest=latest_digest, spec=spec, + published_versions=state.versions, ) - self.log.info(f" ⏭️ Skipping: Version {local_version} unchanged") + if comparison < 0: + self.log.info( + f" ⏭️ {prefix}Skipping: Local version {local_version} is older than published {latest_version}" + ) + reason = ( + f"Version {local_version} is older than published version {latest_version} " + "(no-op; publishing is monotonic)" + ) + else: + self.log.info(f" ⏭️ {prefix}Skipping: Version {local_version} unchanged") + reason = f"Version {local_version} unchanged (matches remote)" return ProcessingResult( outcome=ProcessingOutcome.SKIP, local_version=local_version, latest_version=latest_version, + latest_digest=latest_digest, spec=spec, - reason=f"Version {local_version} unchanged (matches remote)", + reason=reason, + published_versions=state.versions, ) def deprecate_old_components( @@ -528,24 +758,39 @@ def publish_component( outcome=ProcessingOutcome.SUCCESS, local_version=version_check_result.local_version, latest_version=version_check_result.latest_version, + latest_digest=version_check_result.latest_digest, spec=spec, reason=f"Dry-run: would publish {spec.name}", response={"name": spec.name, "text": local_yaml_content}, ) - filter_by = self.published_by or self.current_user_id(client) + filter_by = self.owner_filter(client) if not filter_by: - self.log.error( - "❌ Cannot determine current user — aborting to avoid deprecating components owned by others" - ) return ProcessingResult( outcome=ProcessingOutcome.ERROR, local_version=version_check_result.local_version, latest_version=version_check_result.latest_version, + latest_digest=version_check_result.latest_digest, spec=spec, reason="Cannot determine current user for author filtering", ) - existing_components = client.find_existing_components(spec.search_names, verbose=True, published_by=filter_by) + + # Re-observe the published state immediately before create/deprecate and + # re-apply the same policy: a row that appeared after the first check + # must be able to stop the publish, and must never be deprecated on the + # strength of the earlier decision. + final_state = self.collect_published_state(client, spec, filter_by, verbose=True) + final_check = self._evaluate_published_state(spec, final_state, stage="refreshed check") + if final_check.outcome != ProcessingOutcome.PROCEED: + if final_check.outcome == ProcessingOutcome.SKIP: + self.log.info(f" ⏭️ Skipping API publish: {final_check.reason}") + else: + self.log.error(f" ❌ Cannot proceed due to error: {final_check.reason}") + return final_check + + deprecation_candidates = [ + {"digest": digest} for digest in final_state.digests_older_than(str(spec.version)) + ] try: result = client.published_components_create(name=spec.name, text=local_yaml_content) @@ -554,22 +799,25 @@ def publish_component( if new_digest: self.log.info(f"✅ Published: {spec.name} (digest: {str(new_digest)[:16]}...)") - self.deprecate_old_components(existing_components, str(new_digest)) + self.deprecate_old_components(deprecation_candidates, str(new_digest)) return ProcessingResult( outcome=ProcessingOutcome.SUCCESS, - local_version=version_check_result.local_version, - latest_version=version_check_result.latest_version, + local_version=final_check.local_version, + latest_version=final_check.latest_version, spec=spec, reason=f"Successfully published with digest: {new_digest}", digest=str(new_digest), + latest_digest=final_check.latest_digest, response=result, + published_versions=final_state.versions, ) self.log.warn("⚠️ Component published but no digest returned") return ProcessingResult( outcome=ProcessingOutcome.ERROR, - local_version=version_check_result.local_version, - latest_version=version_check_result.latest_version, + local_version=final_check.local_version, + latest_version=final_check.latest_version, + latest_digest=final_check.latest_digest, spec=spec, reason="Component published but no digest returned", response=result, @@ -578,8 +826,9 @@ def publish_component( self.log.error(f"❌ Request failed: {exc}") return ProcessingResult( outcome=ProcessingOutcome.ERROR, - local_version=version_check_result.local_version, - latest_version=version_check_result.latest_version, + local_version=final_check.local_version, + latest_version=final_check.latest_version, + latest_digest=final_check.latest_digest, spec=spec, reason=f"Request failed: {exc}", ) @@ -772,14 +1021,16 @@ def perform_version_check( client: Any = None, logger: Logger | None = None, published_by: str | None = None, + allow_downgrade: bool = False, ) -> ProcessingResult: - """Perform owner-scoped version checking for a component.""" + """Perform owner-scoped, monotonic version checking for a component.""" return ComponentPublisher( dry_run=dry_run, client=client, logger=logger, published_by=published_by, + allow_downgrade=allow_downgrade, ).perform_version_check(spec) @@ -797,6 +1048,7 @@ def publish_component_to_tangle( client: Any = None, client_factory: Callable[[], Any] | None = None, published_by: str | None = None, + allow_downgrade: bool = False, ) -> ProcessingResult: """Publish one component using ``ComponentPublisher.publish_component``.""" @@ -809,6 +1061,7 @@ def publish_component_to_tangle( git_remote_url=git_remote_url, git_repo=git_repo, published_by=published_by, + allow_downgrade=allow_downgrade, ) return publisher.publish_component( file_path, @@ -830,6 +1083,7 @@ def publish_component(client: Any, component_path: str | Path, **kwargs: Any) -> git_root=kwargs.pop("git_root", None), git_repo=kwargs.pop("git_repo", None), published_by=kwargs.pop("published_by", None), + allow_downgrade=bool(kwargs.pop("allow_downgrade", False)), client=client, client_factory=kwargs.pop("client_factory", None), logger=kwargs.pop("logger", None), diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py b/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py index c850d4d..b7365d4 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py @@ -52,6 +52,7 @@ import yaml +from .authenticated_identity import ME from .component_from_func import build_unwrapped_inputs_schema from .handler import TangleCliHandler from .python_pipeline.cfg import Cfg, _coerce_override, load_cfg @@ -1814,6 +1815,57 @@ def _task_component_identity( } +def _task_publication(ref: CallableRef) -> dict[str, str] | None: + """The ``@Publish`` declaration on one traced ``@task`` call, if any. + + Returns: + The neutral ``publish`` marker mapping, or ``None`` when the task + declares no publication. + """ + if ref._task_publish_name is None: + return None + return { + "component_name": ref._task_publish_name, + "version": ref._task_publish_version or "", + } + + +def _require_agreeing_publication( + existing: dict[str, str] | None, + incoming: dict[str, str] | None, + *, + identity_payload: Mapping[str, Any], +) -> None: + """Refuse when refs sharing ONE component disagree about publication. + + Publication is intentionally not part of the dedup identity, so two call + sites that generate the same component collapse into a single sidecar + entry which can carry only one marker. Every ref folded into that entry + must therefore make the same declaration. + + By construction they do -- ``@Publish`` decorates the function, and every + derived ref copies the fields -- so this is a guard against a future + caller constructing refs directly, not the mechanism that makes the common + case work. + + Raises: + CompileError: If the two declarations differ. + """ + if existing == incoming: + return + + def _render(value: dict[str, str] | None) -> str: + if value is None: + return "no @Publish" + return f"@Publish({value['component_name']!r}, version={value['version']!r})" + + raise CompileError( + f"@Publish disagreement for {_task_identity_label(identity_payload)}: " + f"the same generated component is declared as {_render(existing)} and " + f"{_render(incoming)}. One component can carry one publication." + ) + + def _task_identity_label(identity_payload: Mapping[str, Any]) -> str: """Human-readable ``::`` label for an identity payload. @@ -1946,14 +1998,23 @@ def _plan_task_sidecar( identity = _stable_payload_hash(identity_payload) by_identity = variants.setdefault(base, {}) previous = by_identity.get(identity) + # Publication intent rides ALONGSIDE the identity, never inside it: + # it does not change the generated component, so folding it into the + # identity would split one component into two byte-identical ones. + # Every ref deduped into one component must therefore agree about it. + publication = _task_publication(ref) + if previous is not None: + _require_agreeing_publication( + previous[2], publication, identity_payload=identity_payload + ) if previous is None: - by_identity[identity] = (emitted, identity_payload) + by_identity[identity] = (emitted, identity_payload, publication) elif _emitted_form_rank(emitted) < _emitted_form_rank(previous[0]): # Same component, different SPELLING (e.g. one call site omits # ``mode`` while another writes ``mode="inline"``). Pick the # canonical representative rather than whichever was traced first, # so the sidecar text does not depend on call order. - by_identity[identity] = (emitted, previous[1]) + by_identity[identity] = (emitted, previous[1], previous[2]) legacy_fragments.setdefault((base, identity), _fragment_for_task(ref, unwrapped_schema)) task_identities.append((task_id, base, identity)) @@ -1962,7 +2023,7 @@ def _plan_task_sidecar( identity_labels: dict[str, str] = {} for base, by_identity in variants.items(): if len(by_identity) == 1: - identity, (emitted, _identity_payload) = next(iter(by_identity.items())) + identity, (emitted, _identity_payload, _publication) = next(iter(by_identity.items())) named = [(legacy_fragments[(base, identity)], identity, emitted)] else: # EVERY colliding variant is suffixed — the trace-order "first" one @@ -1970,7 +2031,7 @@ def _plan_task_sidecar( # sorted order so the sidecar text is call-order independent. named = sorted( (f"{base}--{identity}", identity, emitted) - for identity, (emitted, _identity_payload) in by_identity.items() + for identity, (emitted, _identity_payload, _publication) in by_identity.items() ) for fragment, identity, emitted in named: if fragment in entries: # defensive — digests are content-addressed @@ -1981,7 +2042,28 @@ def _plan_task_sidecar( ) fragment_by_identity[(base, identity)] = fragment identity_labels[fragment] = _task_identity_label(by_identity[identity][1]) - entries[fragment] = {"local_from_python": emitted} + entry: dict[str, Any] = {} + publication = by_identity[identity][2] + if publication is not None: + # The declaration is emitted as the resolver's OWN generic + # name/version/publisher fields, so ordinary hydration can look + # the published component up, with ``local_from_python`` as the + # candidate used when none is found. ``publish: true`` is a + # separate explicit marker: publication is never inferred from + # those fields alone, so a hand-written or fallback entry of + # that shape is not silently published. + # + # The publisher is the SYMBOLIC ``me``: compiling is offline and + # must stay so, so it cannot resolve an account id. Hydration + # resolves it to whoever is authenticated, which keeps the + # lookup owner-scoped instead of global. + entry["name"] = publication["component_name"] + entry["version"] = publication["version"] + entry["publisher"] = ME + entry["local_from_python"] = emitted + if publication is not None: + entry["publish"] = True + entries[fragment] = entry fragment_by_task = { task_id: fragment_by_identity[(base, identity)] for task_id, base, identity in task_identities diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_hydrator.py b/packages/tangle-cli/src/tangle_cli/pipeline_hydrator.py index 485aeeb..4646ed5 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_hydrator.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_hydrator.py @@ -23,6 +23,7 @@ import yaml from . import utils +from .authenticated_identity import is_symbolic_me, require_authenticated_user_id from .api_transport import DEFAULT_TIMEOUT_SECONDS from .component_generator import ComponentGenerator from .handler import TangleCliHandler @@ -1149,6 +1150,19 @@ def _resolve_by_name_with_filters( """Resolve a component by name with optional filters.""" component_name = entry["name"] publisher = entry.get("publisher") + if is_symbolic_me(publisher) and entry.get("publish") is True: + # Compilation is offline, so the compiler wrote a symbolic owner + # rather than an account id. Resolve it to whoever is actually + # authenticated: without an owner the lookup is global, and a + # component published by anyone else under this name and version + # would be a valid candidate. Unresolvable identity fails closed + # instead of widening the search. + # + # Gated on the marker as well as the symbol, because the compiler + # always writes both. An entry WITHOUT the marker is hand-authored, + # where ``me`` has always meant a literal account id; reinterpreting + # it would silently retarget or break such a config. + publisher = require_authenticated_user_id(self._api_client()) version_constraint = entry.get("version") required_annotations = entry.get("annotations") diff --git a/packages/tangle-cli/src/tangle_cli/published_components_cli.py b/packages/tangle-cli/src/tangle_cli/published_components_cli.py index 733eb02..0744278 100644 --- a/packages/tangle-cli/src/tangle_cli/published_components_cli.py +++ b/packages/tangle-cli/src/tangle_cli/published_components_cli.py @@ -247,6 +247,16 @@ def published_components_publish( Parameter(help="Custom annotations as a JSON object."), ] = None, dry_run: bool | None = None, + allow_downgrade: Annotated[ + bool | None, + Parameter( + help=( + "Publish even when the local version is older than the latest published " + "owner-scoped version. Off by default: publishing is monotonic and an older " + "local version is a no-op skip." + ) + ), + ] = None, git_remote_sha: str | None = None, git_remote_branch: str | None = None, git_remote_url: str | None = None, @@ -269,6 +279,7 @@ def published_components_publish( description=(description, None), annotations=("annotations", annotations, None, True), dry_run=(dry_run, None), + allow_downgrade=(allow_downgrade, None), git_remote_sha=(git_remote_sha, None), git_remote_branch=(git_remote_branch, None), git_remote_url=(git_remote_url, None), @@ -302,6 +313,7 @@ def published_components_publish( git_remote_url=args.git_remote_url, git_root=args.git_root, published_by=args.published_by, + allow_downgrade=bool(args.allow_downgrade), client=client, logger=logger, ) diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py index 18460b0..56bf560 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py @@ -24,6 +24,7 @@ from .pipeline import pipeline from .raw import raw from .ref import ref +from .publish import Publish from .registered import registered from .subpipeline import subpipeline from .task import task @@ -33,6 +34,7 @@ __all__ = [ "pipeline", "task", + "Publish", "registered", "ref", "raw", diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/publish.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/publish.py new file mode 100644 index 0000000..b1b3c8a --- /dev/null +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/publish.py @@ -0,0 +1,142 @@ +"""``@Publish`` decorator -- an inert metadata stub. + +This package deliberately contains no publication implementation. ``@Publish`` +records a component name and version for a local ``@task``, and the compiler +writes the neutral marker shown below into the generated +``.components.yaml`` resolver sidecar, beside that component's +``local_from_python`` block; neither action publishes, contacts a registry, or +expresses publication policy, and decorating changes nothing about how the +component is generated. Some current consumers are closed source, though their +implementations are expected to be opened in the future. The marker schema is +the integration contract and is kept deliberately neutral so any +implementation can consume it:: + + orders-loader: + name: orders-loader + version: "1.0" + publisher: me + local_from_python: {...} + publish: true + +Separate tooling reads ``publish`` and decides whether, when, and under what +policy to publish. The declaration is emitted as the entry's OWN generic +``name``/``version`` fields, so ordinary hydration looks the published +component up at the exact declared version and uses ``local_from_python`` when +no such component exists yet. A registry error is NOT a fallback: resolution +fails closed rather than quietly building from local source. + +``publish`` is deliberately a separate literal ``true`` and is never inferred +from ``name`` + ``version`` + ``local_from_python``, so a hand-written or +fallback entry of that shape is not silently published. + +``publisher`` is the SYMBOLIC value ``me``, matched exactly and +case-sensitively, and is resolved at hydration time to whoever is +authenticated; compiling stays offline and never learns or records an account +id. This keeps the lookup owner-scoped: an unscoped name search would let a +component published by someone else under the same name and version be +resolved in its place, and a name is not an identity control. If the account +cannot be determined, resolution fails closed rather than widening. + +The symbol is interpreted only on an entry that also carries ``publish: true``, +which the compiler always writes together. ``me`` is therefore NOT a reserved +account id: in a hand-authored entry, and in any entry without the marker, a +publisher is used verbatim -- including one that happens to be ``me``. An entry +with no publisher keeps the ordinary cross-publisher behaviour. + +Ordering: ``@Publish`` goes directly ABOVE ``@task``, so it receives the +``CallableRef`` that ``@task`` produced:: + + from tangle_cli.python_pipeline import Publish, task + + @Publish(component_name="orders-loader", version="1.0") + @task(image="python:3.11") + def load_orders(source: str = "orders") -> str: + '''Load orders.''' + ... + +The declaration is stored as REAL ``CallableRef`` fields, so it survives +fluent composition (``.bind()``, ``.named()``) and is carried by refs imported +from an already-loaded module. This decorator does not consult the registry, +read the filesystem, or perform any I/O. +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from .errors import CompileError +from .ref import CallableRef + +_USAGE = ( + "@Publish(component_name='my-component', version='1.0') directly above " + "the @task it publishes" +) + + +def _refuse(message: str) -> CompileError: + return CompileError(f"@Publish {message}") + + +def _require_text(value: Any, *, field: str) -> str: + """Validate one author-supplied string. + + Both values are echoed into diagnostics and written into a YAML document, + so control characters are rejected at the authoring boundary rather than + escaped at each point of use. + """ + if not isinstance(value, str) or not value.strip(): + raise _refuse(f"requires a non-empty string {field!r}. Write {_USAGE}.") + text = value.strip() + if any(ord(character) < 0x20 or ord(character) == 0x7F for character in text): + raise _refuse(f"{field!r} must not contain control characters.") + return text + + +@dataclass(frozen=True) +class Publish: + """Declare that the decorated ``@task`` component should be published. + + Args: + component_name: Registry name to publish under. Required, non-empty, + and never derived from the function name -- the registry name is a + deliberate, stable identifier that must not change because someone + renamed a Python function. A distinctive name is the author's + responsibility. + version: Version to publish. Required and non-empty; there is no + default and none is inferred. + + Returns: + The same :class:`CallableRef`, carrying the publication declaration. + + Raises: + CompileError: If either value is missing, blank, or contains control + characters, or if the decorated object is not a ``@task`` ref + (a plain function, a ``ref()``/``@registered`` ref, or ``@task`` + applied in the wrong order). + """ + + component_name: str + version: str + + def __post_init__(self) -> None: + object.__setattr__( + self, "component_name", _require_text(self.component_name, field="component_name") + ) + object.__setattr__(self, "version", _require_text(self.version, field="version")) + + def __call__(self, target: Any) -> CallableRef: + if not isinstance(target, CallableRef) or target._task_source_path is None: + raise _refuse( + f"can only publish a @task component, but it was applied to " + f"{type(target).__name__}. Write {_USAGE}." + ) + if target._task_publish_name is not None: + raise _refuse( + f"is already declared on this task as " + f"{target._task_publish_name!r} {target._task_publish_version}. " + f"One publication per task." + ) + return target._replace( + _task_publish_name=self.component_name, + _task_publish_version=self.version, + ) diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py index fa0891f..e86b55c 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py @@ -142,6 +142,16 @@ class CallableRef: _task_resolve_root: Path | None = None _task_custom_annotations: dict[str, str] | None = None _task_unwrap: tuple[str, ...] = () + # ``@Publish`` metadata. ``None`` unless the author declared a publication + # for this ``@task``. REAL dataclass fields rather than attributes stamped + # on the object: ``_replace`` re-copies only a dunder allowlist, so a + # dynamic attribute would be lost by ``.bind()``/``.named()``, whereas + # ``dataclasses.replace`` carries every field. That is what lets the marker + # survive fluent composition and imported/cached refs. Publication is NOT + # generation-affecting -- it never changes the generated component bytes -- + # so it is deliberately absent from the dedup identity. + _task_publish_name: str | None = None + _task_publish_version: str | None = None # ``@registered`` metadata. ``None`` for ``ref()``/``@task`` refs; # populated by the ``@registered`` decorator. Drives the compile-time # rewrite of the ``registered://pending`` sentinel URL to a diff --git a/pyproject.toml b/pyproject.toml index 30c1dd5..ddfb5bf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tangle-cli" -version = "0.1.14" +version = "0.1.15" description = "CLI for Tangle, the open-source ML pipeline orchestration platform" readme = "README.md" authors = [ diff --git a/tests/test_component_publisher.py b/tests/test_component_publisher.py index a6ac801..93466b4 100644 --- a/tests/test_component_publisher.py +++ b/tests/test_component_publisher.py @@ -30,6 +30,7 @@ class User: class ExistingComponent: digest: str name: str = "demo" + deprecated: bool = False class FakeClient: @@ -215,6 +216,445 @@ def test_version_check_skips_unchanged_owner_scoped_version() -> None: assert "unchanged" in (result.reason or "") +# --------------------------------------------------------------------------- +# Monotonic publishing contract +# --------------------------------------------------------------------------- + + +def _version_check( + local_version: str, + published: dict[str, str], + *, + deprecated: set[str] | None = None, + allow_downgrade: bool = False, +) -> tuple[ProcessingResult, FakeClient]: + spec = ComponentSpec.from_yaml( + f"name: demo\nmetadata:\n annotations:\n version: '{local_version}'\n" + ) + client = FakeClient() + client.existing = [ + ExistingComponent(digest, deprecated=digest in (deprecated or set())) for digest in published + ] + client.component_versions = dict(published) + result = perform_version_check( + spec=spec, + dry_run=False, + client=client, + allow_downgrade=allow_downgrade, + ) + return result, client + + +def test_version_check_proceeds_when_nothing_published() -> None: + result, client = _version_check("1.0", {}) + + assert result.outcome == ProcessingOutcome.PROCEED + assert result.latest_version is None + assert result.latest_digest is None + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_version_check_proceeds_when_local_is_newer() -> None: + result, _ = _version_check("1.1", {"sha256:v10": "1.0"}) + + assert result.outcome == ProcessingOutcome.PROCEED + assert result.latest_version == "1.0" + assert result.latest_digest == "sha256:v10" + + +def test_version_check_equal_version_skips_with_exact_digest() -> None: + result, client = _version_check("1.0", {"sha256:v10": "1.0"}) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_version == "1.0" + assert result.latest_digest == "sha256:v10" + assert result.resolved_digest == "sha256:v10" + assert "unchanged" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_version_check_older_local_version_skips_with_newer_digest() -> None: + result, client = _version_check("1.0", {"sha256:v20": "2.0"}) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_version == "2.0" + assert result.latest_digest == "sha256:v20" + assert result.resolved_digest == "sha256:v20" + assert "older" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_version_check_selects_highest_of_multiple_published_versions() -> None: + result, _ = _version_check( + "1.5", + {"sha256:v10": "1.0", "sha256:v201": "2.0.1", "sha256:v20": "2.0"}, + ) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_version == "2.0.1" + assert result.latest_digest == "sha256:v201" + + +def test_version_check_fails_closed_on_ambiguous_latest_digests() -> None: + result, client = _version_check("1.0", {"sha256:a": "2.0", "sha256:b": "2.0"}) + + assert result.outcome == ProcessingOutcome.ERROR + assert result.latest_digest is None + assert result.resolved_digest is None + assert "Ambiguous latest published version 2.0" in (result.reason or "") + assert "sha256:a" in (result.reason or "") and "sha256:b" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_version_check_ignores_deprecated_components_when_selecting_latest() -> None: + result, _ = _version_check( + "1.0", + {"sha256:v10": "1.0", "sha256:v20": "2.0"}, + deprecated={"sha256:v20"}, + ) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_version == "1.0" + assert result.latest_digest == "sha256:v10" + + +def test_version_check_ignores_deprecated_duplicate_of_latest_version() -> None: + result, _ = _version_check( + "1.0", + {"sha256:live": "2.0", "sha256:dead": "2.0"}, + deprecated={"sha256:dead"}, + ) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_digest == "sha256:live" + + +def test_version_check_allow_downgrade_opts_back_into_publishing_older() -> None: + result, _ = _version_check("1.0", {"sha256:v20": "2.0"}, allow_downgrade=True) + + assert result.outcome == ProcessingOutcome.PROCEED + assert result.latest_version == "2.0" + assert result.latest_digest == "sha256:v20" + + +def test_publish_older_version_is_a_noop_that_pins_newer_digest(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="1.0") + client = FakeClient() + client.existing = [ExistingComponent("sha256:v20")] + client.component_versions = {"sha256:v20": "2.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.digest is None + assert result.latest_digest == "sha256:v20" + assert result.resolved_digest == "sha256:v20" + assert client.create_calls == [] + assert client.update_calls == [] + assert result.to_dict()["latest_digest"] == "sha256:v20" + + +def test_publish_newer_version_reports_new_and_previous_digests(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = FakeClient() + client.existing = [ExistingComponent("sha256:v10")] + client.component_versions = {"sha256:v10": "1.0"} + client.publish_response = {"digest": "sha256:v20"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.SUCCESS + assert result.digest == "sha256:v20" + assert result.latest_digest == "sha256:v10" + assert result.resolved_digest == "sha256:v20" + + +def test_publish_ambiguous_latest_digests_publishes_nothing(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="1.0") + client = FakeClient() + client.existing = [ExistingComponent("sha256:a"), ExistingComponent("sha256:b")] + client.component_versions = {"sha256:a": "2.0", "sha256:b": "2.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert result.resolved_digest is None + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_resolved_digest_is_none_for_error_outcomes(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = FakeClient() + client.existing = [ExistingComponent("sha256:v10")] + client.component_versions = {"sha256:v10": "1.0"} + + def failing_create(**kwargs: Any) -> dict[str, Any]: + raise RuntimeError("boom") + + client.published_components_create = failing_create # type: ignore[method-assign] + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert result.latest_digest == "sha256:v10" + assert result.resolved_digest is None + assert client.update_calls == [] + + +# --------------------------------------------------------------------------- +# Fail-closed reads, lookup races, deterministic diagnostics +# --------------------------------------------------------------------------- + + +class RaceClient(FakeClient): + """Client whose published state changes between the two owner-scoped lookups.""" + + def __init__(self, first: list[ExistingComponent], second: list[ExistingComponent]) -> None: + super().__init__() + self._sequence = [first, second] + self.existing = first + + def find_existing_components(self, components: Any, **kwargs: Any) -> list[ExistingComponent]: + self.existing = self._sequence[min(len(self.find_calls), len(self._sequence) - 1)] + return super().find_existing_components(components, **kwargs) + + +def test_version_check_fails_closed_on_unreadable_candidate(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = FakeClient() + client.existing = [ExistingComponent("sha256:v10"), ExistingComponent("sha256:unknown")] + # "sha256:unknown" is absent from component_versions, so get_component_spec raises. + client.component_versions = {"sha256:v10": "1.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert "Cannot read published version" in (result.reason or "") + assert "sha256:unknown" in (result.reason or "") + assert result.resolved_digest is None + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_version_check_fails_closed_on_candidate_without_digest() -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '2.0'\n") + client = FakeClient() + client.existing = [ExistingComponent("", name="demo")] + + result = perform_version_check(spec=spec, dry_run=False, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert "no digest" in (result.reason or "") + assert client.create_calls == [] + + +def test_unreadable_row_appearing_before_publish_blocks_create_and_deprecate(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = RaceClient( + [ExistingComponent("sha256:v10")], + [ExistingComponent("sha256:v10"), ExistingComponent("sha256:unknown")], + ) + client.component_versions = {"sha256:v10": "1.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert "sha256:unknown" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_concurrent_newer_row_between_lookups_skips_without_publishing(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = RaceClient( + [ExistingComponent("sha256:v10")], + [ExistingComponent("sha256:v10"), ExistingComponent("sha256:v30")], + ) + client.component_versions = {"sha256:v10": "1.0", "sha256:v30": "3.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_version == "3.0" + assert result.latest_digest == "sha256:v30" + assert result.resolved_digest == "sha256:v30" + assert "older" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_concurrent_equal_row_between_lookups_skips_without_publishing(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = RaceClient([], [ExistingComponent("sha256:v20")]) + client.component_versions = {"sha256:v20": "2.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_digest == "sha256:v20" + assert "unchanged" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_concurrent_ambiguous_rows_between_lookups_fail_closed(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = RaceClient( + [ExistingComponent("sha256:v10")], + [ExistingComponent("sha256:b"), ExistingComponent("sha256:a")], + ) + client.component_versions = {"sha256:v10": "1.0", "sha256:a": "3.0", "sha256:b": "3.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert "Ambiguous latest published version 3.0" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_row_appearing_after_version_check_is_never_deprecated(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = RaceClient( + [ExistingComponent("sha256:v10")], + [ExistingComponent("sha256:v10"), ExistingComponent("sha256:v11")], + ) + client.component_versions = {"sha256:v10": "1.0", "sha256:v11": "1.1"} + client.publish_response = {"digest": "sha256:v20"} + + result = publish_component_to_tangle(component_path, client=client) + + # Both late rows are proven older than 2.0, so both may be deprecated, in + # sorted order; nothing unverified is ever touched. + assert result.outcome == ProcessingOutcome.SUCCESS + assert [call["digest"] for call in client.update_calls] == ["sha256:v10", "sha256:v11"] + + +def test_allow_downgrade_never_deprecates_newer_rows(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="1.0") + client = FakeClient() + client.existing = [ExistingComponent("sha256:v20"), ExistingComponent("sha256:v005")] + client.component_versions = {"sha256:v20": "2.0", "sha256:v005": "0.5"} + client.publish_response = {"digest": "sha256:v10"} + + result = publish_component_to_tangle(component_path, client=client, allow_downgrade=True) + + assert result.outcome == ProcessingOutcome.SUCCESS + assert [call["digest"] for call in client.update_calls] == ["sha256:v005"] + + +def test_ambiguity_diagnostics_are_sorted_independently_of_api_order() -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '1.0'\n") + reasons = [] + for digests in (["sha256:a", "sha256:b"], ["sha256:b", "sha256:a"]): + client = FakeClient() + client.existing = [ExistingComponent(digest) for digest in digests] + client.component_versions = {"sha256:a": "2.0", "sha256:b": "2.0"} + result = perform_version_check(spec=spec, dry_run=False, client=client) + assert result.outcome == ProcessingOutcome.ERROR + reasons.append(result.reason) + + assert reasons[0] == reasons[1] + assert "(sha256:a, sha256:b)" in (reasons[0] or "") + + +def test_latest_digest_selection_is_independent_of_api_order() -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '1.0'\n") + selected = [] + for digests in (["sha256:v10", "sha256:v30", "sha256:v20"], ["sha256:v30", "sha256:v20", "sha256:v10"]): + client = FakeClient() + client.existing = [ExistingComponent(digest) for digest in digests] + client.component_versions = {"sha256:v10": "1.0", "sha256:v20": "2.0", "sha256:v30": "3.0"} + result = perform_version_check(spec=spec, dry_run=False, client=client) + selected.append((result.latest_version, result.latest_digest)) + + assert selected[0] == selected[1] == ("3.0", "sha256:v30") + + +def test_compare_equal_raw_versions_pick_deterministic_representative() -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '0.9'\n") + picked = [] + for digests in (["sha256:a", "sha256:b"], ["sha256:b", "sha256:a"]): + client = FakeClient() + client.existing = [ExistingComponent(digest) for digest in digests] + # "1.0" and "1.0.0" compare equal, so this is an ambiguous tie either way. + client.component_versions = {"sha256:a": "1.0", "sha256:b": "1.0.0"} + result = perform_version_check(spec=spec, dry_run=False, client=client) + assert result.outcome == ProcessingOutcome.ERROR + picked.append((result.latest_version, result.reason)) + + assert picked[0] == picked[1] + + +def test_dry_run_test_latest_version_applies_monotonic_rules(monkeypatch) -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '2.0'\n") + + monkeypatch.setenv("TEST_LATEST_VERSION", "1.0") + newer = perform_version_check(spec=spec, dry_run=True) + assert newer.outcome == ProcessingOutcome.PROCEED + assert newer.latest_version == "1.0" + assert newer.latest_digest is None + + monkeypatch.setenv("TEST_LATEST_VERSION", "2.0") + equal = perform_version_check(spec=spec, dry_run=True) + assert equal.outcome == ProcessingOutcome.SKIP + assert equal.latest_version == "2.0" + assert equal.latest_digest is None + assert equal.resolved_digest is None + assert "unchanged" in (equal.reason or "") + + monkeypatch.setenv("TEST_LATEST_VERSION", "3.0") + older = perform_version_check(spec=spec, dry_run=True) + assert older.outcome == ProcessingOutcome.SKIP + assert older.latest_version == "3.0" + assert older.latest_digest is None + assert older.resolved_digest is None + assert "older" in (older.reason or "") + + +def test_dry_run_without_test_latest_version_proceeds(monkeypatch) -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '2.0'\n") + monkeypatch.delenv("TEST_LATEST_VERSION", raising=False) + + result = perform_version_check(spec=spec, dry_run=True) + + assert result.outcome == ProcessingOutcome.PROCEED + assert result.latest_version is None + assert result.latest_digest is None + + +def test_dry_run_publish_skips_when_test_latest_version_is_newer(monkeypatch, tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="1.0") + client = FakeClient() + monkeypatch.setenv("TEST_LATEST_VERSION", "2.0") + + result = publish_component_to_tangle(component_path, dry_run=True, client=client) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_version == "2.0" + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_version_check_without_owner_reports_error_and_no_digest() -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '1.0'\n") + client = FakeClient() + client.user = None + + result = perform_version_check(spec=spec, dry_run=False, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert result.latest_digest is None + assert result.resolved_digest is None + assert client.find_calls == [] + assert client.create_calls == [] + + def test_version_check_progress_uses_logger_not_tangle_verbose(monkeypatch, capsys) -> None: spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '1.0'\n") client = FakeClient() @@ -438,3 +878,53 @@ def test_publish_components_returns_nonzero_for_errors(tmp_path: Path) -> None: ProcessingOutcome.SUCCESS, ProcessingOutcome.ERROR, ] + + +def test_the_owner_argument_a_version_check_passes_is_matched_exactly() -> None: + """Pin the semantics of the owner argument ``perform_version_check`` sends. + + This is a CONTRACT test, not an end-to-end publisher run: it calls + ``find_existing_components`` directly, which is the call the version check + makes. The companion test above proves the publisher passes ``published_by``; + this proves what that argument then means. Together they stop the + publisher's owner scope loosening silently. + + It matters because the server query is a substring match: without the + client's exact filter, a component owned by ``alice2`` would be read as part + of ``alice``'s published state -- constraining alice's next version, or + being considered for deprecation. + """ + from tangle_cli.client import TangleApiClient + from tangle_cli.models import ComponentInfo + + rows = [ + ComponentInfo(name="orders-loader", digest="sha256:theirs", version="9.9", published_by="alice2"), + ComponentInfo(name="orders-loader", digest="sha256:mine", version="1.0", published_by="alice"), + ] + + class _Session: + def request(self, *a: Any, **kw: Any) -> Any: # pragma: no cover - never called + raise AssertionError("no HTTP in this test") + + class _Client(TangleApiClient): + def list_published_component_infos( # type: ignore[override] + self, + include_deprecated: bool = False, + name_substring: str | None = None, + published_by_substring: str | None = None, + digest: str | None = None, + *, + fetch_specs: bool = False, + ) -> list[Any]: + out = rows + if published_by_substring: + out = [i for i in out if published_by_substring in (i.published_by or "")] + if name_substring: + out = [i for i in out if name_substring.lower() in i.name.lower()] + return out + + client = _Client("https://api.test", session=_Session()) + + found = client.find_existing_components(["orders-loader"], published_by="alice") + + assert [i.digest for i in found] == ["sha256:mine"] diff --git a/tests/test_components_cli.py b/tests/test_components_cli.py index c52b5d8..d2b23c0 100644 --- a/tests/test_components_cli.py +++ b/tests/test_components_cli.py @@ -88,6 +88,7 @@ def fake_client_from_options(**kwargs: Any) -> object: "git_remote_url": None, "git_root": None, "published_by": None, + "allow_downgrade": False, "client": None, "logger": ANY, } @@ -283,6 +284,7 @@ def fake_client_from_options(**kwargs: Any) -> object: "git_remote_url": None, "git_root": None, "published_by": "first@example.com", + "allow_downgrade": False, "client": fake_client, "logger": ANY, }, @@ -293,6 +295,7 @@ def fake_client_from_options(**kwargs: Any) -> object: "git_remote_url": None, "git_root": None, "published_by": "second@example.com", + "allow_downgrade": False, "client": None, "logger": ANY, }, diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 21f20bd..68d5980 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.14" in metadata + assert "Version: 0.1.15" 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_python_pipeline_dsl.py b/tests/test_python_pipeline_dsl.py index 3a876e9..f4211db 100644 --- a/tests/test_python_pipeline_dsl.py +++ b/tests/test_python_pipeline_dsl.py @@ -53,6 +53,7 @@ def test_all_names_are_exported(self): assert set(pp.__all__) == { "pipeline", "task", + "Publish", "registered", "ref", "raw", diff --git a/tests/test_python_pipeline_publish.py b/tests/test_python_pipeline_publish.py new file mode 100644 index 0000000..b4ccbd3 --- /dev/null +++ b/tests/test_python_pipeline_publish.py @@ -0,0 +1,565 @@ +"""``@Publish`` marker: decoration contract, ref propagation, sidecar emission. + +``@Publish`` is a pure marker. Compiling a marked pipeline records the +declaration in the resolver sidecar beside ``local_from_python`` and publishes +NOTHING; a downstream tool reads the marker and owns the policy. These tests +cover the authoring refusals, the propagation properties that make the marker +trustworthy (fluent composition, imported/cached refs, repeated compiles), and +the emission rules (dedup agreement, subgraph children, hydration unaffected). +""" +from __future__ import annotations + +import importlib +import sys +from pathlib import Path +from typing import Any + +import pytest +import yaml + +from tangle_cli.pipeline_compiler import compile_pipeline +from tangle_cli.python_pipeline import Publish, ref, registered, task +from tangle_cli.python_pipeline.errors import CompileError + +FIXTURES = Path(__file__).parent / "fixtures" / "python_pipeline" + + +def _assert_marked(entry: dict, name: str, version: str) -> None: + """Assert the emitted shape for a declared component. + + The declaration is emitted as the resolver's OWN + ``name``/``version``/``publisher`` fields, so ordinary hydration looks the + published component up; ``local_from_python`` remains as the candidate used + when none is found. ``publish`` is a separate literal ``True`` so + publication is never inferred from those fields alone. + + ``publisher`` is the SYMBOLIC ``me``, never a resolved account id: the + compiler is offline and must not contact the API to learn who is running it. + """ + assert entry["name"] == name + assert entry["version"] == version + assert entry["publisher"] == "me" + assert entry["publish"] is True + assert "local_from_python" in entry + + +def _task_ref(**kwargs): + """A minimal ``@task`` ref to apply ``@Publish`` to.""" + + def load_orders(source: str = "orders") -> str: + """Load orders. + + Metadata: + Name: Load Orders + """ + return source + + return task(image="registry.example/loader:1", **kwargs)(load_orders) + + +# --------------------------------------------------------------------------- +# Authoring contract: both values required, and the target must be a @task. + + +@pytest.mark.parametrize( + "kwargs", + [ + pytest.param({"component_name": "", "version": "1.0"}, id="empty-name"), + pytest.param({"component_name": " ", "version": "1.0"}, id="blank-name"), + pytest.param({"component_name": "orders", "version": ""}, id="empty-version"), + pytest.param({"component_name": None, "version": "1.0"}, id="none-name"), + pytest.param({"component_name": "orders", "version": 1.0}, id="non-string-version"), + pytest.param({"component_name": "ord\ners", "version": "1.0"}, id="newline-in-name"), + pytest.param({"component_name": "orders", "version": "1.0\x7f"}, id="delete-in-version"), + ], +) +def test_publish_refuses_unusable_values(kwargs): + """Both values are required, non-blank, and control-character free: they + are echoed into diagnostics and written into a YAML document.""" + with pytest.raises(CompileError, match="@Publish"): + Publish(**kwargs) + + +def test_publish_requires_both_values(): + with pytest.raises(TypeError): + Publish(component_name="orders") # type: ignore[call-arg] + + +def test_publish_strips_surrounding_whitespace(): + declared = Publish(component_name=" orders-loader ", version=" 1.0 ") + assert (declared.component_name, declared.version) == ("orders-loader", "1.0") + + +def test_publish_refuses_a_plain_function(): + """Applied BELOW @task (wrong order) the decorator sees a function.""" + with pytest.raises(CompileError, match="can only publish a @task component"): + + @Publish(component_name="orders-loader", version="1.0") + def load_orders() -> str: + return "orders" + + +def test_publish_refuses_a_plain_ref(): + """``ref()`` names an existing component; there is no local source to + generate and publish from.""" + with pytest.raises(CompileError, match="can only publish a @task component"): + Publish(component_name="orders-loader", version="1.0")( + ref("file://component.yaml") + ) + + +def test_publish_refuses_a_registered_ref(): + """``@registered`` points at an ALREADY published component.""" + + @registered(fragment="run-query", gen_config="gs://bucket/gen_config.yaml") + def run_query(sql: str = "SELECT 1") -> str: + """Run a query.""" + return sql + + with pytest.raises(CompileError, match="can only publish a @task component"): + Publish(component_name="run-query", version="1.0")(run_query) + + +def test_publish_refuses_a_second_declaration(): + published = Publish(component_name="orders-loader", version="1.0")(_task_ref()) + with pytest.raises(CompileError, match="already declared"): + Publish(component_name="other", version="2.0")(published) + + +# --------------------------------------------------------------------------- +# Propagation: the marker must survive everything that produces a NEW ref. + + +def test_the_marker_is_a_real_field_carried_by_fluent_composition(): + """``_replace`` re-copies only a dunder allowlist, so a dynamically + stamped attribute would be dropped here. Real dataclass fields are copied + by ``dataclasses.replace`` itself.""" + published = Publish(component_name="orders-loader", version="1.0")(_task_ref()) + + derived = published.named("first").bind(source="a") + + assert derived._task_publish_name == "orders-loader" + assert derived._task_publish_version == "1.0" + + +def test_the_marker_does_not_leak_onto_an_undeclared_task(): + plain = _task_ref() + Publish(component_name="orders-loader", version="1.0")(plain) + + assert plain._task_publish_name is None + assert plain._task_publish_version is None + + +# --------------------------------------------------------------------------- +# Sidecar emission. + + +def _keys_named(data: Any, key: str) -> bool: + """Whether ``key`` appears as a mapping key anywhere in ``data``.""" + if isinstance(data, dict): + return key in data or any(_keys_named(value, key) for value in data.values()) + if isinstance(data, list): + return any(_keys_named(item, key) for item in data) + return False + + +def _sidecar(pipeline_path: Path, out: Path): + result = compile_pipeline(pipeline_path, out) + return yaml.safe_load(result.components_path.read_text()) + + +def _project_with(tmp_path: Path, body: str) -> Path: + """Write a one-module pipeline project and return its pipeline path.""" + src = tmp_path / "project" / "src" + src.mkdir(parents=True) + pipeline_path = src / "pipeline.py" + pipeline_path.write_text(body, encoding="utf-8") + return pipeline_path + + +_PUBLISHING_PIPELINE = ( + "from tangle_cli.python_pipeline import Out, Publish, pipeline, task\n\n" + "@Publish(component_name='orders-loader', version='1.0')\n" + "@task(image='registry.example/loader:1')\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n' + " Metadata:\n" + " Name: Load Orders\n" + ' """\n' + " return source\n\n" + "@pipeline('Publishing Pipeline')\n" + "def publishing_pipeline() -> Out[str]:\n" + " loaded = load_orders(source='orders')\n" + " return loaded\n" +) + + +def test_a_declared_task_emits_a_publish_marker_beside_local_from_python(tmp_path): + pipeline_path = _project_with(tmp_path, _PUBLISHING_PIPELINE) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + entry = sidecar["load-orders"] + _assert_marked(entry, "orders-loader", "1.0") + # The local resolver is untouched and stays the fallback candidate. + assert entry["local_from_python"]["function"] == "load_orders" + + +def test_an_undeclared_task_emits_no_publish_key(tmp_path): + """Absence is expressed by omission, never by a null or empty mapping.""" + pipeline_path = _project_with( + tmp_path, _PUBLISHING_PIPELINE.replace( + "@Publish(component_name='orders-loader', version='1.0')\n", "" + ) + ) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + assert "publish" not in sidecar["load-orders"] + + +def test_compiling_a_declared_pipeline_publishes_nothing(tmp_path, monkeypatch): + """Compile records intent and performs no registry call. The marker is + inert: publication is a separate, downstream decision.""" + import tangle_cli.component_publisher as component_publisher + + def _fail(*args, **kwargs): # pragma: no cover - must never run + raise AssertionError("compile must not publish") + + monkeypatch.setattr( + component_publisher.ComponentPublisher, "publish_components", _fail, raising=False + ) + pipeline_path = _project_with(tmp_path, _PUBLISHING_PIPELINE) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + assert sidecar["load-orders"]["name"] == "orders-loader" + + +def test_two_call_sites_of_one_declared_task_share_one_entry_and_one_marker(tmp_path): + """Dedup is unchanged by publication: one generated component, one entry, + one marker.""" + pipeline_path = _project_with( + tmp_path, + "from tangle_cli.python_pipeline import Out, Publish, pipeline, task\n\n" + "@Publish(component_name='orders-loader', version='1.0')\n" + "@task(image='registry.example/loader:1')\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n' + " Metadata:\n" + " Name: Load Orders\n" + ' """\n' + " return source\n\n" + "@pipeline('Twice Pipeline')\n" + "def twice_pipeline() -> Out[str]:\n" + " load_orders.named('first')(source='a')\n" + " return load_orders.named('second')(source='b')\n", + ) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + assert list(sidecar) == ["load-orders"] + _assert_marked(sidecar["load-orders"], "orders-loader", "1.0") + + +def test_two_declarations_folded_into_one_component_are_refused(tmp_path): + """Publication is deliberately NOT part of the dedup identity, so two + refs that generate the same component collapse into one entry that can + carry only one marker. Reachable by applying @Publish functionally to one + of two otherwise identical refs — so this refusal is not merely defensive. + """ + pipeline_path = _project_with( + tmp_path, + "from tangle_cli.python_pipeline import Out, Publish, pipeline, task\n\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n' + " Metadata:\n" + " Name: Load Orders\n" + ' """\n' + " return source\n\n" + "base = task(image='registry.example/loader:1')(load_orders)\n" + "declared = Publish(component_name='orders-loader', version='1.0')(base)\n\n" + "@pipeline('Disagreeing Pipeline')\n" + "def disagreeing_pipeline() -> Out[str]:\n" + " base.named('plain')(source='a')\n" + " return declared.named('declared')(source='b')\n", + ) + + with pytest.raises(CompileError, match="@Publish disagreement"): + compile_pipeline(pipeline_path, pipeline_path.parent / "compiled.yaml") + + +@pytest.mark.parametrize( + ("first", "second"), + [ + pytest.param("", ", mode='inline'", id="lean-then-explicit"), + pytest.param(", mode='inline'", "", id="explicit-then-lean"), + ], +) +def test_the_marker_survives_the_canonical_spelling_choice(tmp_path, first, second): + """``@task()`` and ``@task(mode='inline')`` are one component spelled two + ways, and the compiler keeps the leanest spelling regardless of call + order. Whichever emitted form wins must still carry the declaration -- + in the order where the SECOND call site supplies the canonical form, the + representative is replaced after the marker was first recorded. + """ + src = tmp_path / "project" / "src" + src.mkdir(parents=True) + pipeline_path = src / "pipeline.py" + pipeline_path.write_text( + "from tangle_cli.python_pipeline import Out, Publish, pipeline, task\n\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n Metadata:\n Name: Load Orders\n """\n' + " return source\n\n" + "publish = Publish(component_name='orders-loader', version='1.0')\n" + f"first = publish(task(image='registry.example/loader:1'{first})(load_orders))\n" + f"second = publish(task(image='registry.example/loader:1'{second})(load_orders))\n\n" + "@pipeline('Spelling Pipeline')\n" + "def spelling_pipeline() -> Out[str]:\n" + " a = first.named('a')(source='a')\n" + " b = second.named('b')(source='b')\n" + " return b\n", + encoding="utf-8", + ) + + sidecar = _sidecar(pipeline_path, src / "compiled.yaml") + + assert list(sidecar) == ["load-orders"] + _assert_marked(sidecar["load-orders"], "orders-loader", "1.0") + + +def test_two_different_components_each_carry_their_own_marker(tmp_path): + """Multiple distinct publications per compile are representable upstream: + the one-per-ship rule is a downstream policy, not a compiler rule.""" + pipeline_path = _project_with( + tmp_path, + "from tangle_cli.python_pipeline import Out, Publish, pipeline, task\n\n" + "@Publish(component_name='orders-loader', version='1.0')\n" + "@task(image='registry.example/loader:1')\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n Metadata:\n Name: Load Orders\n """\n' + " return source\n\n" + "@Publish(component_name='orders-shipper', version='2.0')\n" + "@task(image='registry.example/shipper:1')\n" + "def ship_orders(source: str = 'orders') -> str:\n" + ' """Ship orders.\n\n Metadata:\n Name: Ship Orders\n """\n' + " return source\n\n" + "@pipeline('Two Publications Pipeline')\n" + "def two_publications_pipeline() -> Out[str]:\n" + " loaded = load_orders(source='orders')\n" + " shipped = ship_orders(source=loaded)\n" + " return shipped\n", + ) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + assert sidecar["load-orders"]["name"] == "orders-loader" + assert sidecar["ship-orders"]["name"] == "orders-shipper" + assert sidecar["ship-orders"]["version"] == "2.0" + + +# --------------------------------------------------------------------------- +# The case that defeated a capture-based design: the declaration lives in a +# module that is ALREADY imported, so its decorators do not run again. + + +def _shared_task_project(tmp_path: Path, module: str) -> Path: + """A project whose pipeline imports its declared @task from ``module``.""" + src = tmp_path / "project" / "src" + src.mkdir(parents=True) + (src / f"{module}.py").write_text( + "from tangle_cli.python_pipeline import Publish, task\n\n" + "@Publish(component_name='orders-loader', version='1.0')\n" + "@task(image='registry.example/loader:1')\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n' + " Metadata:\n" + " Name: Load Orders\n" + ' """\n' + " return source\n", + encoding="utf-8", + ) + pipeline_path = src / "pipeline.py" + pipeline_path.write_text( + "from tangle_cli.python_pipeline import Out, pipeline\n" + f"from {module} import load_orders\n\n" + "@pipeline('Imported Pipeline')\n" + "def imported_pipeline() -> Out[str]:\n" + " loaded = load_orders(source='orders')\n" + " return loaded\n", + encoding="utf-8", + ) + return pipeline_path + + +def test_a_declaration_in_an_imported_module_is_emitted(tmp_path): + """The marker travels on the ref, so it does not matter which module the + declaration was written in.""" + module = f"pubshared_{abs(hash(str(tmp_path))):x}" + pipeline_path = _shared_task_project(tmp_path, module) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + assert sidecar["load-orders"]["name"] == "orders-loader" + + +def test_a_preloaded_declaring_module_changes_nothing(tmp_path): + """The decisive property, and the exact case that defeats a capture-based + design: when the declaring module is ALREADY imported its ``@Publish`` + does not execute again, so a mechanism that observed decoration would see + nothing. The marker rides on the ref, so a cold compile and a compile with + the module preloaded produce the same sidecar. + """ + module = f"pubcached_{abs(hash(str(tmp_path))):x}" + pipeline_path = _shared_task_project(tmp_path, module) + source_dir = pipeline_path.parent + + cold = _sidecar(pipeline_path, source_dir / "cold.yaml") + + sys.path.insert(0, str(source_dir)) + try: + preloaded = importlib.import_module(module) + # Verify the preload resolved to the module just written, rather than + # some same-named module left behind by another test. + assert Path(preloaded.__file__ or "") == source_dir / f"{module}.py" + assert preloaded.load_orders._task_publish_name == "orders-loader" + + warm = _sidecar(pipeline_path, source_dir / "warm.yaml") + finally: + sys.path.remove(str(source_dir)) + sys.modules.pop(module, None) + + _assert_marked(cold["load-orders"], "orders-loader", "1.0") + assert warm == cold + + +# --------------------------------------------------------------------------- +# Subgraph children and hydration. + + +def test_a_declaration_inside_a_subpipeline_child_is_emitted(tmp_path): + """A child graph gets its own ``.components.yaml``; the marker must + ride into the child's sidecar, not the root's.""" + src = tmp_path / "project" / "src" + src.mkdir(parents=True) + pipeline_path = src / "pipeline.py" + pipeline_path.write_text( + "from tangle_cli.python_pipeline import Out, Publish, pipeline, subpipeline, task\n\n" + "@Publish(component_name='orders-loader', version='1.0')\n" + "@task(image='registry.example/loader:1')\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n Metadata:\n Name: Load Orders\n """\n' + " return source\n\n" + "@pipeline('Child Pipeline')\n" + "def child_pipeline() -> Out[str]:\n" + " loaded = load_orders(source='orders')\n" + " return loaded\n\n" + "@pipeline('Parent Pipeline')\n" + "def parent_pipeline() -> Out[str]:\n" + " child = subpipeline(child_pipeline)()\n" + " return child\n", + encoding="utf-8", + ) + out = src / "compiled.yaml" + + result = compile_pipeline(pipeline_path, out, pipeline_name="Parent Pipeline") + + assert len(result.subgraph_paths) == 1 + child_graph = result.subgraph_paths[0] + child_sidecar_path = child_graph.with_name(f"{child_graph.stem}.components.yaml") + child_sidecar = yaml.safe_load(child_sidecar_path.read_text()) + _assert_marked(child_sidecar["load-orders"], "orders-loader", "1.0") + + +def test_the_marker_is_stripped_from_the_baked_operation_program(): + """``@Publish`` is authoring-only. The baked program drops the authoring + import, so a surviving ``@Publish`` line would raise ``NameError`` at + container startup for every marked task. + """ + from tangle_cli.component_from_func import _strip_authoring_constructs + + baked = _strip_authoring_constructs( + "from tangle_cli.python_pipeline import Publish, task\n\n" + '@Publish(component_name="orders-loader", version="1.0")\n' + '@task(image="registry.example/loader:1")\n' + 'def load_orders(source: str = "orders") -> str:\n' + " return source\n" + ) + + assert "Publish" not in baked + assert "tangle_cli.python_pipeline" not in baked + # The decisive check: it must RUN with no authoring names in scope. + namespace: dict[str, Any] = {} + exec(compile(baked, "", "exec"), namespace) + assert namespace["load_orders"]("orders") == "orders" + + +def test_the_generated_component_command_carries_no_authoring_construct(tmp_path): + """End-to-end counterpart: the command actually baked into the resolved + component must be free of the decorator and its import.""" + from unittest.mock import MagicMock + + from tangle_cli.pipeline_hydrator import PipelineHydrator + + pipeline_path = _project_with(tmp_path, _PUBLISHING_PIPELINE) + out = pipeline_path.parent / "compiled.yaml" + compile_pipeline(pipeline_path, out) + + hydrated = PipelineHydrator(client=MagicMock()).hydrate_file(out) + + tasks = hydrated.data["implementation"]["graph"]["tasks"] + spec = next(iter(tasks.values()))["componentRef"]["spec"] + program = "\n".join( + part for part in spec["implementation"]["container"]["command"] if isinstance(part, str) + ) + assert "@Publish" not in program + assert "from tangle_cli.python_pipeline import" not in program + assert "def load_orders" in program + + +def test_a_marked_component_still_hydrates_from_local_source(tmp_path): + """Hydration reads ``local_from_python`` and ignores the marker, so a + declared component resolves exactly as an undeclared one does.""" + from unittest.mock import MagicMock + + from tangle_cli.pipeline_hydrator import PipelineHydrator + + declared_path = _project_with(tmp_path, _PUBLISHING_PIPELINE) + declared_out = declared_path.parent / "compiled.yaml" + compile_pipeline(declared_path, declared_out) + + hydrated = PipelineHydrator(client=MagicMock()).hydrate_file(declared_out) + + tasks = hydrated.data["implementation"]["graph"]["tasks"] + task_spec = next(iter(tasks.values())) + # The component resolved from local source, exactly as an unmarked one. + assert "spec" in task_spec["componentRef"] + assert task_spec["componentRef"]["spec"]["name"] == "Load Orders" + # The marker did not travel into the hydrated graph. Checked STRUCTURALLY: + # the rendered YAML embeds the pipeline source, whose own identifiers + # contain the substring "publish". + assert not _keys_named(hydrated.data, "publish") + + +def test_compilation_emits_a_symbolic_publisher_without_contacting_the_api(tmp_path): + """The emitted owner must be symbolic, not a resolved account. + + Resolving an id at compile time would make compilation require + authentication and network access, and would bake one author's account into + an artifact that another author may legitimately rebuild. + """ + pipeline_path = _project_with(tmp_path, _PUBLISHING_PIPELINE) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + entry = sidecar["load-orders"] + assert entry["publisher"] == "me" + + +def test_the_symbolic_publisher_is_matched_exactly_and_case_sensitively() -> None: + """A literal account id that merely looks like the sentinel stays literal.""" + from tangle_cli.authenticated_identity import is_symbolic_me + + assert is_symbolic_me("me") + for other in ("ME", "Me", " me", "me ", "me@example.com", "", None, True): + assert not is_symbolic_me(other) diff --git a/tests/test_static_client.py b/tests/test_static_client.py index 9668b13..418930c 100644 --- a/tests/test_static_client.py +++ b/tests/test_static_client.py @@ -830,3 +830,97 @@ def _refresh_auth(self) -> None: assert client.refreshes == 2 assert len(session.calls) == 2 assert session.calls[-1]["headers"]["Authorization"] == "Bearer refreshed-2" + + +# --- exact vs substring owner scoping --------------------------------------- +# +# ``published_by`` and ``published_by_substring`` are distinct public +# arguments. The server query is a SUBSTRING match, so for an exact request the +# server result is only a prefilter and exactness must be enforced locally. +# An owner scope is used as an identity control by callers (owner-scoped +# publication/deprecation, and resolution of publish-marked components), so +# these are contract tests, not cosmetics. + + +def _owner_client(rows: list[dict[str, Any]], *, server_filters: bool = True) -> TangleApiClient: + """A client whose listing layer behaves like the server: substring owner. + + ``server_filters=False`` models a server that returns a row the owner query + should have excluded -- absent publication metadata, or a backend that does + not apply the filter. The client must enforce the identity control itself + rather than trusting the response. + """ + from tangle_cli.models import ComponentInfo + + infos = [ComponentInfo(**row) for row in rows] + + class _Client(TangleApiClient): + def list_published_component_infos( # type: ignore[override] + self, + include_deprecated: bool = False, + name_substring: str | None = None, + published_by_substring: str | None = None, + digest: str | None = None, + *, + fetch_specs: bool = False, + ) -> list[Any]: + out = infos + if published_by_substring and server_filters: + out = [i for i in out if published_by_substring in (i.published_by or "")] + if name_substring: + out = [i for i in out if name_substring.lower() in i.name.lower()] + return out + + return _Client("https://api.test", session=FakeSession([])) + + +def test_exact_published_by_rejects_a_superset_owner_id() -> None: + """``alice`` must not match ``alice2``: the server substring query returns + it, so only the local exact filter can refuse it.""" + client = _owner_client( + [{"name": "orders-loader", "digest": "sha256:lookalike", "published_by": "alice2"}] + ) + + found = client.find_existing_components(["orders-loader"], published_by="alice") + + assert found == [] + + +def test_exact_published_by_accepts_only_the_exact_owner() -> None: + client = _owner_client( + [ + {"name": "orders-loader", "digest": "sha256:theirs", "published_by": "alice2"}, + {"name": "orders-loader", "digest": "sha256:mine", "published_by": "alice"}, + ] + ) + + found = client.find_existing_components(["orders-loader"], published_by="alice") + + assert [i.digest for i in found] == ["sha256:mine"] + + +def test_exact_published_by_rejects_a_row_with_no_owner() -> None: + """A missing owner field is not evidence of ownership, and the client must + not assume the server applied the filter it was given.""" + client = _owner_client( + [{"name": "orders-loader", "digest": "sha256:ownerless", "published_by": None}], + server_filters=False, + ) + + found = client.find_existing_components(["orders-loader"], published_by="alice") + + assert found == [] + + +def test_substring_published_by_keeps_partial_matching() -> None: + """The exactness applies to exact requests only; substring callers rely on + partial matching.""" + client = _owner_client( + [{"name": "orders-loader", "digest": "sha256:bot", "published_by": "team-alpha-bot"}] + ) + + found = client.find_existing_components( + ["orders-loader"], published_by_substring="team-alpha" + ) + + assert [i.digest for i in found] == ["sha256:bot"] diff --git a/tests/test_symbolic_publisher_hydration.py b/tests/test_symbolic_publisher_hydration.py new file mode 100644 index 0000000..fca903a --- /dev/null +++ b/tests/test_symbolic_publisher_hydration.py @@ -0,0 +1,189 @@ +"""Owner scoping of entries whose publisher is the symbolic ``me``. + +The compiler emits ``publisher: me`` because compilation is offline. Hydration +resolves it to the authenticated account, which is what keeps the lookup from +being a global name search in which anyone's same-named component is a +candidate. These tests pin that identity control at the layer every consumer +inherits, not just in one downstream CLI. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from tangle_cli.authenticated_identity import IdentityUnavailableError +from tangle_cli.client import TangleApiClient +from tangle_cli.models import ComponentInfo +from tangle_cli.pipeline_hydrator import PipelineHydrator + +_ME = "user-me" +_GENERATION = {"file": "daily_pulse.py", "function": "load_orders", "image": "img:1"} + + +def _rows(*specs: tuple[str, str | None]) -> list[ComponentInfo]: + return [ + ComponentInfo(name="orders-loader", digest=digest, version="1.0", published_by=owner) + for digest, owner in specs + ] + + +class _Client: + """Client whose listing layer behaves like the server: substring owner.""" + + base_url = "https://tangle.example" + logger = None + + def __init__(self, rows: list[ComponentInfo], *, identity: Any = _ME) -> None: + self._rows = rows + self._identity = identity + self.owner_filters: list[str | None] = [] + + def users_me(self) -> Any: + if isinstance(self._identity, Exception): + raise self._identity + return {"id": self._identity} if self._identity is not None else None + + def list_published_component_infos( + self, + include_deprecated: bool = False, + name_substring: str | None = None, + published_by_substring: str | None = None, + digest: str | None = None, + *, + fetch_specs: bool = False, + ) -> list[ComponentInfo]: + out = self._rows + if published_by_substring: + out = [i for i in out if published_by_substring in (i.published_by or "")] + if name_substring: + out = [i for i in out if name_substring.lower() in i.name.lower()] + return out + + def find_existing_components(self, *args: Any, **kwargs: Any) -> list[ComponentInfo]: + self.owner_filters.append(kwargs.get("published_by")) + return TangleApiClient.find_existing_components(self, *args, **kwargs) + + def get_component_spec(self, digest: str) -> dict[str, Any]: + return {"name": "orders-loader", "digest": digest} + + +def _entry(**over: Any) -> dict[str, Any]: + base = { + "name": "orders-loader", + "version": "1.0", + "publisher": "me", + "local_from_python": dict(_GENERATION), + "publish": True, + } + base.update(over) + return base + + +def test_symbolic_me_resolves_to_the_authenticated_account() -> None: + client = _Client(_rows(("sha256:mine", _ME))) + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters(_entry()) + + assert resolved is not None + assert resolved[0] == "sha256:mine" + assert client.owner_filters == [_ME] # the symbol never reaches the API + + +def test_a_foreign_component_is_not_selected_even_when_returned_first() -> None: + client = _Client(_rows(("sha256:attacker", "attacker"), ("sha256:mine", _ME))) + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters(_entry()) + + assert resolved is not None + assert resolved[0] == "sha256:mine" + + +def test_a_foreign_only_collision_yields_no_candidate() -> None: + """Someone else holding the name must not supply the code; the caller then + falls back to the local candidate.""" + client = _Client(_rows(("sha256:attacker", "attacker"))) + + assert PipelineHydrator(client=client)._resolve_by_name_with_filters(_entry()) is None + + +@pytest.mark.parametrize( + "identity", + [ + pytest.param(RuntimeError("auth expired"), id="lookup-raises"), + pytest.param(None, id="no-user"), + pytest.param("", id="empty-id"), + ], +) +def test_an_unresolvable_identity_fails_closed(identity: Any) -> None: + """Never widen to a global search: that is the vulnerability.""" + client = _Client(_rows(("sha256:mine", _ME)), identity=identity) + + with pytest.raises(IdentityUnavailableError, match="unscoped"): + PipelineHydrator(client=client)._resolve_by_name_with_filters(_entry()) + assert client.owner_filters == [] # no request was issued at all + + +def test_a_literal_publisher_id_is_used_verbatim() -> None: + client = _Client(_rows(("sha256:theirs", "other-team"))) + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters( + _entry(publisher="other-team") + ) + + assert resolved is not None + assert client.owner_filters == ["other-team"] + + +def test_an_entry_without_a_publisher_keeps_the_existing_global_behaviour() -> None: + """Hand-authored cross-publisher resolution is unchanged.""" + client = _Client(_rows(("sha256:someone-elses", "someone-else"))) + entry = {"name": "orders-loader", "version": "1.0"} + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters(entry) + + assert resolved is not None + assert client.owner_filters == [None] + + +def test_a_publisher_that_merely_resembles_the_sentinel_stays_literal() -> None: + """``ME`` is not ``me``: only the exact symbol is self-referential, so a + real account id is never swapped for the current user.""" + client = _Client(_rows(("sha256:odd", "ME"))) + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters( + _entry(publisher="ME") + ) + + assert resolved is not None + assert client.owner_filters == ["ME"] + + +def test_an_unmarked_entry_treats_me_as_a_literal_account_id() -> None: + """``me`` is only self-referential on a compiler-authored marked entry. + + Hand-authored configs predate the sentinel, where ``publisher: me`` meant a + registry account literally named ``me`` -- the API does not reserve the + string. Substituting there would silently retarget an existing config, or + fail it on an auth error, so the marker is required too. + """ + client = _Client(_rows(("sha256:literal-me", "me"))) + entry = {"name": "orders-loader", "version": "1.0", "publisher": "me"} + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters(entry) + + assert resolved is not None + assert resolved[0] == "sha256:literal-me" + assert client.owner_filters == ["me"] # passed through, not substituted + + +def test_an_unmarked_entry_with_me_does_not_require_authentication() -> None: + """The compatibility guarantee has teeth: an unmarked config must keep + resolving even when no account can be determined.""" + client = _Client(_rows(("sha256:literal-me", "me")), identity=RuntimeError("no auth")) + entry = {"name": "orders-loader", "version": "1.0", "publisher": "me"} + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters(entry) + + assert resolved is not None diff --git a/uv.lock b/uv.lock index 8c0b313..ac7fbb8 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.14" +version = "0.1.15" source = { editable = "." } dependencies = [ { name = "cloud-pipelines" },