diff --git a/docs/CONTRACT_LOADING_API.md b/docs/CONTRACT_LOADING_API.md new file mode 100644 index 00000000..0b689f78 --- /dev/null +++ b/docs/CONTRACT_LOADING_API.md @@ -0,0 +1,243 @@ +# Loading a contract as `fluid plan` sees it + +`fluid_build.api.load_contract` returns the contract that `fluid plan` plans: +the same dict `plan.json` embeds as `contract`, after every rewrite the engine +makes on the way in. Use it when your code has to agree with the engine about +what a contract *is*: comparing a contract to the plan made from it, diffing +two revisions, rendering a contract in a UI, or checking it in CI. + +Added in `fluid_build.api` **1.1**. + +## Examples + +### Load a contract file + +```python +from fluid_build.api import load_contract + +loaded = load_contract("contracts/orders/contract.fluid.yaml", env="prod") + +loaded.contract # dict, the contract `fluid plan ... --env prod` plans +loaded.digest # "sha256:…", the plan digest's canonicalisation of that dict +loaded.files # (contract.fluid.yaml, parts/orders.yaml, overlays/prod.yaml) +loaded.overlay # Path(".../overlays/prod.yaml") +``` + +`env` is an environment *name*, never a path. The engine turns `env` into +overlay paths (`overlays/.yaml`, `.json`, …), so an env holding `..` +or an absolute path would merge whichever `.yaml`/`.yml`/`.json` file it named +into the returned contract. An env that is not a single path component is +refused before a file is read: + +```python +load_contract("contracts/orders/contract.fluid.yaml", env="../../home/me/.docker/config") +# ContractLoadError: contract_env_invalid +``` + +Refused: `""` (not read as `None`), `.`, `..`, anything holding `/`, `\` or a +NUL byte (so any absolute path), and a drive-qualified name (`C:prod`). The rule +is the same on every platform: `C:prod` is refused on Linux and macOS too, so +one env never means two things. Every other string loads exactly as +`fluid plan --env` loads it, including names `fluid publish --env` would not +accept (`_staging`, `prod+eu`, a name longer than 64 characters). Pass `None` +for no env. + +A `fluid bundle` archive loads the same way, and is refused for an env it was +not built for, as on the CLI: + +```python +load_contract("runtime/bundle.tgz", env="prod") +``` + +### Is this the contract that plan was made from? + +```python +import json + +from fluid_build.api import load_contract_from_text +from fluid_build.forge.core.plan_digest import compute_contract_digest + +plan = json.loads(open("runtime/plan.json").read()) +submitted = load_contract_from_text(contract_yaml) # text from a form, a DB row, a PR + +if submitted.unresolved_refs: + raise ValueError(f"contract references files: {submitted.unresolved_refs}") +same = submitted.digest == compute_contract_digest(plan["contract"]) +``` + +Formatting, comments, key order, quoting, Unicode normal form, an alias beside +its canonical value (`format: bigquery-table` / `bigquery_table`) and a legacy +`build:` beside `builds:` do not change the digest. Every other value does. + +Compare digests rather than dicts. `loaded.contract` keeps a YAML magic-word +or numeric key (`on:`, `no:`, `1:` in an open block such as `extensions`) as +the Python `bool` / `int` the engine plans with, while `plan.json` writes every +key as a string. The digest coerces keys the same way `plan.json` does. + +`loaded.digest` is the digest of the **planned** contract: normalised, +composed and overlaid. It is not the value `fluid contract digest` prints, and +not what a federation `upstreamDigest` pins: those hash the file as parsed, +before any alias rewrite, `$ref` or overlay. The two differ whenever the file +uses one of those, so do not pin `upstreamDigest` from `loaded.digest`. + +### Contract text or a parsed dict, without touching the disk + +```python +from fluid_build.api import load_contract_from_dict, load_contract_from_text + +loaded = load_contract_from_text(text) # YAML by default +loaded = load_contract_from_text(text, suffix=".json") # or JSON +loaded = load_contract_from_dict(document) # already parsed +``` + +Neither reads a file. A `$ref` is left in place and listed in +`loaded.unresolved_refs`; you decide whether that is an error. + +### Text with its fragments and an overlay + +```python +loaded = load_contract_from_text( + text, + base_dir="contracts/orders", # resolve $ref against this directory + overlay={"exposes": [{"binding": {"location": {"database": "prod_s"}}}]}, +) +``` + +With `base_dir` set to a contract's directory and `overlay` set to the parsed +overlay file `--env` would select, the result equals +`load_contract(that_file, env=...)`, including the case below where the engine +drops the overlay. + +Without `base_dir`, a document holding file `$ref` values cannot say whether +the engine would apply an overlay (that depends on what the fragments hold), +so passing `overlay` then raises `contract_overlay_needs_base_dir` instead of +guessing. + +## What "as plan sees it" means + +In the engine's order: + +1. **Parse**: JSON, or YAML through the engine's billion-laughs guard. +2. **`$ref` composition**: each `{"$ref": "./file.yaml#/pointer"}` replaced by + its target, resolved against the contract's directory (or `base_dir`) by the + engine's resolver and under its path rules. Same-document `#/...` pointers + are kept, as the engine keeps them, and listed in `unresolved_refs`. +3. **Overlay**: for `env`, the first of `overlays/.yaml|yml|json`, + `.yaml|yml|json`, `..yaml|yml|json` next to the + contract, deep-merged over the base (objects key by key, lists of objects by + position, anything else replaced). A bundle is never re-overlaid. +4. **Alias values**: human-friendly values rewritten to the schema's enum + value, for example `source.kind: pg` → `postgres`, `source.mode: incremental` + → `incremental_append`, `binding.format: kafka` → `kafka_topic`, + `iceberg-table` → `iceberg`. +5. **Legacy `build:`**: a singular `build:` becomes `builds: [build]`; when + both are present `builds:` wins. + +Step 4 runs before step 5, exactly as in the engine, so an alias under a +legacy singular `build:` is **not** rewritten (and `fluid plan` then rejects +it at the schema gate). Write `builds:` to get alias rewriting for builds. + +Validation is not part of loading: a schema-invalid contract loads, and +`fluid validate` / `fluid plan` reject it. + +### Known engine behaviour: an overlay next to a `$ref` + +When the overlay file, or the merged contract, still holds a `$ref` (an +overlay that references a fragment, or a same-document `#/...` pointer in +the contract), the engine's auto-bundle step reloads the contract from the +base file and the overlay is dropped: `fluid plan --env prod` plans the base +contract. `load_contract` returns what plan plans, so it returns the base too, +but its provenance does not pretend otherwise: `overlay` is `None`, the +overlay is not in `files`, and a `contract_overlay_not_applied` WARNING names +the file. The in-memory forms do the same with an `overlay` mapping: they +return the base and log the same WARNING (to `logger`, or the +`fluid.api.contract` logger). Keep `$ref` out of overlays, and use file +references rather than `#/...` pointers in a contract that has overlays. + +## Reference + +### `load_contract(path, *, env=None, logger=None) -> LoadedContract` + +Loads a contract file or a bundle through `fluid plan`'s own loader. `path` is +resolved to an absolute path first, as `fluid plan` does. The CLI's gate on +operator-typed paths (no `..`, no symlink) is not applied to `path`; a library +caller chooses its own paths. `env` must be a single path component (see the +example above) or `None`; anything else raises `contract_env_invalid`. + +### `load_contract_from_text(text, *, suffix=".yaml", base_dir=None, overlay=None, logger=None) -> LoadedContract` + +Parses `text` with the engine's parser (`suffix` picks it as a file extension +would), then loads the result as `load_contract_from_dict` does. + +### `load_contract_from_dict(document, *, base_dir=None, overlay=None, logger=None) -> LoadedContract` + +Loads a parsed document. `document` and `overlay` are never modified. `logger` +receives the `contract_overlay_not_applied` WARNING. + +### `LoadedContract` + +A frozen dataclass. + +| Field | Type | Meaning | +|---|---|---| +| `contract` | `dict` | The contract as planned, keys as the engine holds them (see the digest note above). A fresh dict each call; yours to mutate. | +| `origin` | `"file"` \| `"bundle"` \| `"memory"` | Which entry point and input shape produced it. | +| `source` | `Path \| None` | The resolved contract or bundle path. | +| `env` | `str \| None` | The env requested. | +| `overlay` | `Path \| None` | The overlay file merged for `env` (never set for a bundle; see the known engine behaviour above). | +| `files` | `tuple[Path, ...]` | Every file composed: the source, each `$ref` target in first-read order, the overlay. | +| `unresolved_refs` | `tuple[str, ...]` | `$ref` values left in `contract`, in document order. | +| `digest` | `str` (property) | `sha256:` of the planned contract via `compute_contract_digest`, the plan digest's canonicalisation. Not the `fluid contract digest` / `upstreamDigest` value. Raises `contract_not_serialisable` when JSON cannot represent the contract. | + +### `ContractLoadError` + +Every failure raises `ContractLoadError` with a stable `event`, the path when +there is one, and the engine's exception as `__cause__`: + +| `event` | When | +|---|---| +| `contract_not_found` | The contract file does not exist, or `path` / `base_dir` cannot name a file (it holds a NUL byte). | +| `contract_parse_failed` | The text is not valid JSON/YAML, is not UTF-8, or trips the YAML size/anchor guard. | +| `contract_not_a_mapping` | The document (or overlay) root is not an object, from a file (with or without `env`, an overlay or not) or from text. | +| `contract_ref_unresolved` | A `$ref` target is missing, cyclic, blocked, or its pointer does not resolve. | +| `contract_env_invalid` | `env` is not a single path component: it is empty, `.` or `..`, holds `/`, `\` or NUL, or is drive-qualified (`C:prod`, on every platform). | +| `contract_overlay_needs_base_dir` | In-memory form: `overlay` given for a document with file `$ref` values but no `base_dir`. | +| `contract_not_serialisable` | Raised by `.digest`: the contract holds a value JSON cannot represent (an unquoted YAML date, a set, binary, a self-referencing alias). `fluid plan` cannot write it either; quote the value. | +| `contract_load_failed` | Any other loader failure, including a document that contains itself through a YAML alias (the engine fails on it too). | +| *engine event* | Passed through unchanged, e.g. `overlay_declared_but_missing`, `bundle_not_found` (a `.tgz` path that does not exist), `bundle_env_mismatch`, `bundle_manifest_invalid`. | + +## Stability + +`fluid_build.api` is governed by SemVer through `fluid_build.api.__api_version__` +(`tests/api/test_api_surface_snapshot.py` locks the exported names). The +behavioural promise: `load_contract(path, env=env).contract` is the contract +`fluid plan path --env env` plans, equal to `plan.json["contract"]` once keys +are written as strings (and always equal by `.digest`). + +How each form keeps that promise: + +- **The file form** (`load_contract`) calls the engine's loader itself, so a + new step in the loader reaches it with no change here. + `tests/api/test_contract_load.py` runs the real `fluid plan` on fixtures + that exercise every rewrite (alias values, legacy `build:`, `$ref`, overlay, + bundle, a magic-word key) and fails if the two differ. +- **The in-memory forms** have no file to hand the loader, so they replay a + fixed sequence: `$ref` resolution, the overlay merge and the auto-bundle + step's decision to drop it, then the loader's rewrites by name. The tests + pin them to the file form on the same fixtures, and a guard test parses the + engine loader's source (`load_contract_with_overlay`) and fails when it + gains, loses or reorders a call that takes `contract` or a part of it + (positionally or by keyword), changes `contract` any other way (an + assignment that is not a known step, an item or attribute write, a method + call, `del`, a rebinding), returns anything but `contract`, or reads + `contract` anywhere else (an alias, a loop over it, a container holding + it). It does not decide whether such a read changes the contract: it fails + on it, so a new step there turns the build red until the in-memory forms + replay it too. The guard reads that one function, outside its bundle + branch: a change inside a function it calls (`load_with_overlay`, the `$ref` + resolver, a rewrite itself) is caught only where the fixtures exercise it. + +Do not import the helpers in `fluid_build._contract_loader` (for example +`_normalize_contract_aliases` / `_normalize_singular_build_key`) to reproduce +this: they are private, their order matters (see above), and they are not +the whole pipeline. diff --git a/fluid_build/api/__init__.py b/fluid_build/api/__init__.py index 6fc234ea..1065750c 100644 --- a/fluid_build/api/__init__.py +++ b/fluid_build/api/__init__.py @@ -20,11 +20,24 @@ Anything outside ``fluid_build.api`` is internal and may change without notice. Anything inside is governed. + +Consumers that need a contract exactly as ``fluid plan`` sees it (a +control plane comparing a contract to a plan, CI tooling, editors) use +:func:`load_contract` and its in-memory siblings from ``.contract`` +(added in 1.1). """ from __future__ import annotations from .catalog import CatalogRegistrar, RegistrationResult +from .contract import ( + ContractLoadError, + ContractOrigin, + LoadedContract, + load_contract, + load_contract_from_dict, + load_contract_from_text, +) from .cost import BudgetCap, ChargebackTag, CostTracker from .hooks import HookChain, HookResult, PreLandHook from .lineage import DatasetFacet, LineageEmitter, RunEvent @@ -42,7 +55,7 @@ ) from .state import Cursor, RunLock, StateStore, Watermark -__api_version__ = "1.0" +__api_version__ = "1.1" __all__ = [ "__api_version__", @@ -96,4 +109,11 @@ # security "ImageSignatureVerifier", "SovereigntyChecker", + # contract loading (1.1) + "LoadedContract", + "ContractLoadError", + "ContractOrigin", + "load_contract", + "load_contract_from_text", + "load_contract_from_dict", ] diff --git a/fluid_build/api/contract.py b/fluid_build/api/contract.py new file mode 100644 index 00000000..2240a49e --- /dev/null +++ b/fluid_build/api/contract.py @@ -0,0 +1,737 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Load a contract exactly as ``fluid plan`` and ``fluid apply`` see it. + +Three entry points, one result type: + +* :func:`load_contract` reads a contract file (or a ``fluid bundle`` ``.tgz``) + through the engine's own loader, the function ``fluid plan`` calls. The + returned ``contract`` is the dict ``plan.json`` embeds as ``contract``. +* :func:`load_contract_from_text` parses contract text held in memory. +* :func:`load_contract_from_dict` takes an already-parsed document. + +The two in-memory forms never touch the filesystem unless ``base_dir`` is +passed; without it a ``$ref`` is left in place and listed in +:attr:`LoadedContract.unresolved_refs`, so the caller decides what an +unresolved reference means. + +What "as plan sees it" covers, in the engine's order: + +1. parsing (JSON, or YAML with the billion-laughs guard); +2. ``$ref`` composition against the contract's directory (``base_dir`` for + the in-memory forms); +3. the environment overlay (``env`` for a file, ``overlay`` for the + in-memory forms), deep-merged over the base; +4. alias values rewritten to their canonical enum value + (``binding.format: bigquery-table`` becomes ``bigquery_table``); +5. a legacy singular ``build:`` rewritten to ``builds: [build]``. + +This module only composes the engine's functions; it adds no rewrite of its +own. :func:`load_contract` calls the engine's loader itself, so a new step +there reaches it with no change here. The in-memory forms have no file to +hand that loader, so they replay its steps: the auto-bundle decision (which +drops an overlay when a ``$ref`` survives), then the rewrites named in +:data:`_ENGINE_REWRITES`. ``tests/api/test_contract_load.py`` pins the file +form to the ``contract`` of a real ``fluid plan`` run, pins the in-memory +forms to the file form, and parses the engine loader's source to fail when +it gains a step the in-memory forms do not replay. + +Part of the governed ``fluid_build.api`` surface: SemVer applies through +``fluid_build.api.__api_version__``. +""" + +from __future__ import annotations + +import copy +import logging +import ntpath +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Dict, List, Literal, Mapping, Optional, Set, Tuple, Union + +__all__ = [ + "ContractLoadError", + "ContractOrigin", + "LoadedContract", + "load_contract", + "load_contract_from_dict", + "load_contract_from_text", +] + +LOG = logging.getLogger("fluid.api.contract") + +#: Where a :class:`LoadedContract` came from. +ContractOrigin = Literal["file", "bundle", "memory"] + +PathLike = Union[str, "os.PathLike[str]"] + +# Aligned with ``fluid_build.loader._MAX_REF_DEPTH``: the provenance walk +# never descends further than the resolver it describes. +_MAX_REF_DEPTH = 20 + +#: The rewrites ``_contract_loader.load_contract_with_overlay`` applies after +#: its auto-bundle step, in its order, each ``contract -> contract``. The +#: in-memory forms replay exactly these, by name; a guard test parses the +#: engine function and fails when its sequence and this tuple differ. +_ENGINE_REWRITES: Tuple[str, ...] = ( + "_normalize_contract_aliases", + "_normalize_singular_build_key", +) + + +class ContractLoadError(Exception): + """A contract could not be loaded. + + ``event`` is a stable snake_case identity, safe to route on: + + * ``contract_not_found``: the contract (or a file it names) does not exist, + or ``path`` / ``base_dir`` cannot name a file (a NUL byte); + * ``contract_parse_failed``: the text is not valid JSON/YAML (or not + UTF-8); + * ``contract_not_a_mapping``: the document root, or the overlay root, is + not an object; + * ``contract_ref_unresolved``: a ``$ref`` could not be resolved + (missing target, cycle, blocked path, bad pointer); + * ``contract_env_invalid``: ``env`` is not a single path component (see + :func:`load_contract`); + * ``contract_overlay_needs_base_dir``: an in-memory load was given an + overlay and a document with file ``$ref`` values but no ``base_dir``, + so whether the engine would apply the overlay cannot be decided; + * ``contract_not_serialisable``: raised by :attr:`LoadedContract.digest` + for a contract JSON cannot represent (an unquoted YAML date, a set, + binary, a self-referencing alias); ``fluid plan`` cannot write it either; + * ``contract_load_failed``: any other loader failure, including a document + that contains itself through a YAML alias; + * any event the engine's loader raises itself, passed through unchanged + (for example ``overlay_declared_but_missing``, ``bundle_not_found``, + ``bundle_env_mismatch``, ``bundle_manifest_invalid``). + + The underlying exception is chained as ``__cause__``. + """ + + def __init__(self, event: str, message: str, *, path: Optional[Path] = None) -> None: + super().__init__(message) + self.event = event + self.message = message + self.path = path + + +@dataclass(frozen=True) +class LoadedContract: + """A contract as the engine plans it, plus where it came from. + + ``contract`` is a fresh dict owned by the caller: nothing else holds a + reference to it, and mutating it changes no later load. + """ + + #: The contract dict the engine plans for the same input and env. Equal to + #: ``plan.json``'s ``contract`` once non-string keys are written as + #: strings, as ``plan.json`` writes them: a YAML ``on:`` / ``no:`` / ``1:`` + #: key in an open block stays a ``bool`` / ``int`` here, as it is inside + #: the engine. Compare contracts with :attr:`digest`, which coerces keys. + contract: Dict[str, Any] + #: ``"file"``, ``"bundle"`` (a ``fluid bundle`` ``.tgz``) or ``"memory"``. + origin: ContractOrigin + #: The resolved contract or bundle path; ``None`` for the in-memory forms. + source: Optional[Path] = None + #: The environment requested (``env=`` of :func:`load_contract`). + env: Optional[str] = None + #: The overlay file merged into ``contract`` for ``env``; ``None`` when no + #: env was requested, none matched, the engine did not apply the one it + #: found (logged as ``contract_overlay_not_applied``), or the source is a + #: bundle (whose overlay was applied when it was built). + overlay: Optional[Path] = None + #: Every file composed into ``contract``, in the order the engine reads + #: them: the source, then each ``$ref`` target (first occurrence), then + #: the overlay. Empty for the in-memory forms without ``base_dir``. + files: Tuple[Path, ...] = field(default_factory=tuple) + #: ``$ref`` values left in ``contract``, in document order: same-document + #: ``#/...`` pointers (the engine keeps them), and for the in-memory forms + #: without ``base_dir`` every reference. + unresolved_refs: Tuple[str, ...] = field(default_factory=tuple) + + @property + def digest(self) -> str: + """``sha256:`` of the *planned* ``contract``, canonicalised as ``planDigest`` is. + + Computed with ``forge.core.plan_digest.compute_contract_digest`` over + ``contract``, the normalised, composed and overlaid dict. Two inputs + with equal digests are one contract to ``fluid plan``: formatting, + comments, key order, quoting, Unicode normal form, an alias beside its + canonical value and a legacy ``build:`` beside ``builds:`` do not + count. + + It is **not** the value ``fluid contract digest`` prints or a + federation ``upstreamDigest`` pins: those hash the file as parsed, + before any rewrite, ``$ref`` or overlay, so they differ whenever the + file uses one. + + Raises: + ContractLoadError: ``contract_not_serialisable`` when JSON cannot + represent ``contract`` (``fluid plan`` fails on it too). + """ + from fluid_build.forge.core.plan_digest import compute_contract_digest + + try: + return compute_contract_digest(self.contract) + except (TypeError, ValueError, RecursionError) as exc: + raise ContractLoadError( + "contract_not_serialisable", + f"the contract cannot be written as JSON, so it has no digest: {exc}", + path=self.source, + ) from exc + + +def load_contract( + path: PathLike, + *, + env: Optional[str] = None, + logger: Optional[logging.Logger] = None, +) -> LoadedContract: + """Load the contract at ``path`` as ``fluid plan [--env ]`` does. + + ``path`` is a contract file (``.yaml`` / ``.yml`` / ``.json``) or a + ``fluid bundle`` archive (``.tgz`` / ``.tar.gz``); a bundle is never + re-overlaid, and an ``env`` it was not built for is refused + (``bundle_env_mismatch``), exactly as on the CLI. + + The operator-path gate the CLI applies to its own arguments (no ``..``, + no symlink) is not applied to ``path``: a library caller chooses its + paths. The ``$ref`` resolver's own confinement applies in full. + + ``env`` is a name, never a path. The engine builds overlay paths from it + (``overlays/.yaml`` and so on), so an env such as ``../x`` or + ``/abs/x`` would merge a file outside the contract's directory into the + result. An env that is not a single path component is therefore refused + with ``contract_env_invalid`` before any file is read: one holding ``/``, + ``\\`` or a NUL (so every absolute one), a drive-qualified one (``C:prod``, + on every platform), ``.``, ``..``, and ``""`` (refused rather than read as + ``None``). Every other string loads as ``fluid plan --env`` loads it. + ``None`` means no env. + + Raises: + ContractLoadError: the contract could not be loaded. + """ + from fluid_build import _contract_loader + + log = logger or LOG + resolved = _resolve_input_path(path, "contract path") + if env is not None and not _is_env_component(env): + raise ContractLoadError( + "contract_env_invalid", + f"env {env!r} is not an environment name: {_ENV_RULE}", + path=resolved, + ) + if env is not None and not _contract_loader._is_bundle_path(str(resolved)): + _refuse_a_base_root_the_overlay_merge_would_coerce(resolved) + try: + contract = _contract_loader.load_contract_with_overlay(str(resolved), env, log) + except Exception as exc: # noqa: BLE001 - every failure is mapped to one typed error + raise _as_load_error(exc, resolved) from exc + if not isinstance(contract, dict): + # A JSON file whose root is an array loads without error and then + # fails somewhere inside the planner; say so here instead. + raise _not_a_mapping(contract, resolved) + + if _contract_loader._is_bundle_path(str(resolved)): + return LoadedContract( + contract=contract, + origin="bundle", + source=resolved, + env=env, + files=(resolved,), + unresolved_refs=_ref_values(contract), + ) + + # Provenance re-reads the files the load just read; a file changed or + # removed in between surfaces as the same typed error a load would raise. + try: + overlay = _applied_overlay(resolved, env, contract, log) + ref_files = _composed_ref_files(resolved) + except Exception as exc: # noqa: BLE001 - mapped to one typed error + raise _as_load_error(exc, resolved) from exc + files: List[Path] = [resolved] + for ref_file in ref_files: + if ref_file not in files: + files.append(ref_file) + if overlay is not None: + files.append(overlay) + return LoadedContract( + contract=contract, + origin="file", + source=resolved, + env=env, + overlay=overlay, + files=tuple(files), + unresolved_refs=_ref_values(contract), + ) + + +def load_contract_from_text( + text: str, + *, + suffix: str = ".yaml", + base_dir: Optional[PathLike] = None, + overlay: Optional[Mapping[str, Any]] = None, + logger: Optional[logging.Logger] = None, +) -> LoadedContract: + """Parse contract ``text`` and load it as :func:`load_contract` would load that file. + + ``suffix`` selects the parser the way a file extension does: ``.json``, + ``.yaml`` / ``.yml``, anything else tries JSON then YAML. YAML goes + through the engine's billion-laughs guard. + + Equal to ``load_contract(f).contract`` for a file ``f`` holding ``text`` + when ``base_dir`` is ``f``'s directory and ``overlay`` is the parsed + overlay ``load_contract(f, env=...)`` would select. See + :func:`load_contract_from_dict` for ``base_dir`` and ``overlay``. + + Raises: + ContractLoadError: ``contract_parse_failed``, ``contract_not_a_mapping``, + ``contract_overlay_needs_base_dir``, or a ``$ref`` failure when + ``base_dir`` is given. + """ + from fluid_build import loader + + try: + document = loader.parse_contract_text(text, suffix=suffix) + except Exception as exc: # noqa: BLE001 - mapped to one typed error + raise _parse_error(exc) from exc + return load_contract_from_dict(document, base_dir=base_dir, overlay=overlay, logger=logger) + + +def load_contract_from_dict( + document: Mapping[str, Any], + *, + base_dir: Optional[PathLike] = None, + overlay: Optional[Mapping[str, Any]] = None, + logger: Optional[logging.Logger] = None, +) -> LoadedContract: + """Load an already-parsed contract ``document`` as the engine would. + + * ``base_dir`` given: each ``$ref`` is resolved against it with the + engine's resolver, reading the files it names (listed in ``files``). + Not given: nothing is read from disk, and every ``$ref`` stays in + place, listed in ``unresolved_refs``. + * ``overlay`` given: deep-merged over the base after ``$ref`` resolution, + with the engine's merge (dicts key by key, lists of objects by + position, anything else replaced), as an overlay file is. Exactly as + in the engine, the overlay is **not** applied when it holds a ``$ref`` + or when the merged contract still holds one (a same-document ``#/...`` + pointer): ``fluid plan --env`` plans the base in that shape, so this + returns the base and logs a ``contract_overlay_not_applied`` WARNING. + Without ``base_dir``, a document holding file ``$ref`` values leaves + that decision open (it depends on what the fragments hold), so an + overlay is then refused with ``contract_overlay_needs_base_dir``. + + ``document`` and ``overlay`` are never modified. + + Raises: + ContractLoadError: ``contract_not_a_mapping``, + ``contract_overlay_needs_base_dir``, or a ``$ref`` failure when + ``base_dir`` is given. + """ + from fluid_build import _contract_loader, loader + + log = logger or LOG + + if not isinstance(document, Mapping): + raise ContractLoadError( + "contract_not_a_mapping", + f"a contract document must be a mapping, got {type(document).__name__}", + ) + if overlay is not None and not isinstance(overlay, Mapping): + raise ContractLoadError( + "contract_not_a_mapping", + f"an overlay document must be a mapping, got {type(overlay).__name__}", + ) + + contract: Dict[str, Any] = copy.deepcopy(dict(document)) + files: Tuple[Path, ...] = () + if base_dir is not None: + base = _resolve_input_path(base_dir, "base_dir") + # ``loader.load_contract`` does exactly this after parsing the file. + try: + contract = loader._resolve_refs(contract, base) + except Exception as exc: # noqa: BLE001 - mapped to one typed error + raise _as_load_error(exc, base) from exc + files = tuple(_walk_ref_files(document, base)) + else: + # The resolver rebuilds every dict and list it passes through, so in + # the engine no two places in the base contract share one object + # (``deepcopy`` keeps YAML alias sharing). Without that, the overlay + # merge and the rewrites below, which change nodes in place, would + # reach every alias of the node they change. + contract = _unshare(contract) + if overlay is not None: + contract = _replay_overlay( + contract, copy.deepcopy(dict(overlay)), composed=base_dir is not None, log=log + ) + # ``_contract_loader.load_contract_with_overlay``'s rewrites, in its order. + for rewrite in _ENGINE_REWRITES: + contract = getattr(_contract_loader, rewrite)(contract) + return LoadedContract( + contract=contract, + origin="memory", + files=files, + unresolved_refs=_ref_values(contract), + ) + + +# ── helpers ──────────────────────────────────────────────────────────── + +#: :func:`_is_env_component` in words, for ``contract_env_invalid``. +_ENV_RULE = ( + "an env names an overlay file next to the contract, so it must be one path " + "component: not empty, not '.' or '..', no '/', '\\' or NUL, not drive-qualified ('C:prod')" +) + + +def _is_env_component(env: Any) -> bool: + """True when ``env`` keeps every overlay path the engine builds from it in the + contract's directory. + + Only what makes ``env`` a path is refused, so every other name loads as + ``fluid plan --env`` loads it (``_staging``, ``prod+eu``, a long name). + The rule does not depend on the platform, so one env means the same thing + everywhere: both separators are refused, which also refuses every absolute + path, and a Windows drive (``C:prod``) is found with ``ntpath`` on every + platform, not with ``os.path``, which finds none on POSIX. + """ + if not isinstance(env, str) or env in ("", ".", ".."): + return False + if "\x00" in env or "/" in env or "\\" in env: + return False + return not ntpath.splitdrive(env)[0] + + +def _not_a_mapping(root: Any, path: Path) -> ContractLoadError: + return ContractLoadError( + "contract_not_a_mapping", + f"the contract root must be an object, got {type(root).__name__}", + path=path, + ) + + +def _refuse_a_base_root_the_overlay_merge_would_coerce(contract_path: Path) -> None: + """Raise ``contract_not_a_mapping`` when the base ``load_with_overlay`` merges is + not an object. + + The loader checks a YAML root, but not a JSON one (or one a root ``$ref`` + composes). With an overlay present it merges into ``dict(base)``, which + turns a list root into a dict (``[["k", "v"]]``, ``[{"a": 1, "b": 2}]``, + ``[]``) or fails with a message that names no root, so the same file + would load, or fail with ``contract_load_failed``, only because an + overlay exists. Without an env, :func:`load_contract`'s post-load check + catches it. + """ + from fluid_build import loader + + try: + # The base exactly as ``load_with_overlay`` reads it, ``$ref`` composed. + base = loader.load_contract(contract_path) + except Exception: # noqa: BLE001 - the engine's own load raises it, typed there + return + if not isinstance(base, dict): + raise _not_a_mapping(base, contract_path) + + +def _resolve_input_path(value: PathLike, what: str) -> Path: + """``value`` made absolute; a path the OS cannot hold is ``contract_not_found``.""" + try: + return Path(os.fspath(value)).resolve() + except (OSError, ValueError, RuntimeError) as exc: + # ``ValueError``: a NUL byte. ``RuntimeError``: a symlink loop + # (Python < 3.13). No file can exist at such a path. + raise ContractLoadError( + "contract_not_found", f"the {what} {value!r} cannot name a file: {exc}" + ) from exc + + +def _unshare(node: Any) -> Any: + """``node`` with every dict and list rebuilt, so no two places share one object. + + What ``loader._resolve_refs`` does to the trees it walks, without its + recursion: iterative, so depth costs no stack. A container that contains + itself (a self-referencing YAML alias) cannot be rebuilt; the engine's + resolver fails on it too, and so does this, with ``contract_load_failed``. + """ + if not isinstance(node, (dict, list)): + return node + root: Any = {} if isinstance(node, dict) else [] + on_path: Set[int] = set() + stack: List[Tuple[Any, Any]] = [(node, root)] + while stack: + source, target = stack.pop() + if target is None: # every child of ``source`` is rebuilt + on_path.discard(id(source)) + continue + if id(source) in on_path: + raise ContractLoadError( + "contract_load_failed", + "the contract contains itself through a YAML alias, so it cannot be " + "loaded (the engine's loader fails on it too)", + ) + on_path.add(id(source)) + stack.append((source, None)) + items = source.items() if isinstance(source, dict) else enumerate(source) + for key, value in items: + child = value + if isinstance(value, (dict, list)): + child = {} if isinstance(value, dict) else [] + stack.append((value, child)) + if isinstance(target, dict): + target[key] = child + else: + target.append(child) + return root + + +def _replay_overlay( + contract: Dict[str, Any], + overlay: Dict[str, Any], + *, + composed: bool, + log: logging.Logger, +) -> Dict[str, Any]: + """The engine's overlay merge plus its auto-bundle decision, for a parsed overlay. + + ``loader.load_with_overlay`` deep-merges the overlay; then + ``_contract_loader._auto_bundle_if_needed`` sees a ``$ref`` left in the + merged contract (from the overlay, or a same-document pointer) and + reloads the base file, which drops the overlay. ``contract`` is that + reloaded base when ``composed`` (its file references were resolved + against ``base_dir``), so dropping the overlay means returning it. + """ + from fluid_build import _contract_loader, loader + + if not _contract_loader._has_ref_pointers(overlay): + if not composed and any(loader._parse_ref(ref)[0] for ref in _ref_values(contract)): + raise ContractLoadError( + "contract_overlay_needs_base_dir", + "the document holds file $ref values and no base_dir was given, so " + "whether the engine applies the overlay cannot be decided: pass " + "base_dir (the contract's directory) to load it as fluid plan does", + ) + # ``_deep_merge`` mutates nested dicts of its base: merge into a copy, + # so the un-overlaid contract survives if the overlay is dropped. + merged = loader._deep_merge(copy.deepcopy(contract), overlay) + if not _contract_loader._has_ref_pointers(merged): + return merged + log.warning( + "contract_overlay_not_applied: the overlay was not merged, because the " + "overlay or the contract holds a $ref; `fluid plan --env` plans the base " + "contract in this shape too", + extra={"event": "contract_overlay_not_applied"}, + ) + return contract + + +def _cause_chain(exc: BaseException) -> List[BaseException]: + chain: List[BaseException] = [] + current: Optional[BaseException] = exc + while current is not None and current not in chain: + chain.append(current) + current = current.__cause__ + return chain + + +def _is_syntax_error(exc: BaseException) -> bool: + """True when the text could not be decoded, or the JSON/YAML parser (or its + size/anchor guard) rejected it.""" + import json + + from fluid_build.util.safe_yaml import UnsafeYamlError + + syntax: Tuple[type, ...] = (json.JSONDecodeError, UnsafeYamlError, UnicodeError) + try: + import yaml + + syntax = syntax + (yaml.YAMLError,) + except ImportError: # pragma: no cover - PyYAML is a core dependency + pass + return any(isinstance(e, syntax) for e in _cause_chain(exc)) + + +#: How the loader words its root checks (``_parse_file``, +#: ``parse_contract_text``, ``load_with_overlay``, ``load_overlay_document``), +#: each a plain ``ValueError``. Only these mean ``contract_not_a_mapping``: +#: other plain ``ValueError``s reach the loader too (``Path.resolve`` on a +#: ``$ref`` holding a NUL byte), and are not about the root. +_ROOT_CHECK_MESSAGES: Tuple[str, ...] = ( + "YAML root must be an object/dict", + "Overlay root must be an object/dict", + "contract root must be an object/dict", +) + + +def _is_root_check(exc: BaseException) -> bool: + """True when the loader refused a contract or overlay root that is not an object.""" + return any( + type(e) is ValueError and str(e).startswith(_ROOT_CHECK_MESSAGES) for e in _cause_chain(exc) + ) + + +def _parse_error(exc: BaseException) -> ContractLoadError: + """:func:`loader.parse_contract_text`'s failure as a typed error.""" + if not _is_syntax_error(exc) and _is_root_check(exc): + return ContractLoadError("contract_not_a_mapping", str(exc)) + return ContractLoadError("contract_parse_failed", str(exc)) + + +def _as_load_error(exc: BaseException, path: Path) -> ContractLoadError: + """One :class:`ContractLoadError` for whatever the engine's loader raised.""" + from fluid_build import loader + + event = getattr(exc, "event", None) + if isinstance(event, str) and event: + context = getattr(exc, "context", None) + context = context if isinstance(context, dict) else {} + detail = context.get("error") or context.get("message") or context.get("hint") + return ContractLoadError(event, str(detail or exc), path=path) + if isinstance(exc, FileNotFoundError): + return ContractLoadError("contract_not_found", str(exc), path=path) + if isinstance(exc, loader.RefResolutionError): + return ContractLoadError("contract_ref_unresolved", str(exc), path=path) + if _is_syntax_error(exc): + return ContractLoadError("contract_parse_failed", str(exc), path=path) + if _is_root_check(exc): + return ContractLoadError("contract_not_a_mapping", str(exc), path=path) + return ContractLoadError("contract_load_failed", str(exc), path=path) + + +def _applied_overlay( + contract_path: Path, + env: Optional[str], + contract: Dict[str, Any], + log: logging.Logger, +) -> Optional[Path]: + """The overlay file the engine applied for ``env``, or ``None``. + + The candidate is the file ``load_with_overlay`` selects (same search). + It is not always applied: when the overlay or the merged contract holds + a ``$ref``, the engine's auto-bundle step reloads the contract from the + base file, and the overlay is dropped without a word. Provenance must + not claim a file that did not reach ``contract``, so in exactly that + shape the result is compared with the base alone, and an equal result + reports no overlay (with a WARNING, because ``fluid plan`` is planning + the base contract for that env). + """ + from fluid_build import _contract_loader, loader + + found = loader.load_overlay_document(contract_path, env) + if found is None: + return None + overlay_path, overlay_doc = found + if not ( + _contract_loader._has_ref_pointers(overlay_doc) + or _contract_loader._has_ref_pointers(contract) + ): + return overlay_path.resolve() + base_only = _contract_loader.load_contract_with_overlay(str(contract_path), None, log) + if base_only != contract: + return overlay_path.resolve() + log.warning( + "contract_overlay_not_applied: the engine selected overlay %s for --env %r but " + "planned %s without it, because the overlay or the contract holds a $ref; " + "`fluid plan --env %s` plans the base contract too", + overlay_path, + env, + contract_path, + env, + extra={"event": "contract_overlay_not_applied", "env": env}, + ) + return None + + +def _is_ref_node(node: Any) -> bool: + # The resolver's own test (``loader._is_ref_node``), kept local so the + # walk below reads as one unit. + return isinstance(node, dict) and "$ref" in node and len(node) == 1 + + +def _ref_values(node: Any) -> Tuple[str, ...]: + """Every ``$ref`` value left in ``node``, in document order, deduplicated. + + Iterative, and each container is visited once, so neither depth nor a + YAML alias that makes the document self-referential can stop it short. + """ + out: List[str] = [] + seen: Set[int] = set() + stack: List[Any] = [node] + while stack: + n = stack.pop() + if not isinstance(n, (dict, list)) or id(n) in seen: + continue + seen.add(id(n)) + if isinstance(n, dict): + ref = n.get("$ref") + if isinstance(ref, str) and ref not in out: + out.append(ref) + stack.extend(reversed(list(n.values()))) + else: + stack.extend(reversed(n)) + return tuple(out) + + +def _composed_ref_files(contract_path: Path) -> List[Path]: + """The ``$ref`` target files composed into the contract at ``contract_path``.""" + from fluid_build import loader + + raw = loader.load_contract(contract_path, resolve_refs=False) + return _walk_ref_files(raw, contract_path.parent) + + +def _walk_ref_files(document: Any, base_dir: Path) -> List[Path]: + """Files the engine's ``$ref`` resolver reads for ``document``, first-read order. + + Called only after the resolver succeeded on the same document, so every + reference here is one it accepted; this walk reports, it never decides. + """ + from fluid_build import loader + + found: List[Path] = [] + + def _walk(node: Any, base: Path, ancestry: Set[str], depth: int) -> None: + if depth > _MAX_REF_DEPTH: + return + if _is_ref_node(node): + ref_value = node["$ref"] + if not isinstance(ref_value, str): + return + file_part, pointer = loader._parse_ref(ref_value) + if not file_part: + return # same-document pointer: left in place, reads nothing + target = (base / file_part).resolve() + key = f"{target}#{pointer or ''}" + if key in ancestry: + return + if target not in found: + found.append(target) + subtree = loader.load_contract(target, resolve_refs=False) + if pointer: + subtree = loader._resolve_pointer(subtree, pointer) + _walk(subtree, target.parent, ancestry | {key}, depth + 1) + return + if isinstance(node, Mapping): + for value in node.values(): + _walk(value, base, ancestry, depth) + elif isinstance(node, list): + for item in node: + _walk(item, base, ancestry, depth) + + _walk(document, base_dir, set(), 0) + return found diff --git a/tests/api/test_api_surface_snapshot.py b/tests/api/test_api_surface_snapshot.py index 1ec4fa2a..73a1ea19 100644 --- a/tests/api/test_api_surface_snapshot.py +++ b/tests/api/test_api_surface_snapshot.py @@ -26,7 +26,7 @@ import fluid_build.api as api -EXPECTED_API_VERSION = "1.0" +EXPECTED_API_VERSION = "1.1" EXPECTED_ALL = { "__api_version__", @@ -80,6 +80,13 @@ # security "ImageSignatureVerifier", "SovereigntyChecker", + # contract loading (added in 1.1) + "LoadedContract", + "ContractLoadError", + "ContractOrigin", + "load_contract", + "load_contract_from_text", + "load_contract_from_dict", } diff --git a/tests/api/test_contract_load.py b/tests/api/test_contract_load.py new file mode 100644 index 00000000..0224f621 --- /dev/null +++ b/tests/api/test_contract_load.py @@ -0,0 +1,1116 @@ +# Copyright 2024-2026 Agentics Transformation Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""``fluid_build.api.load_contract*`` returns the contract ``fluid plan`` plans. + +The stability contract of the public loader is one equation, held here for +every rewrite the engine makes (alias values, a legacy ``build:``, ``$ref`` +composition, an environment overlay, a bundle): + + load_contract(path, env=env).contract == plan.json["contract"] + == load_contract_from_text(text, ...).contract + +``plan.json`` comes from the real ``fluid plan`` command ``run()``, so a +change to the engine's loader that this API does not follow fails here. +""" + +from __future__ import annotations + +import argparse +import ast +import builtins +import copy +import inspect +import json +import logging +import os +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional, Tuple + +import pytest + +from fluid_build import _contract_loader, api +from fluid_build._contract_loader import load_contract_with_overlay +from fluid_build.api import ContractLoadError, LoadedContract +from fluid_build.api import contract as contract_api +from fluid_build.cli import bundle as bundle_cmd +from fluid_build.cli import plan as plan_cmd +from fluid_build.forge.core.plan_digest import coerce_keys_to_str, compute_contract_digest +from fluid_build.util.safe_yaml import load_yaml_safe + +pytestmark = [pytest.mark.unit] + +LOG = logging.getLogger("test.api.contract_load") + +# Alias values in three places the alias table covers: build source kind +# (``pg``), build source mode (``incremental``), expose binding format +# (``iceberg_table``). None of them is a schema enum value; all are planned. +_ALIASES = """\ +fluidVersion: "0.7.5" +kind: DataProduct +id: bronze.orders +name: Orders +metadata: + layer: Bronze + owner: {team: dp, email: dp@example.com} +builds: + - id: ingest + pattern: acquisition + engine: kafka-connect + properties: + source: {kind: pg, mode: incremental} + sink: {format: iceberg} +exposes: + - exposeId: orders + kind: table + binding: + platform: aws + format: iceberg_table + location: {database: s, table: o} + contract: + schema: + - {name: id, type: integer, required: true} +""" + +# The legacy singular ``build:`` (schema-valid), which the engine plans as +# ``builds: [build]``. +_LEGACY_BUILD = """\ +fluidVersion: "0.7.5" +kind: DataProduct +id: bronze.orders +name: Orders +metadata: + layer: Bronze + owner: {team: dp, email: dp@example.com} +build: + id: ingest + pattern: acquisition + engine: kafka-connect + properties: + source: {kind: postgres, mode: incremental_append} + sink: {format: iceberg} +exposes: + - exposeId: orders + kind: table + binding: + platform: aws + format: iceberg + location: {database: s, table: o} + contract: + schema: + - {name: id, type: integer, required: true} +""" + +# ``$ref`` composition plus an overlay that patches a field the ``$ref`` +# pulled in, and an alias inside the referenced fragment. +_COMPOSED = """\ +fluidVersion: "0.7.5" +kind: DataProduct +id: bronze.orders +name: Orders +metadata: + layer: Bronze + owner: {team: dp, email: dp@example.com} +build: + id: ingest + pattern: acquisition + engine: kafka-connect + properties: + source: {kind: postgres, mode: incremental_append} + sink: {format: iceberg} +exposes: + - {"$ref": "./parts/orders.yaml"} +""" + +_ORDERS_FRAGMENT = """\ +exposeId: orders +kind: table +binding: + platform: aws + format: iceberg-table + location: {database: s, table: o} +contract: + schema: + - {name: id, type: integer, required: true} +""" + +_PROD_OVERLAY = """\ +name: Orders (prod) +exposes: + - binding: + location: {database: prod_s} +""" + + +def _write(root: Path, files: Dict[str, str]) -> Path: + for rel, text in files.items(): + target = root / rel + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(text, encoding="utf-8") + return root / "contract.fluid.yaml" + + +_CASES: Dict[str, Dict[str, str]] = { + "aliases": {"contract.fluid.yaml": _ALIASES}, + "legacy_build": {"contract.fluid.yaml": _LEGACY_BUILD}, + "composed": { + "contract.fluid.yaml": _COMPOSED, + "parts/orders.yaml": _ORDERS_FRAGMENT, + "overlays/prod.yaml": _PROD_OVERLAY, + }, +} + + +def _parse(register: Callable[[Any], None], argv: List[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(prog="fluid") + parser.add_argument("--provider", default=None) + parser.add_argument("--project", default=None) + parser.add_argument("--region", default=None) + sub = parser.add_subparsers(dest="cmd") + register(sub) + return parser.parse_args(argv) + + +def _plan(src: Path, out: Path, env: Optional[str] = None) -> Dict[str, Any]: + argv = ["plan", str(src), "--out", str(out)] + (["--env", env] if env else []) + assert plan_cmd.run(_parse(plan_cmd.register, argv), LOG) == 0 + return json.loads(out.read_text(encoding="utf-8")) + + +@pytest.fixture +def ws(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + monkeypatch.chdir(tmp_path) + return tmp_path + + +# ── the stability contract: equal to what plan plans ────────────────── + + +@pytest.mark.parametrize( + "case, env", + [ + ("aliases", None), + ("legacy_build", None), + ("composed", None), + ("composed", "prod"), + ], +) +def test_load_contract_equals_the_contract_plan_embeds( + ws: Path, case: str, env: Optional[str] +) -> None: + contract_path = _write(ws / case, _CASES[case]) + planned = _plan(contract_path, ws / "plan.json", env)["contract"] + + loaded = api.load_contract(contract_path, env=env) + + assert loaded.contract == planned + assert loaded.digest == compute_contract_digest(planned) + + +@pytest.mark.parametrize( + "case, env", + [ + ("aliases", None), + ("legacy_build", None), + ("composed", None), + ("composed", "prod"), + ], +) +def test_text_form_equals_the_path_form(ws: Path, case: str, env: Optional[str]) -> None: + contract_path = _write(ws / case, _CASES[case]) + overlay_file = contract_path.parent / "overlays" / f"{env}.yaml" + overlay = load_yaml_safe(overlay_file.read_text("utf-8")) if env else None + + from_text = api.load_contract_from_text( + contract_path.read_text("utf-8"), base_dir=contract_path.parent, overlay=overlay + ) + + assert from_text.contract == api.load_contract(contract_path, env=env).contract + assert from_text.digest == api.load_contract(contract_path, env=env).digest + + +def test_the_rewrites_plan_relies_on_are_applied(ws: Path) -> None: + """The pin above is only meaningful if the fixtures exercise each rewrite.""" + aliases = api.load_contract(_write(ws / "a", _CASES["aliases"])).contract + props = aliases["builds"][0]["properties"] + assert props["source"] == {"kind": "postgres", "mode": "incremental_append"} + assert aliases["exposes"][0]["binding"]["format"] == "iceberg" + + legacy = api.load_contract(_write(ws / "l", _CASES["legacy_build"])).contract + assert "build" not in legacy and legacy["builds"][0]["id"] == "ingest" + + composed = api.load_contract(_write(ws / "c", _CASES["composed"]), env="prod").contract + expose = composed["exposes"][0] + assert composed["name"] == "Orders (prod)" + assert expose["binding"]["format"] == "iceberg" # alias inside the fragment + assert expose["binding"]["location"] == {"database": "prod_s", "table": "o"} + + +def test_alias_under_a_legacy_build_matches_the_engine_loader(ws: Path) -> None: + """The engine rewrites aliases BEFORE promoting ``build:`` to ``builds:``, + so an alias under a legacy ``build:`` is not rewritten. This API mirrors + the engine, order included, in both forms (plan then refuses the value at + the schema gate, as it does on the CLI).""" + text = _LEGACY_BUILD.replace("kind: postgres, mode: incremental_append", "kind: pg") + contract_path = _write(ws, {"contract.fluid.yaml": text}) + engine = load_contract_with_overlay(str(contract_path), None, LOG) + + assert api.load_contract(contract_path).contract == engine + assert api.load_contract_from_text(text).contract == engine + assert engine["builds"][0]["properties"]["source"]["kind"] == "pg" + + +def test_a_bundle_loads_as_plan_loads_it(ws: Path) -> None: + contract_path = _write(ws / "c", _CASES["composed"]) + tgz = ws / "bundle.tgz" + argv = ["bundle", str(contract_path), "--format", "tgz", "--out", str(tgz), "--env", "prod"] + assert bundle_cmd.run(_parse(bundle_cmd.register, argv), LOG) == 0 + planned = _plan(tgz, ws / "plan.json", "prod")["contract"] + + loaded = api.load_contract(tgz, env="prod") + + assert loaded.contract == planned + assert loaded.origin == "bundle" + assert loaded.source == tgz.resolve() + assert loaded.files == (tgz.resolve(),) + assert loaded.overlay is None + assert loaded.contract["name"] == "Orders (prod)" + + +def test_a_bundle_refuses_an_env_it_was_not_built_for(ws: Path) -> None: + contract_path = _write(ws / "c", _CASES["composed"]) + tgz = ws / "bundle.tgz" + argv = ["bundle", str(contract_path), "--format", "tgz", "--out", str(tgz), "--env", "prod"] + assert bundle_cmd.run(_parse(bundle_cmd.register, argv), LOG) == 0 + + with pytest.raises(ContractLoadError) as err: + api.load_contract(tgz, env="staging") + assert err.value.event == "bundle_env_mismatch" + + +def test_overlay_provenance_matches_what_plan_applied(ws: Path, caplog: Any) -> None: + """An overlay holding a ``$ref`` is dropped by the engine's auto-bundle step + today (``fluid plan --env prod`` plans the base). Whatever the engine + does, the contract equals plan's, and ``overlay`` / ``files`` name the + overlay exactly when it reached the contract.""" + files = dict(_CASES["composed"]) + files["overlays/prod.yaml"] = 'name: Orders (prod)\ndescription: {"$ref": "./parts/d.yaml"}\n' + files["parts/d.yaml"] = "text: prod\n" + contract_path = _write(ws, files) + planned = _plan(contract_path, ws / "plan.json", "prod")["contract"] + + with caplog.at_level(logging.WARNING, logger="fluid.api.contract"): + loaded = api.load_contract(contract_path, env="prod") + + overlay_file = contract_path.resolve().parent / "overlays" / "prod.yaml" + applied = loaded.contract["name"] == "Orders (prod)" + assert loaded.contract == planned + assert (loaded.overlay == overlay_file) is applied + assert (overlay_file in loaded.files) is applied + warned = any("contract_overlay_not_applied" in r.getMessage() for r in caplog.records) + assert warned is not applied + + +# ── provenance ────────────────────────────────────────────────────────── + + +def test_provenance_names_every_file_composed(ws: Path) -> None: + contract_path = _write(ws, _CASES["composed"]) + + loaded = api.load_contract(contract_path, env="prod") + + root = contract_path.resolve() + assert loaded.origin == "file" + assert loaded.source == root + assert loaded.env == "prod" + assert loaded.overlay == root.parent / "overlays" / "prod.yaml" + assert loaded.files == ( + root, + root.parent / "parts" / "orders.yaml", + root.parent / "overlays" / "prod.yaml", + ) + assert loaded.unresolved_refs == () + + +def test_provenance_without_env_or_refs(ws: Path) -> None: + contract_path = _write(ws, _CASES["aliases"]) + + loaded = api.load_contract(str(contract_path)) + + assert loaded.overlay is None + assert loaded.env is None + assert loaded.files == (contract_path.resolve(),) + + +def test_nested_and_pointer_refs_are_listed_once_in_read_order(ws: Path) -> None: + contract_path = _write( + ws, + { + "contract.fluid.yaml": ( + "id: x\n" + 'metadata: {"$ref": "./meta.yaml#/inner"}\n' + 'exposes: [{"$ref": "./e.yaml"}, {"$ref": "./e.yaml"}]\n' + ), + "meta.yaml": 'unused: {"$ref": "./never.yaml"}\ninner: {"$ref": "./owner.yaml"}\n', + "owner.yaml": "owner: {team: dp}\n", + "e.yaml": "exposeId: e\n", + }, + ) + + loaded = api.load_contract(contract_path) + + root = contract_path.resolve().parent + assert loaded.contract["metadata"] == {"owner": {"team": "dp"}} + # never.yaml sits outside the pointer's subtree: the resolver never + # reads it (it does not even exist), so it is not provenance. + assert loaded.files == ( + root / "contract.fluid.yaml", + root / "meta.yaml", + root / "owner.yaml", + root / "e.yaml", + ) + + +def test_a_same_document_pointer_is_reported_unresolved(ws: Path) -> None: + contract_path = _write( + ws, + {"contract.fluid.yaml": 'id: x\ndefs: {n: 1}\ndescription: {"$ref": "#/defs/n"}\n'}, + ) + + loaded = api.load_contract(contract_path) + + assert loaded.contract["description"] == {"$ref": "#/defs/n"} + assert loaded.unresolved_refs == ("#/defs/n",) + + +# ── in-memory forms ───────────────────────────────────────────────────── + + +def test_dict_form_without_base_dir_reads_no_file(monkeypatch: pytest.MonkeyPatch) -> None: + document = load_yaml_safe(_COMPOSED) + # Warm every lazy import first: importing a module stats files. + api.load_contract_from_dict(document) + + def _no_fs(*args: Any, **kwargs: Any) -> Any: + raise AssertionError(f"filesystem touched: {args!r}") + + try: + for name in ("stat", "lstat", "open", "listdir", "scandir"): + monkeypatch.setattr(os, name, _no_fs) + monkeypatch.setattr(builtins, "open", _no_fs) + loaded = api.load_contract_from_dict(document) + from_text = api.load_contract_from_text(_COMPOSED) + finally: + # Restore before anything else runs: pytest itself stats files. + monkeypatch.undo() + assert loaded.contract["exposes"] == [{"$ref": "./parts/orders.yaml"}] + assert loaded.unresolved_refs == ("./parts/orders.yaml",) + assert loaded.origin == "memory" + assert loaded.files == () and loaded.source is None + assert from_text.contract == loaded.contract + assert loaded.contract["builds"][0]["id"] == "ingest" # rewrites still applied + + +def test_dict_form_with_base_dir_composes_refs(ws: Path) -> None: + contract_path = _write(ws, _CASES["composed"]) + + loaded = api.load_contract_from_dict(load_yaml_safe(_COMPOSED), base_dir=contract_path.parent) + + assert loaded.contract == api.load_contract(contract_path).contract + assert loaded.files == (contract_path.resolve().parent / "parts" / "orders.yaml",) + assert loaded.unresolved_refs == () + + +def test_inputs_are_never_modified_and_results_are_independent() -> None: + document = load_yaml_safe(_ALIASES) + overlay = {"exposes": [{"binding": {"format": "kafka"}}]} + before_doc, before_overlay = copy.deepcopy(document), copy.deepcopy(overlay) + + first = api.load_contract_from_dict(document, overlay=overlay) + first.contract["exposes"][0]["binding"]["format"] = "mutated" + second = api.load_contract_from_dict(document, overlay=overlay) + + assert document == before_doc and overlay == before_overlay + assert second.contract["exposes"][0]["binding"]["format"] == "kafka_topic" + + +def test_json_text_is_supported() -> None: + document = load_yaml_safe(_ALIASES) + loaded = api.load_contract_from_text(json.dumps(document), suffix=".json") + assert loaded.contract == api.load_contract_from_dict(document).contract + + +def test_loaded_contract_is_frozen() -> None: + loaded = api.load_contract_from_text(_ALIASES) + assert isinstance(loaded, LoadedContract) + with pytest.raises(AttributeError): + loaded.origin = "file" # type: ignore[misc] + + +# ── typed failures ────────────────────────────────────────────────────── + + +def test_missing_file_is_contract_not_found(tmp_path: Path) -> None: + with pytest.raises(ContractLoadError) as err: + api.load_contract(tmp_path / "absent.fluid.yaml") + assert err.value.event == "contract_not_found" + assert err.value.path == (tmp_path / "absent.fluid.yaml").resolve() + assert isinstance(err.value.__cause__, FileNotFoundError) + + +def test_unresolvable_ref_is_contract_ref_unresolved(ws: Path) -> None: + contract_path = _write( + ws, {"contract.fluid.yaml": 'id: x\nexposes: [{"$ref": "./gone.yaml"}]\n'} + ) + with pytest.raises(ContractLoadError) as err: + api.load_contract(contract_path) + assert err.value.event == "contract_ref_unresolved" + + with pytest.raises(ContractLoadError) as err: + api.load_contract_from_text(contract_path.read_text("utf-8"), base_dir=ws) + assert err.value.event == "contract_ref_unresolved" + + +@pytest.mark.parametrize( + "text, suffix, event", + [ + ("id: [unclosed\n", ".yaml", "contract_parse_failed"), + ("{not json", ".json", "contract_parse_failed"), + ("- a\n- b\n", ".yaml", "contract_not_a_mapping"), + ("[1, 2]", ".json", "contract_not_a_mapping"), + ], +) +def test_bad_text_fails_typed(text: str, suffix: str, event: str) -> None: + with pytest.raises(ContractLoadError) as err: + api.load_contract_from_text(text, suffix=suffix) + assert err.value.event == event + + +def test_bad_file_fails_typed(tmp_path: Path) -> None: + broken = tmp_path / "broken.fluid.yaml" + broken.write_text("id: [unclosed\n", encoding="utf-8") + with pytest.raises(ContractLoadError) as err: + api.load_contract(broken) + assert err.value.event == "contract_parse_failed" + + array = tmp_path / "array.json" + array.write_text("[1, 2]", encoding="utf-8") + with pytest.raises(ContractLoadError) as err: + api.load_contract(array) + assert err.value.event == "contract_not_a_mapping" + + +def test_non_mapping_inputs_fail_typed() -> None: + with pytest.raises(ContractLoadError) as err: + api.load_contract_from_dict(["not", "a", "mapping"]) # type: ignore[arg-type] + assert err.value.event == "contract_not_a_mapping" + with pytest.raises(ContractLoadError) as err: + api.load_contract_from_dict({"id": "x"}, overlay=["nope"]) # type: ignore[arg-type] + assert err.value.event == "contract_not_a_mapping" + + +# ── env is a name, never a path ───────────────────────────────────────── + + +@pytest.mark.parametrize( + "shape", + ["absolute", "parent", "empty", "separator", "backslash", "nul", "dot", "dotdot", "drive"], +) +def test_env_must_be_a_single_path_component(ws: Path, shape: str) -> None: + """The engine builds overlay paths from ``env`` (``/.json`` among + them), so an absolute or ``..`` env would merge a file from anywhere into + the returned contract. The API refuses it before reading a file.""" + contract_path = _write(ws / "ws", _CASES["aliases"]) + secret = ws / "elsewhere" / "creds.json" + secret.parent.mkdir() + secret.write_text('{"auths": {"registry": {"auth": "c2VjcmV0"}}}', encoding="utf-8") + env = { + "absolute": str(secret.with_suffix("")), + "parent": "../elsewhere/creds", + "empty": "", + "separator": "prod/eu", + "backslash": "..\\elsewhere\\creds", + "nul": "prod\x00", + "dot": ".", + "dotdot": "..", + # A plain file name on POSIX, a drive-relative path on Windows: refused + # on every platform, so one env never names two different files. + "drive": "C:prod", + }[shape] + + with pytest.raises(ContractLoadError) as err: + api.load_contract(contract_path, env=env) + + assert err.value.event == "contract_env_invalid" + assert err.value.path == contract_path.resolve() + + +# Names ``fluid plan --env`` loads although ``fluid publish --env`` refuses +# them: the API refuses only what makes an env a path, so it loads them too. +_ENVS_PLAN_LOADS = ("prod", "Prod.eu-1_a", "_staging", "prod+eu", "eu prod", "e" * 65) + + +@pytest.mark.parametrize("env", _ENVS_PLAN_LOADS) +def test_every_env_plan_loads_selects_its_overlay(ws: Path, env: str) -> None: + contract_path = _write(ws, _CASES["composed"]) + (contract_path.parent / "overlays" / f"{env}.yaml").write_text(_PROD_OVERLAY, "utf-8") + + loaded = api.load_contract(contract_path, env=env) + + assert loaded.contract == load_contract_with_overlay(str(contract_path), env, LOG) + assert loaded.contract["name"] == "Orders (prod)" + assert loaded.overlay == (contract_path.parent / "overlays" / f"{env}.yaml").resolve() + + +# ── the in-memory overlay follows the engine's auto-bundle step ───────── + +# Two shapes in which ``fluid plan --env prod`` drops the overlay, because a +# ``$ref`` survives the merge and the auto-bundle step reloads the base file. +_OVERLAY_DROPPED: Dict[str, Dict[str, str]] = { + # A same-document pointer in the base contract (in an open block, so the + # contract still plans). + "same_document_pointer": { + "contract.fluid.yaml": _ALIASES + 'extensions: {n: 1, copy: {"$ref": "#/extensions/n"}}\n', + "overlays/prod.yaml": "name: Orders (prod)\n", + }, + # A ``$ref`` inside the overlay itself. + "ref_in_overlay": { + **_CASES["composed"], + "overlays/prod.yaml": 'name: Orders (prod)\ndescription: {"$ref": "./parts/d.yaml"}\n', + "parts/d.yaml": "text: prod\n", + }, +} + + +@pytest.mark.parametrize("shape", sorted(_OVERLAY_DROPPED)) +def test_in_memory_overlay_matches_the_file_form_when_the_engine_drops_it( + ws: Path, shape: str, caplog: Any +) -> None: + contract_path = _write(ws, _OVERLAY_DROPPED[shape]) + planned = _plan(contract_path, ws / "plan.json", "prod")["contract"] + overlay = load_yaml_safe((contract_path.parent / "overlays" / "prod.yaml").read_text("utf-8")) + from_file = api.load_contract(contract_path, env="prod") + + with caplog.at_level(logging.WARNING, logger="fluid.api.contract"): + caplog.clear() + from_text = api.load_contract_from_text( + contract_path.read_text("utf-8"), base_dir=contract_path.parent, overlay=overlay + ) + + assert from_file.contract == planned + assert from_text.contract == from_file.contract + assert from_text.digest == from_file.digest + dropped = from_file.contract["name"] == "Orders" + warned = any("contract_overlay_not_applied" in r.getMessage() for r in caplog.records) + assert warned is dropped + + +def test_in_memory_overlay_without_base_dir_follows_a_same_document_pointer(ws: Path) -> None: + """No file ``$ref``: the engine's decision is knowable without a directory.""" + contract_path = _write(ws, _OVERLAY_DROPPED["same_document_pointer"]) + overlay = load_yaml_safe((contract_path.parent / "overlays" / "prod.yaml").read_text("utf-8")) + + from_text = api.load_contract_from_text(contract_path.read_text("utf-8"), overlay=overlay) + + assert from_text.contract == api.load_contract(contract_path, env="prod").contract + + +def test_in_memory_overlay_without_base_dir_refuses_an_undecidable_document() -> None: + """With file ``$ref`` values unresolved, whether the engine applies the overlay + depends on what the fragments hold; the API says so instead of guessing.""" + with pytest.raises(ContractLoadError) as err: + api.load_contract_from_text(_COMPOSED, overlay=load_yaml_safe(_PROD_OVERLAY)) + assert err.value.event == "contract_overlay_needs_base_dir" + + +def test_in_memory_overlay_logs_to_the_callers_logger() -> None: + records: List[logging.LogRecord] = [] + + class _Collect(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + records.append(record) + + mine = logging.getLogger("test.api.contract_load.caller") + mine.addHandler(_Collect()) + document = {"id": "x", "name": "base", "defs": {"n": 1}, "d": {"$ref": "#/defs/n"}} + + loaded = api.load_contract_from_dict(document, overlay={"name": "prod"}, logger=mine) + + assert loaded.contract["name"] == "base" + assert [getattr(r, "event", None) for r in records] == ["contract_overlay_not_applied"] + + +# ── YAML aliases: shared in the parse, separate in the engine ─────────── + +# The engine's ``$ref`` resolver rebuilds every dict and list of the base +# contract, so a node a YAML alias shares is two objects by the time the +# overlay merge and the rewrites change nodes in place. ``deepcopy`` keeps +# the sharing; without ``base_dir`` the in-memory forms must break it too. +_ALIASED: Dict[str, Dict[str, str]] = { + # The overlay patches one alias of a shared node. + "overlay_merge": { + "contract.fluid.yaml": _ALIASES + "extensions: {a: &x {k: 1}, b: *x}\n", + "overlays/prod.yaml": "extensions: {a: {k: 2}}\n", + }, + # The alias rewrite changes ``exposes[0].binding``; its alias in an open + # block is not a binding, and the engine leaves it as written. + "alias_rewrite": { + "contract.fluid.yaml": _ALIASES.replace(" binding:\n", " binding: &b\n") + + "extensions: {copy: *b}\n", + "overlays/prod.yaml": "name: Orders (prod)\n", + }, +} + + +@pytest.mark.parametrize("shape", sorted(_ALIASED)) +def test_in_memory_form_without_base_dir_unshares_yaml_aliases(ws: Path, shape: str) -> None: + contract_path = _write(ws, _ALIASED[shape]) + text = contract_path.read_text("utf-8") + overlay = load_yaml_safe((contract_path.parent / "overlays" / "prod.yaml").read_text("utf-8")) + planned = _plan(contract_path, ws / "plan.json", "prod")["contract"] + from_file = api.load_contract(contract_path, env="prod") + + from_text = api.load_contract_from_text(text, overlay=overlay) + + assert from_file.contract == planned + assert from_text.contract == from_file.contract + assert from_text.digest == from_file.digest + if shape == "overlay_merge": + assert from_text.contract["extensions"] == {"a": {"k": 2}, "b": {"k": 1}} + else: + assert from_text.contract["exposes"][0]["binding"]["format"] == "iceberg" + assert from_text.contract["extensions"]["copy"]["format"] == "iceberg_table" + # No overlay at all: the rewrite alone must not reach the alias either. + assert api.load_contract_from_text(text).digest == api.load_contract(contract_path).digest + + +def test_aliases_inside_an_overlay_stay_shared_as_in_the_engine(ws: Path) -> None: + """The engine parses an overlay file without the resolver, so sharing *inside* + the overlay survives the merge, and a rewrite reaches every alias of the + node it changes. The in-memory forms keep it the same way. (The base has + no ``exposes``, so the merge places the overlay's own node.)""" + contract_path = _write( + ws, + { + "contract.fluid.yaml": "id: x\nname: Orders\n", + "overlays/prod.yaml": ( + "exposes:\n - exposeId: e\n binding: &b {platform: aws, format: kafka}\n" + "extensions: {copy: *b}\n" + ), + }, + ) + overlay = load_yaml_safe((contract_path.parent / "overlays" / "prod.yaml").read_text("utf-8")) + from_file = api.load_contract(contract_path, env="prod") + + from_text = api.load_contract_from_text(contract_path.read_text("utf-8"), overlay=overlay) + + assert from_file.contract["extensions"]["copy"]["format"] == "kafka_topic" + assert from_text.contract == from_file.contract + assert from_text.digest == from_file.digest + + +def test_a_document_that_contains_itself_fails_as_the_engine_fails(ws: Path) -> None: + contract_path = _write(ws, {"contract.fluid.yaml": _ALIASES + "extensions: &e {self: *e}\n"}) + with pytest.raises(ContractLoadError) as from_file: + api.load_contract(contract_path) + + with pytest.raises(ContractLoadError) as from_text: + api.load_contract_from_text(contract_path.read_text("utf-8")) + + assert from_file.value.event == "contract_load_failed" + assert from_text.value.event == from_file.value.event + + +def test_unsharing_is_iterative_and_keeps_order() -> None: + deep: Dict[str, Any] = {} + node = deep + for _ in range(5000): # far past the interpreter's recursion limit + node["n"] = {} + node = node["n"] + shared = {"z": 1, "a": [1, {"k": "v"}]} + document = {"id": "x", "deep": deep, "p": shared, "q": shared, "r": [shared, shared]} + + copied = contract_api._unshare(document) + + depth, node, original = 0, copied["deep"], deep + while node: # walked, not compared: ``==`` itself recurses + assert node is not original and list(node) == ["n"] + depth, node, original = depth + 1, node["n"], original["n"] + assert depth == 5000 + assert {k: v for k, v in copied.items() if k != "deep"} == { + k: v for k, v in document.items() if k != "deep" + } + assert list(copied) == list(document) and list(copied["p"]) == ["z", "a"] + containers = [copied["p"], copied["q"], copied["r"][0], copied["r"][1]] + assert len({id(c) for c in containers}) == 4 + assert len({id(c["a"]) for c in containers}) == 4 + + +# ── the in-memory replay cannot fall behind the engine loader ─────────── + + +_ENGINE_SOURCE = inspect.getsource(_contract_loader.load_contract_with_overlay) + + +def _is_contract(node: ast.AST) -> bool: + return isinstance(node, ast.Name) and node.id == "contract" + + +def _rooted_at_contract(node: ast.AST) -> bool: + while isinstance(node, (ast.Subscript, ast.Attribute)): + node = node.value + return _is_contract(node) + + +def _callee(node: ast.Call) -> str: + func = node.func + if isinstance(func, ast.Attribute): + return func.attr + return func.id if isinstance(func, ast.Name) else ast.dump(func) + + +def _engine_post_load_effects(source: str = _ENGINE_SOURCE) -> Tuple[List[str], List[str]]: + """What ``load_contract_with_overlay`` does to ``contract``, in source order, + outside its bundle branch (the in-memory forms never load a bundle). + + Returns ``(steps, writes)``. ``steps``: every call handed ``contract`` or a + part of it (``contract["builds"]``), positionally or by keyword. + ``writes``: every statement that changes ``contract``, as the callee whose + result is assigned to it, or as ``<...>`` for any other change (an + assignment from a non-call, an item or attribute write, an augmented + assignment, ``del``, a method call on it, a rebinding as a loop or tuple + target), any return of something other than ``contract`` itself, and any + other read of ``contract``: an alias, a part of it passed on, a loop over + it, a container or expression holding it. A read is accounted for only as + a whole-``contract`` argument of a call (listed in ``steps``) or as the + returned value; anything else could change the contract unseen, so it is + reported, and the guard fails closed. + """ + tree = ast.parse(source) + function = tree.body[0] + assert isinstance(function, ast.FunctionDef) + parents = { + child: parent for parent in ast.walk(function) for child in ast.iter_child_nodes(parent) + } + steps: List[str] = [] + writes: List[str] = [] + + def _argument(node: ast.AST) -> ast.AST: + return node.value if isinstance(node, ast.Starred) else node + + def _record_write(target: ast.AST, value: Optional[ast.AST]) -> None: + if _is_contract(target): + writes.append(_callee(value) if isinstance(value, ast.Call) else "") + elif _rooted_at_contract(target): + writes.append("") + + class _Effects(ast.NodeVisitor): + def visit_If(self, node: ast.If) -> None: + test = node.test + if isinstance(test, ast.Call) and getattr(test.func, "id", "") == "_is_bundle_path": + return + self.generic_visit(node) + + def visit_Call(self, node: ast.Call) -> None: + self.generic_visit(node) # inner calls first: they run first + func = node.func + if isinstance(func, ast.Attribute) and _rooted_at_contract(func.value): + writes.append(f"") + if any(_rooted_at_contract(_argument(a)) for a in node.args) or any( + _rooted_at_contract(k.value) for k in node.keywords + ): + steps.append(_callee(node)) + + def visit_Assign(self, node: ast.Assign) -> None: + self.generic_visit(node) + for target in node.targets: + _record_write(target, node.value) + + def visit_AnnAssign(self, node: ast.AnnAssign) -> None: + self.generic_visit(node) + _record_write(node.target, node.value) + + def visit_NamedExpr(self, node: ast.NamedExpr) -> None: + self.generic_visit(node) + _record_write(node.target, node.value) + + def visit_AugAssign(self, node: ast.AugAssign) -> None: + self.generic_visit(node) + if _rooted_at_contract(node.target): + writes.append("") + + def visit_Delete(self, node: ast.Delete) -> None: + self.generic_visit(node) + if any(_rooted_at_contract(t) for t in node.targets): + writes.append("") + + def visit_Return(self, node: ast.Return) -> None: + self.generic_visit(node) + if node.value is None or not _is_contract(node.value): + writes.append("") + + def visit_Name(self, node: ast.Name) -> None: + if node.id != "contract": + return + parent = parents[node] + if isinstance(node.ctx, ast.Store): + # A plain or annotated assignment, a walrus or an augmented + # assignment is recorded above; any other binding is not. + if not isinstance( + parent, (ast.Assign, ast.AnnAssign, ast.NamedExpr, ast.AugAssign) + ): + writes.append("") + return + if isinstance(node.ctx, ast.Del): + return # recorded by ``visit_Delete`` + whole_argument = (isinstance(parent, ast.Call) and node in parent.args) or isinstance( + parent, ast.keyword + ) + if whole_argument or isinstance(parent, ast.Return): + return + writes.append("") + + _Effects().visit(function) + return steps, writes + + +# The engine loads (``load_with_overlay``, or ``load_contract`` for a loader +# without it), then runs the auto-bundle step and the rewrites. +_EXPECTED_STEPS = ["_auto_bundle_if_needed", *contract_api._ENGINE_REWRITES] +_EXPECTED_WRITES = ["load_with_overlay", "load_contract", *_EXPECTED_STEPS] + + +def test_in_memory_forms_replay_every_engine_loader_step() -> None: + """The file form calls the engine loader; the in-memory forms replay its steps + (``_replay_overlay`` for the auto-bundle decision, then + ``_ENGINE_REWRITES`` by name). A step added to, removed from or moved in + the engine fails here until the in-memory forms follow it.""" + assert _engine_post_load_effects() == (_EXPECTED_STEPS, _EXPECTED_WRITES) + for name in contract_api._ENGINE_REWRITES: + assert callable(getattr(_contract_loader, name)) + + +@pytest.mark.parametrize( + "added", + [ + "contract = _normalize_new(contract)", + "_normalize_builds_in_place(contract['builds'])", + "_normalize_x(builds=contract['builds'])", + "_normalize_x(*contract['builds'])", + "c = contract\n c['x'] = 1", + "c = contract['builds']\n c.append({})", + "for b in contract['builds']:\n b['x'] = 1", + "[b.update(x=1) for b in contract['builds']]", + "_mutate_all([contract])", + "for contract in [{}]:\n pass", + "contract, _ = {}, None", + "if contract.get('x'):\n pass", + "contract = resolve_contract_env_templates(value=contract)", + "_mutate_in_place(contract=contract)", + "contract = {**contract, 'x': 1}", + "contract = dict(contract, x=1)", + "contract['x'] = 1", + "contract['a']['b'] = 1", + "contract.update(x=1)", + "contract['a'].setdefault('b', 1)", + "contract |= {'x': 1}", + "del contract['x']", + "_ = (contract := {})", + "contract: dict = {}", + ], +) +def test_the_step_guard_sees_every_way_the_engine_can_change_the_contract(added: str) -> None: + """Negative control for the guard above: the engine's real source, with one + more change to ``contract`` before it returns, no longer matches.""" + head, sep, tail = _ENGINE_SOURCE.rpartition(" return contract\n") + assert sep, "the engine loader no longer ends with `return contract`" + mutated = f"{head} {added}\n{sep}{tail}" + + assert _engine_post_load_effects(mutated) != (_EXPECTED_STEPS, _EXPECTED_WRITES) + + +@pytest.mark.parametrize( + "returned", + ["{**contract, 'x': 1}", "dict(contract, x=1)", "_normalize_new(contract)", "None", ""], +) +def test_the_step_guard_sees_a_changed_return(returned: str) -> None: + """Negative control: the engine's real source returning anything but + ``contract`` itself no longer matches.""" + head, sep, tail = _ENGINE_SOURCE.rpartition(" return contract\n") + assert sep, "the engine loader no longer ends with `return contract`" + mutated = f"{head} return {returned}".rstrip() + f"\n{tail}" + + assert _engine_post_load_effects(mutated) != (_EXPECTED_STEPS, _EXPECTED_WRITES) + + +# ── typed failures, file form ─────────────────────────────────────────── + + +def test_a_yaml_list_root_file_is_contract_not_a_mapping(tmp_path: Path) -> None: + """The same input gives the same event through the file and text forms.""" + listed = tmp_path / "list.fluid.yaml" + listed.write_text("- a\n- b\n", encoding="utf-8") + + with pytest.raises(ContractLoadError) as err: + api.load_contract(listed) + + assert err.value.event == "contract_not_a_mapping" + with pytest.raises(ContractLoadError) as text_err: + api.load_contract_from_text(listed.read_text("utf-8")) + assert text_err.value.event == err.value.event + + +def test_a_list_root_overlay_is_contract_not_a_mapping(ws: Path) -> None: + contract_path = _write(ws, {**_CASES["aliases"], "overlays/prod.yaml": "- a\n"}) + with pytest.raises(ContractLoadError) as err: + api.load_contract(contract_path, env="prod") + assert err.value.event == "contract_not_a_mapping" + + +@pytest.mark.parametrize( + "root", ['[{"a": 1}]', '[{"a": 1, "b": 2}]', '[["k", "v"]]', "[]", "1", "null", "ref"] +) +def test_a_json_root_that_is_not_an_object_is_contract_not_a_mapping_with_or_without_an_overlay( + tmp_path: Path, root: str +) -> None: + """The loader checks a YAML root but not a JSON one (nor one a root ``$ref`` + composes), and merges an overlay into ``dict(base)``: a list root then + loads as a dict, or fails with a message that names no root. One file, + one event, whether or not an overlay exists for the env.""" + contract_path = tmp_path / "c.json" + if root == "ref": + (tmp_path / "frag.json").write_text('[{"a": 1}]', encoding="utf-8") + root = '{"$ref": "./frag.json"}' + contract_path.write_text(root, encoding="utf-8") + (tmp_path / "overlays").mkdir() + (tmp_path / "overlays" / "prod.yaml").write_text("x: 1\n", encoding="utf-8") + + for env in (None, "prod", "staging"): # no env, an overlay, no overlay + with pytest.raises(ContractLoadError) as err: + api.load_contract(contract_path, env=env) + assert (err.value.event, err.value.path) == ( + "contract_not_a_mapping", + contract_path.resolve(), + ), env + + +def test_with_an_env_a_base_that_fails_to_load_keeps_its_own_event(tmp_path: Path) -> None: + """The root check before an overlay merge reads the base first; a base that + cannot be read still fails with the event the engine's load gives it.""" + (tmp_path / "overlays").mkdir() + (tmp_path / "overlays" / "prod.yaml").write_text("x: 1\n", encoding="utf-8") + broken = tmp_path / "broken.json" + broken.write_text("[1,", encoding="utf-8") + dangling = tmp_path / "dangling.json" + dangling.write_text('{"id": "x", "m": {"$ref": "./absent.yaml"}}', encoding="utf-8") + + events = [] + for contract_path in (tmp_path / "absent.json", broken, dangling): + with pytest.raises(ContractLoadError) as err: + api.load_contract(contract_path, env="prod") + events.append(err.value.event) + + assert events == ["contract_not_found", "contract_parse_failed", "contract_ref_unresolved"] + + +def test_a_nul_byte_in_a_ref_is_a_ref_error_not_a_root_check( + tmp_path: Path, +) -> None: + """A ``$ref`` holding a NUL byte cannot be resolved as a path. The ref + confinement check reports that as a typed ref error, through both forms, + and it is never ``contract_not_a_mapping``: only the loader's root checks + mean that.""" + from fluid_build.loader import RefResolutionError + + contract_path = tmp_path / "contract.fluid.yaml" + contract_path.write_text('id: x\nname: base\nmeta: {"$ref": "./a\\0b.yaml"}\n', "utf-8") + + with pytest.raises(ContractLoadError) as from_file: + api.load_contract(contract_path) + with pytest.raises(ContractLoadError) as from_dict: + api.load_contract_from_dict( + {"id": "x", "meta": {"$ref": "./a\x00b.yaml"}}, base_dir=tmp_path + ) + + for raised in (from_file.value, from_dict.value): + assert raised.event == "contract_ref_unresolved" + assert raised.event != "contract_not_a_mapping" + assert isinstance(raised.__cause__, RefResolutionError) + + +def test_a_path_no_file_can_have_is_contract_not_found(tmp_path: Path) -> None: + with pytest.raises(ContractLoadError) as from_path: + api.load_contract(f"{tmp_path}/c\x00.fluid.yaml") + with pytest.raises(ContractLoadError) as from_base_dir: + api.load_contract_from_dict({"id": "x"}, base_dir=f"{tmp_path}/d\x00") + + for err in (from_path, from_base_dir): + assert err.value.event == "contract_not_found" + assert err.value.path is None + assert isinstance(err.value.__cause__, ValueError) + + +def test_a_file_that_is_not_utf8_is_contract_parse_failed(tmp_path: Path) -> None: + """``UnicodeDecodeError`` is a ``ValueError``; it is still a parse failure.""" + binary = tmp_path / "binary.fluid.yaml" + binary.write_bytes(b"id: \xff\xfe\n") + with pytest.raises(ContractLoadError) as err: + api.load_contract(binary) + assert err.value.event == "contract_parse_failed" + + +def test_a_missing_bundle_is_the_engines_bundle_not_found(tmp_path: Path) -> None: + with pytest.raises(ContractLoadError) as err: + api.load_contract(tmp_path / "absent.tgz") + assert err.value.event == "bundle_not_found" + + +# ── keys and the digest ───────────────────────────────────────────────── + + +def test_a_magic_word_key_equals_plan_once_keys_are_strings(ws: Path) -> None: + """``plan.json`` writes keys as strings; ``contract`` keeps the engine's + ``bool`` key. The documented equation holds after that coercion, and the + digest agrees without it.""" + contract_path = _write(ws, {"contract.fluid.yaml": _ALIASES + "extensions: {on: x}\n"}) + planned = _plan(contract_path, ws / "plan.json")["contract"] + + loaded = api.load_contract(contract_path) + + assert loaded.contract["extensions"] == {True: "x"} + assert coerce_keys_to_str(loaded.contract) == planned + assert loaded.digest == compute_contract_digest(planned) + assert api.load_contract_from_text(_ALIASES + "extensions: {on: x}\n").digest == loaded.digest + + +def test_digest_of_an_unserialisable_contract_fails_typed(ws: Path) -> None: + contract_path = _write( + ws, {"contract.fluid.yaml": _ALIASES + "extensions: {created: 2024-01-01}\n"} + ) + loaded = api.load_contract(contract_path) # loading is fine; plan is not + + with pytest.raises(ContractLoadError) as err: + _ = loaded.digest + + assert err.value.event == "contract_not_serialisable" + assert err.value.path == contract_path.resolve() + assert isinstance(err.value.__cause__, TypeError) + + +def test_digest_is_the_planned_contracts_not_the_raw_files(ws: Path) -> None: + """Documented: ``.digest`` hashes the normalised contract, so it differs + from ``fluid contract digest`` (the raw parse) whenever a rewrite applies.""" + contract_path = _write(ws, _CASES["aliases"]) + raw = compute_contract_digest(load_yaml_safe(contract_path.read_text("utf-8"))) + assert api.load_contract(contract_path).digest != raw