diff --git a/CHANGELOG.md b/CHANGELOG.md index 19ea4172..495cd4c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 as its `id`. A bundle that relied on a file named after the port's `contractId` but declaring another id fails a strict import, or skips the port under `--lenient` (an input port is always skipped); make the port's `contractId` and the ODCS `id` agree. +- In a contract whose `fluidVersion` is 0.7.6 (the schema 0.7.6 preview) or later, an + overlay patches `builds`, `exposes` and `consumes` by identity, not by position. An + overlay that names each entry's id merges as before when it lists them in the base's + order; one that names an id at another position now patches the entry with that id, and + an entry whose id the base lacks is appended and logged as `overlay_entry_added`. When + the base's entries carry unique ids, an overlay entry with no id, or two naming the same + id, now fails the load (`contract_load_failed`): give each the id of the entry it + patches. Id-less patches still apply to a base with at most one entry, as long as no + overlay entry for it names an id. A 0.7.5 or older contract still merges by position. - A build with `properties.language: flink_sql` is no longer run as SQL of forge-cli's own. `fluid apply --mode amend-and-build` ran its Flink SQL on DuckDB, or refused an unknown embedded-SQL platform, and the local, AWS, GCP and Snowflake planners planned it as a @@ -50,6 +59,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 to schedule and exits 0), and `fluid generate transformation` says there is no code to generate for it. Deployment settings (parallelism, upgrade mode, Flink configuration) go under `properties.parameters`. 0.7.5 is unchanged. +- **Overlays merge `builds`, `exposes` and `consumes` by identity (schema 0.7.6 preview).** + Lists of maps merged by position, so inserting or reordering a build in the base silently + retargeted every overlay: with `builds: [ingest, transform]` and `overlays/prod.yaml` + raising `builds[1]`, adding a `backfill` build first made prod patch `ingest`, copying + `id: transform` onto it, and `fluid plan --env prod` failed with `planner_failed: + 'schedule_transform'`. From `fluidVersion` 0.7.6, `builds[].id`, `exposes[].exposeId` + and `consumes[]`'s `productId` + `exposeId` match an overlay entry to its base entry, as + Kubernetes server-side apply matches list entries by their list-map keys. An entry that + matches nothing is appended and logged as `overlay_entry_added`, which names the base's + ids, so a typo shows. An overlay entry the merge cannot match (no id, or an id named + twice) fails the load with the index, where falling back to position would retarget it; + a base whose own ids are missing or repeated keeps the positional merge and logs + `overlay_merged_by_position`. Lists inside an entry, such as an expose's schema, still + merge by position. Every command that applies an overlay, and the contract API, merges + this way. +- **Older contracts warn when a positional patch lands on another entry.** Before 0.7.6, + an overlay entry that names an id (or part of a `consumes` key) other than the base + entry at its position logs `overlay_patch_retargeted`, naming both, and the merge is + unchanged. Each merge warning names its contract and is logged once per process, so a + command that loads the contract twice says it once. ### Fixed diff --git a/fluid_build/api/contract.py b/fluid_build/api/contract.py index 2240a49e..98b0d5e8 100644 --- a/fluid_build/api/contract.py +++ b/fluid_build/api/contract.py @@ -327,8 +327,10 @@ def load_contract_from_dict( 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 + with the engine's merge (dicts key by key; lists of objects by + position, or ``builds``/``exposes``/``consumes`` by identity from + fluidVersion 0.7.6; 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 @@ -377,9 +379,14 @@ def load_contract_from_dict( # 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 - ) + try: + contract = _replay_overlay( + contract, copy.deepcopy(dict(overlay)), composed=base_dir is not None, log=log + ) + except loader.OverlayMergeError as exc: + # An overlay entry the merge cannot match: the same failure + # ``load_contract(path, env=...)`` reports. + raise ContractLoadError("contract_load_failed", f"overlay: {exc}") from exc # ``_contract_loader.load_contract_with_overlay``'s rewrites, in its order. for rewrite in _ENGINE_REWRITES: contract = getattr(_contract_loader, rewrite)(contract) diff --git a/fluid_build/cli/bundle.py b/fluid_build/cli/bundle.py index 14138de8..24b539ce 100644 --- a/fluid_build/cli/bundle.py +++ b/fluid_build/cli/bundle.py @@ -220,12 +220,20 @@ def materialize_contract( compiled = compile_contract(contract_path, logger=compile_logger or logger) overlay_path: Optional[Path] = None if env: - from ..loader import _deep_merge, load_overlay_document, note_missing_overlay + from ..loader import ( + OverlayMergeError, + _deep_merge, + load_overlay_document, + note_missing_overlay, + ) found = load_overlay_document(contract_path, env) if found is not None: overlay_path, overlay = found - compiled = _deep_merge(dict(compiled), overlay) + try: + compiled = _deep_merge(dict(compiled), overlay) + except OverlayMergeError as exc: + raise OverlayMergeError(f"{overlay_path}: {exc}") from exc logger.info("overlay_applied", extra={"overlay": str(overlay_path)}) else: note_missing_overlay(contract_path, env, logger, contract=compiled) diff --git a/fluid_build/loader.py b/fluid_build/loader.py index 40468b02..2cc227b2 100644 --- a/fluid_build/loader.py +++ b/fluid_build/loader.py @@ -164,19 +164,46 @@ def parse_contract_text(text: str, *, suffix: str = ".yaml") -> Dict[str, Any]: raise RuntimeError(f"Failed to parse contract text: {e}") from e -def _deep_merge(base: Dict[str, Any], overlay: Dict[str, Any]) -> Dict[str, Any]: +#: Top-level contract lists whose items an overlay patches by identity rather +#: than by position, from fluidVersion 0.7.6 on. Kubernetes server-side apply +#: merges a list the same way (``x-kubernetes-list-map-keys``, compound keys +#: such as ``containerPort`` + ``protocol``), as does jsonmerge's +#: ``arrayMergeById``: an entry matches on every key field, and an overlay entry +#: that matches none is appended. +_OVERLAY_IDENTITY_KEYS: Dict[str, Tuple[str, ...]] = { + "builds": ("id",), + "exposes": ("exposeId",), + "consumes": ("productId", "exposeId"), +} +_KEYED_OVERLAY_MERGE_SINCE = (0, 7, 6) + + +class OverlayMergeError(ValueError): + """An overlay entry cannot be matched to the base entry it patches.""" + + +def _deep_merge( + base: Dict[str, Any], overlay: Dict[str, Any], *, _root: bool = True +) -> Dict[str, Any]: """ Recursively merge overlay into base. - Dicts (both sides): merged key-by-key recursively. - - List of dicts (both sides): merged positionally so overlay entries act as - partial patches — each overlay item is deep-merged into the corresponding - base item by index. Extra overlay items beyond the base length are appended. + - List of dicts (both sides): each overlay entry is a partial patch, + deep-merged into a base entry. ``builds``, ``exposes`` and ``consumes`` + at the top of a contract whose fluidVersion is 0.7.6 or later are matched + by identity (``builds[].id``, ``exposes[].exposeId``, ``consumes[]`` by + ``productId`` + ``exposeId``) when the base entries carry unique ones; an + overlay entry that matches nothing is appended. Any other list of dicts + is merged by position, and extra overlay entries are appended. See + :func:`_merge_dict_list`. - Scalar lists / mismatched types: overlay value replaces base value entirely. """ + keyed = _root and _keyed_overlay_merge(base, overlay) + contract_id = str(base.get("id") or "the contract") if _root else "" for k, v in overlay.items(): if k in base and isinstance(base[k], dict) and isinstance(v, dict): - base[k] = _deep_merge(base[k], v) + base[k] = _deep_merge(base[k], v, _root=False) elif ( k in base and isinstance(base[k], list) @@ -184,22 +211,211 @@ def _deep_merge(base: Dict[str, Any], overlay: Dict[str, Any]) -> Dict[str, Any] and v and all(isinstance(item, dict) for item in v) ): - # Positional merge for list-of-dicts: patch each overlay entry into - # the corresponding base entry, preserving unmentioned fields. - merged: List[Any] = list(base[k]) - for i, overlay_item in enumerate(v): - if i < len(merged) and isinstance(merged[i], dict): - merged[i] = _deep_merge(dict(merged[i]), overlay_item) - elif i < len(merged): - merged[i] = overlay_item - else: - merged.append(overlay_item) - base[k] = merged + identity = _OVERLAY_IDENTITY_KEYS.get(k) if _root else None + base[k] = _merge_dict_list( + k, base[k], v, identity, keyed=keyed, contract_id=contract_id + ) else: base[k] = v return base +def _keyed_overlay_merge(base: Mapping[str, Any], overlay: Mapping[str, Any]) -> bool: + """True when the merged contract's fluidVersion is 0.7.6 or later.""" + version = overlay.get("fluidVersion", base.get("fluidVersion")) + parts = str(version or "").strip().split(".") + if not all(part.isascii() and part.isdigit() and len(part) < 10 for part in parts): + return False + return tuple(int(part) for part in parts) >= _KEYED_OVERLAY_MERGE_SINCE + + +def _merge_dict_list( + name: str, + base_items: List[Any], + overlay_items: List[Dict[str, Any]], + identity: Optional[Tuple[str, ...]], + *, + keyed: bool, + contract_id: str = "", +) -> List[Any]: + """Patch ``overlay_items`` into ``base_items``, by identity or by position. + + Merging by position retargets an overlay as soon as the base gains, + loses or reorders an entry, without any error. From 0.7.6 a list with an + identity key merges by it when the base's entries carry unique ones, and + an overlay entry that names none, or names one twice, is refused: falling + back to position would retarget it. Before 0.7.6, or when the base's ids + are missing or repeated, the merge is by position and warns when a patch + lands on an entry that names another id. + """ + base_ids = [_item_identity(item, identity) for item in base_items] if identity else [] + overlay_ids = [_item_identity(item, identity) for item in overlay_items] if identity else [] + if identity and keyed and _unique_ids(base_ids): + if len(base_items) > 1 or any(overlay_ids): + return _merge_by_identity( + name, identity, base_items, overlay_items, base_ids, overlay_ids, contract_id + ) + # At most one base entry and id-less patches: nothing else they could mean. + + if identity and len(base_items) > 1: + _warn_positional_overlay( + name, identity, base_items, overlay_items, keyed=keyed, contract_id=contract_id + ) + merged = list(base_items) + for i, overlay_item in enumerate(overlay_items): + if i < len(merged) and isinstance(merged[i], dict): + merged[i] = _deep_merge(dict(merged[i]), overlay_item, _root=False) + elif i < len(merged): + merged[i] = overlay_item + else: + merged.append(overlay_item) + return merged + + +def _merge_by_identity( + name: str, + keys: Tuple[str, ...], + base_items: List[Any], + overlay_items: List[Dict[str, Any]], + base_ids: List[Optional[Tuple[str, ...]]], + overlay_ids: List[Optional[Tuple[str, ...]]], + contract_id: str = "", +) -> List[Any]: + key_label = " + ".join(keys) + seen: Dict[Tuple[str, ...], int] = {} + for i, ident in enumerate(overlay_ids): + if ident is None: + raise OverlayMergeError( + f"the overlay's {name}[{i}] names no {key_label}. In a contract whose " + f"fluidVersion is 0.7.6 or later, every overlay entry of {name} names the " + f"{key_label} of the entry it patches, or of a new entry to add." + ) + if ident in seen: + raise OverlayMergeError( + f"the overlay's {name}[{seen[ident]}] and {name}[{i}] both name " + f"{key_label} {'/'.join(ident)}; patch each entry once." + ) + seen[ident] = i + merged: List[Any] = list(base_items) + position = {ident: i for i, ident in enumerate(base_ids)} + for ident, overlay_item in zip(overlay_ids, overlay_items, strict=True): + i = position.get(ident) + if i is None: + # A new entry, or a patch whose id has a typo: say which. + _warn_once( + "overlay_entry_added", + name, + None, + "overlay_entry_added: the overlay for %s adds %s %s %s, which the base does " + "not have (the base has %s). If it was meant to patch one of those, fix its %s.", + contract_id, + name, + key_label, + "/".join(ident), + ", ".join("/".join(b) for b in base_ids if b) or "none", + key_label, + ) + merged.append(overlay_item) + else: + merged[i] = _deep_merge(dict(merged[i]), overlay_item, _root=False) + return merged + + +def _item_identity(item: Any, keys: Tuple[str, ...]) -> Optional[Tuple[str, ...]]: + if not isinstance(item, Mapping): + return None + values = tuple(item.get(key) for key in keys) + if all(isinstance(value, str) and value for value in values): + return values + return None + + +def _unique_ids(ids: List[Optional[Tuple[str, ...]]]) -> bool: + return all(ids) and len(set(ids)) == len(ids) + + +def _warn_positional_overlay( + name: str, + keys: Tuple[str, ...], + base_items: List[Any], + overlay_items: List[Dict[str, Any]], + *, + keyed: bool, + contract_id: str = "", +) -> None: + key_label = " + ".join(keys) + remedy = ( + f"give every base entry of {name} a unique {key_label}" + if keyed + else f"set fluidVersion 0.7.6 or later to merge {name} by {key_label}" + ) + # Only the positions both lists have: an overlay entry past the base's end is appended. + for i, (overlay_item, base_item) in enumerate(zip(overlay_items, base_items, strict=False)): + overlay_id = _key_fields(overlay_item, keys) + base_id = _key_fields(base_item, keys) + # A key field both entries name, with different values: the patch is + # for another entry, even when the overlay names only part of the key. + if any(k in overlay_id and k in base_id and overlay_id[k] != base_id[k] for k in keys): + _warn_once( + "overlay_patch_retargeted", + name, + i, + "overlay_patch_retargeted: in %s, the overlay's %s[%d] names %s %s but " + "patches the base's %s[%d], %s %s, because lists merge by position. Put " + "the overlay's entries in the base's order, or %s.", + contract_id, + name, + i, + " + ".join(overlay_id), + "/".join(overlay_id.values()), + name, + i, + " + ".join(base_id), + "/".join(base_id.values()), + remedy, + ) + if keyed: + _warn_once( + "overlay_merged_by_position", + name, + None, + "overlay_merged_by_position: in %s, %s is merged by position, not by %s, " + "because not every base entry carries a unique %s; %s.", + contract_id, + name, + key_label, + key_label, + remedy, + ) + + +def _key_fields(item: Any, keys: Tuple[str, ...]) -> Dict[str, str]: + """The identity fields ``item`` names, of the ones in ``keys``.""" + if not isinstance(item, Mapping): + return {} + return {k: item[k] for k in keys if isinstance(item.get(k), str) and item[k]} + + +#: Overlay merge warnings already logged by this process, each naming its +#: contract: a command that loads its contract twice (``fluid apply`` does) +#: says each once, as :func:`note_missing_overlay` does for a missing overlay. +#: Tests reset it with ``_NOTED_OVERLAY_MERGE_WARNINGS.clear()``. +_NOTED_OVERLAY_MERGE_WARNINGS: Set[str] = set() +_NOTED_OVERLAY_MERGE_WARNINGS_LOCK = threading.Lock() + + +def _warn_once(event: str, name: str, index: Optional[int], message: str, *args: Any) -> None: + text = message % args + with _NOTED_OVERLAY_MERGE_WARNINGS_LOCK: + if text in _NOTED_OVERLAY_MERGE_WARNINGS: + return + _NOTED_OVERLAY_MERGE_WARNINGS.add(text) + extra: Dict[str, Any] = {"event": event, "list": name} + if index is not None: + extra["index"] = index + LOG.warning("%s", text, extra=extra) + + # --------------------------------------------------------------------------- # $ref resolution — multi-file contract composition # --------------------------------------------------------------------------- diff --git a/tests/test_overlay_keyed_merge.py b/tests/test_overlay_keyed_merge.py new file mode 100644 index 00000000..81531d66 --- /dev/null +++ b/tests/test_overlay_keyed_merge.py @@ -0,0 +1,337 @@ +# 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. + +"""Overlays patch builds, exposes and consumes by identity from fluidVersion 0.7.6. + +Merging a list of maps by position retargets every overlay as soon as the base +gains, loses or reorders an entry: a ``backfill`` build inserted first made +``overlays/prod.yaml``'s patch for ``transform`` land on ``ingest``, silently. +From 0.7.6 the three top-level lists merge by their identity key; older +contracts keep the positional merge, and warn when a patch lands on an entry +that names another id. +""" + +from __future__ import annotations + +import copy +import logging +import re +from pathlib import Path +from typing import Any, Dict + +import pytest +import yaml + +from fluid_build.loader import ( + _NOTED_OVERLAY_MERGE_WARNINGS, + OverlayMergeError, + _deep_merge, +) + + +@pytest.fixture(autouse=True) +def _fresh_warnings(): + # Each merge warning is logged once per process; every test starts afresh. + _NOTED_OVERLAY_MERGE_WARNINGS.clear() + yield + _NOTED_OVERLAY_MERGE_WARNINGS.clear() + + +#: The base contract's builds, with ``backfill`` inserted before the two the +#: prod overlay was written against. +BUILDS = [ + {"id": "backfill", "engine": "spark"}, + {"id": "ingest", "engine": "kafka-connect"}, + {"id": "transform", "engine": "spark", "properties": {"executors": 2}}, +] + + +def _base(version: str) -> Dict[str, Any]: + return {"fluidVersion": version, "builds": copy.deepcopy(BUILDS)} + + +PROD = {"builds": [{"id": "transform", "properties": {"executors": 8}}]} + + +def _merge(base: Dict[str, Any], overlay: Dict[str, Any]) -> Dict[str, Any]: + return _deep_merge(copy.deepcopy(base), copy.deepcopy(overlay)) + + +def _events(caplog: pytest.LogCaptureFixture) -> list: + return [getattr(r, "event", None) for r in caplog.records if r.name == "fluid.loader"] + + +class TestKeyedMerge: + def test_a_patch_follows_its_build_when_the_base_is_reordered( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level(logging.WARNING, logger="fluid.loader") + builds = _merge(_base("0.7.6"), PROD)["builds"] + assert [b["id"] for b in builds] == ["backfill", "ingest", "transform"] + assert builds[2]["properties"] == {"executors": 8} + assert "properties" not in builds[0] + assert builds[2]["engine"] == "spark" # unmentioned fields are kept + assert _events(caplog) == [] + + def test_an_overlay_entry_matching_nothing_is_appended_and_named( + self, caplog: pytest.LogCaptureFixture + ) -> None: + # A new entry, or a patch whose id has a typo: either way, say so. + caplog.set_level(logging.WARNING, logger="fluid.loader") + overlay = {"builds": [{"id": "transfrom", "engine": "spark"}, PROD["builds"][0]]} + builds = _merge(_base("0.7.6"), overlay)["builds"] + assert [b["id"] for b in builds] == [ + "backfill", + "ingest", + "transform", + "transfrom", + ] + assert builds[2]["properties"] == {"executors": 8} + assert _events(caplog) == ["overlay_entry_added"] + assert "adds builds id transfrom" in caplog.records[0].getMessage() + assert "backfill, ingest, transform" in caplog.records[0].getMessage() + + @pytest.mark.parametrize("version", ["0.7.10", "0.8", "1.0"]) + def test_later_versions_merge_by_identity(self, version: str) -> None: + assert _merge(_base(version), PROD)["builds"][2]["properties"] == {"executors": 8} + + @pytest.mark.parametrize("version", ["0.7.6-rc1", "0.7.\u00b2", None]) + def test_a_version_that_is_not_plain_numbers_merges_by_position(self, version) -> None: + base = _base("0.7.6") + base["fluidVersion"] = version + assert _merge(base, PROD)["builds"][0]["properties"] == {"executors": 8} + + def test_exposes_match_by_expose_id(self) -> None: + base = { + "fluidVersion": "0.7.6", + "exposes": [ + {"exposeId": "a", "binding": {"location": {"topic": "a.dev"}}}, + {"exposeId": "b", "binding": {"location": {"topic": "b.dev"}}}, + ], + } + overlay = {"exposes": [{"exposeId": "b", "binding": {"location": {"topic": "b.prod"}}}]} + exposes = _merge(base, overlay)["exposes"] + assert exposes[0]["binding"]["location"]["topic"] == "a.dev" + assert exposes[1]["binding"]["location"]["topic"] == "b.prod" + + def test_consumes_match_on_product_and_expose_together(self) -> None: + base = { + "fluidVersion": "0.7.6", + "consumes": [ + {"productId": "p", "exposeId": "x", "versionConstraint": "^1.0"}, + {"productId": "p", "exposeId": "y", "versionConstraint": "^1.0"}, + ], + } + overlay = {"consumes": [{"productId": "p", "exposeId": "y", "versionConstraint": "^2.0"}]} + consumes = _merge(base, overlay)["consumes"] + assert [c["versionConstraint"] for c in consumes] == ["^1.0", "^2.0"] + + def test_an_overlay_can_opt_in_by_raising_the_version(self) -> None: + overlay = {"fluidVersion": "0.7.6", **PROD} + assert _merge(_base("0.7.5"), overlay)["builds"][2]["properties"] == {"executors": 8} + + def test_lists_inside_a_matched_entry_still_merge_by_position(self) -> None: + base = { + "fluidVersion": "0.7.6", + "exposes": [ + { + "exposeId": "a", + "contract": {"schema": [{"name": "id"}, {"name": "v"}]}, + }, + {"exposeId": "b"}, + ], + } + overlay = {"exposes": [{"exposeId": "a", "contract": {"schema": [{"type": "string"}]}}]} + schema = _merge(base, overlay)["exposes"][0]["contract"]["schema"] + assert schema == [{"name": "id", "type": "string"}, {"name": "v"}] + + +class TestPositionalMerge: + def test_before_0_7_6_a_reordered_base_retargets_the_patch_and_warns( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level(logging.WARNING, logger="fluid.loader") + builds = _merge(_base("0.7.5"), PROD)["builds"] + # Unchanged behaviour: the patch lands on index 0, keeping its id... + assert builds[0]["properties"] == {"executors": 8} + assert builds[2]["properties"] == {"executors": 2} + # ...and now says so. + assert _events(caplog) == ["overlay_patch_retargeted"] + message = caplog.records[0].getMessage() + assert "builds[0] names id transform but patches the base's builds[0], id backfill" in ( + message + ) + + def test_a_patch_in_the_base_order_does_not_warn( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level(logging.WARNING, logger="fluid.loader") + overlay = {"builds": [{"id": "backfill"}, {"id": "ingest", "engine": "debezium"}]} + builds = _merge(_base("0.7.5"), overlay)["builds"] + assert builds[1]["engine"] == "debezium" + assert _events(caplog) == [] + + @pytest.mark.parametrize( + "overlay, problem", + [ + pytest.param( + {"builds": [{"properties": {"executors": 8}}]}, + "builds[0] names no id", + id="no-id", + ), + pytest.param( + {"builds": [{"id": "transform"}, {"id": "transform"}]}, + "builds[0] and builds[1] both name id transform", + id="repeated-id", + ), + ], + ) + def test_0_7_6_refuses_an_overlay_entry_it_cannot_match( + self, overlay: Dict[str, Any], problem: str + ) -> None: + # Falling back to position would retarget it, the failure this fixes. + with pytest.raises(OverlayMergeError, match=re.escape(problem)): + _merge(_base("0.7.6"), overlay) + + def test_0_7_6_merges_by_position_when_the_base_has_no_unique_ids( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level(logging.WARNING, logger="fluid.loader") + base = { + "fluidVersion": "0.7.6", + "builds": [{"engine": "sql"}, {"engine": "dbt"}], + } + merged = _merge(base, {"builds": [{"engine": "spark"}]}) + assert [b["engine"] for b in merged["builds"]] == ["spark", "dbt"] + assert _events(caplog) == ["overlay_merged_by_position"] + assert "give every base entry of builds a unique id" in caplog.records[0].getMessage() + + def test_a_single_entry_is_not_ambiguous(self, caplog: pytest.LogCaptureFixture) -> None: + caplog.set_level(logging.WARNING, logger="fluid.loader") + base = {"fluidVersion": "0.7.6", "builds": [{"id": "only", "engine": "sql"}]} + merged = _merge(base, {"builds": [{"engine": "dbt"}]}) + assert merged["builds"] == [{"id": "only", "engine": "dbt"}] + assert _events(caplog) == [] + + def test_a_nested_key_named_builds_is_not_an_identity_list( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level(logging.WARNING, logger="fluid.loader") + base = { + "fluidVersion": "0.7.6", + "extensions": {"x": {"builds": [{"id": "a", "v": 1}, {"id": "b", "v": 1}]}}, + } + overlay = {"extensions": {"x": {"builds": [{"id": "b", "v": 2}]}}} + nested = _merge(base, overlay)["extensions"]["x"]["builds"] + assert nested == [{"id": "b", "v": 2}, {"id": "b", "v": 1}] + assert _events(caplog) == [] + + def test_a_retarget_on_part_of_a_compound_key_warns( + self, caplog: pytest.LogCaptureFixture + ) -> None: + caplog.set_level(logging.WARNING, logger="fluid.loader") + base = { + "fluidVersion": "0.7.5", + "consumes": [ + {"productId": "a", "exposeId": "e"}, + {"productId": "b", "exposeId": "e"}, + ], + } + _merge(base, {"consumes": [{"productId": "b", "purpose": "x"}]}) + assert _events(caplog) == ["overlay_patch_retargeted"] + assert "names productId b but patches" in caplog.records[0].getMessage() + + def test_each_warning_is_logged_once(self, caplog: pytest.LogCaptureFixture) -> None: + # `fluid apply` loads the contract twice. + caplog.set_level(logging.WARNING, logger="fluid.loader") + _merge(_base("0.7.5"), PROD) + _merge(_base("0.7.5"), PROD) + assert _events(caplog) == ["overlay_patch_retargeted"] + + +@pytest.mark.parametrize("version, executors", [("0.7.6", [2, None, 8]), ("0.7.5", [8, None, 2])]) +def test_load_contract_with_overlay_applies_the_rule( + tmp_path: Path, version: str, executors: list +) -> None: + from fluid_build.loader import load_with_overlay + + base = _base(version) + base["builds"][0]["properties"] = {"executors": 2} + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(base), encoding="utf-8") + (tmp_path / "overlays").mkdir() + (tmp_path / "overlays" / "prod.yaml").write_text(yaml.safe_dump(PROD), encoding="utf-8") + contract = load_with_overlay(tmp_path / "contract.fluid.yaml", env="prod") + assert [(b.get("properties") or {}).get("executors") for b in contract["builds"]] == (executors) + + +def test_load_with_overlay_reports_an_entry_it_cannot_match(tmp_path: Path) -> None: + from fluid_build.loader import load_with_overlay + + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(_base("0.7.6")), encoding="utf-8") + (tmp_path / "overlays").mkdir() + (tmp_path / "overlays" / "prod.yaml").write_text( + yaml.safe_dump({"builds": [{"properties": {"executors": 8}}]}), encoding="utf-8" + ) + with pytest.raises(RuntimeError, match=r"prod\.yaml: the overlay's builds\[0\] names no id"): + load_with_overlay(tmp_path / "contract.fluid.yaml", env="prod") + + +def test_the_contract_api_reports_an_unmatched_entry_as_a_load_error() -> None: + from fluid_build.api.contract import ContractLoadError, load_contract_from_dict + + with pytest.raises(ContractLoadError) as info: + load_contract_from_dict(_base("0.7.6"), overlay={"builds": [{"engine": "dbt"}]}) + assert info.value.event == "contract_load_failed" + assert "builds[0] names no id" in str(info.value) + + +def test_bundle_names_the_overlay_it_cannot_apply(tmp_path: Path) -> None: + from fluid_build.cli.bundle import materialize_contract + + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(_base("0.7.6")), encoding="utf-8") + (tmp_path / "overlays").mkdir() + (tmp_path / "overlays" / "prod.yaml").write_text( + yaml.safe_dump({"builds": [{"engine": "dbt"}]}), encoding="utf-8" + ) + with pytest.raises(OverlayMergeError, match=r"overlays/prod\.yaml: the overlay's builds\[0\]"): + materialize_contract(tmp_path / "contract.fluid.yaml", "prod", logging.getLogger("t")) + + +def test_a_warning_names_its_contract_and_each_contract_gets_its_own( + caplog: pytest.LogCaptureFixture, +) -> None: + caplog.set_level(logging.WARNING, logger="fluid.loader") + for product in ("p1", "p2"): + base = _base("0.7.5") + base["id"] = product + _merge(base, PROD) + messages = [r.getMessage() for r in caplog.records if r.name == "fluid.loader"] + assert len(messages) == 2 + assert "in p1, the overlay's builds[0]" in messages[0] + assert "in p2, the overlay's builds[0]" in messages[1] + + +def test_an_added_entry_names_its_contract_too( + caplog: pytest.LogCaptureFixture, +) -> None: + caplog.set_level(logging.WARNING, logger="fluid.loader") + for product in ("p1", "p2"): + base = _base("0.7.6") + base["id"] = product + _merge(base, {"builds": [{"id": "reindex", "engine": "spark"}]}) + messages = [r.getMessage() for r in caplog.records if r.name == "fluid.loader"] + assert [m.split(" adds ")[0] for m in messages] == [ + "overlay_entry_added: the overlay for p1", + "overlay_entry_added: the overlay for p2", + ]