From 481166c1307edaf15624f20ced9c0909f4dba2cb Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 19 Oct 2026 16:42:27 +0200 Subject: [PATCH 1/4] feat(loader): overlays patch builds, exposes and consumes by identity from 0.7.6 _deep_merge patched lists of maps by position, so inserting or reordering an entry 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; the patch also copied id: transform onto it, and `fluid plan --env prod` failed with planner_failed 'schedule_transform' and no word of why. In a contract whose fluidVersion is 0.7.6 or later (the overlay's own fluidVersion counts), builds[].id, exposes[].exposeId and consumes[] by productId + exposeId match an overlay entry to its base entry when the base's entries carry unique ones, the way Kubernetes server-side apply merges a list by its list-map keys and jsonmerge's arrayMergeById (MIT) merges by id; the semantics are borrowed, not the dependency. An overlay entry that matches nothing is appended and logged as overlay_entry_added, naming the base's ids so a typo shows. An overlay entry the merge cannot match, no id or an id named twice, raises OverlayMergeError with its index: falling back to position would retarget it, the failure this fixes. load_with_overlay and `fluid bundle --env` report it with the overlay's path, and the contract API as contract_load_failed. Id-less patches still apply to a base with at most one entry. A base whose own ids are missing or repeated keeps the positional merge and logs overlay_merged_by_position. Older contracts merge as before and log overlay_patch_retargeted when a patch lands on an entry that names another id, or another part of a consumes key. Each warning names its contract and is logged once per process, since apply loads its contract twice; note_missing_overlay dedupes the same way. Only the three top-level lists are keyed; lists inside an entry, and a nested key named builds, still merge by position. The rule sits in _deep_merge, so every command that applies an overlay, `fluid bundle --env` and the contract API merge the same way. --- fluid_build/api/contract.py | 17 +- fluid_build/cli/bundle.py | 12 +- fluid_build/loader.py | 248 +++++++++++++++++++++-- tests/test_overlay_keyed_merge.py | 316 ++++++++++++++++++++++++++++++ 4 files changed, 570 insertions(+), 23 deletions(-) create mode 100644 tests/test_overlay_keyed_merge.py 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..d58053d1 --- /dev/null +++ b/tests/test_overlay_keyed_merge.py @@ -0,0 +1,316 @@ +# 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 + +import fluid_build.loader as loader +from fluid_build.loader import OverlayMergeError, _deep_merge + + +@pytest.fixture(autouse=True) +def _fresh_warnings(): + # Each merge warning is logged once per process; every test starts afresh. + loader._NOTED_OVERLAY_MERGE_WARNINGS.clear() + yield + loader._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", + ] From 2b147ebbef63a9a844bc3834acb5ae890abf8666 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 19 Oct 2026 16:42:27 +0200 Subject: [PATCH 2/4] docs(changelog): overlays merge builds, exposes and consumes by identity (schema 0.7.6) An Upgrade note for 0.7.6 contracts whose overlay names ids at other positions, adds entries, or leaves an id out, and Added entries for the keyed merge and the retarget warning older contracts now get. --- CHANGELOG.md | 32 ++++++++++++++++++++++++++++++++ 1 file changed, 32 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0762fb90..76f48e26 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,38 @@ 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. + +### Added + +- **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 From f6e0d33e2930e173123d511d0afcc608f6693969 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 19 Oct 2026 17:16:42 +0200 Subject: [PATCH 3/4] test(loader): import the warning set by name, one import style for loader The overlay tests imported fluid_build.loader both as a module and by name (code-quality review on #732). --- tests/test_overlay_keyed_merge.py | 133 ++++++++++++++++++++++-------- 1 file changed, 100 insertions(+), 33 deletions(-) diff --git a/tests/test_overlay_keyed_merge.py b/tests/test_overlay_keyed_merge.py index d58053d1..8769887f 100644 --- a/tests/test_overlay_keyed_merge.py +++ b/tests/test_overlay_keyed_merge.py @@ -33,16 +33,19 @@ import pytest import yaml -import fluid_build.loader as loader -from fluid_build.loader import OverlayMergeError, _deep_merge +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. - loader._NOTED_OVERLAY_MERGE_WARNINGS.clear() + _NOTED_OVERLAY_MERGE_WARNINGS.clear() yield - loader._NOTED_OVERLAY_MERGE_WARNINGS.clear() + _NOTED_OVERLAY_MERGE_WARNINGS.clear() #: The base contract's builds, with ``backfill`` inserted before the two the @@ -66,7 +69,9 @@ def _merge(base: Dict[str, Any], overlay: Dict[str, Any]) -> Dict[str, Any]: def _events(caplog: pytest.LogCaptureFixture) -> list: - return [getattr(r, "event", None) for r in caplog.records if r.name == "fluid.loader"] + return [ + getattr(r, "event", None) for r in caplog.records if r.name == "fluid.loader" + ] class TestKeyedMerge: @@ -86,9 +91,16 @@ def test_an_overlay_entry_matching_nothing_is_appended_and_named( ) -> 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]]} + 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 [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() @@ -96,10 +108,14 @@ def test_an_overlay_entry_matching_nothing_is_appended_and_named( @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} + 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: + 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} @@ -112,7 +128,9 @@ def test_exposes_match_by_expose_id(self) -> None: {"exposeId": "b", "binding": {"location": {"topic": "b.dev"}}}, ], } - overlay = {"exposes": [{"exposeId": "b", "binding": {"location": {"topic": "b.prod"}}}]} + 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" @@ -125,23 +143,34 @@ def test_consumes_match_on_product_and_expose_together(self) -> None: {"productId": "p", "exposeId": "y", "versionConstraint": "^1.0"}, ], } - overlay = {"consumes": [{"productId": "p", "exposeId": "y", "versionConstraint": "^2.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} + 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": "a", + "contract": {"schema": [{"name": "id"}, {"name": "v"}]}, + }, {"exposeId": "b"}, ], } - overlay = {"exposes": [{"exposeId": "a", "contract": {"schema": [{"type": "string"}]}}]} + overlay = { + "exposes": [{"exposeId": "a", "contract": {"schema": [{"type": "string"}]}}] + } schema = _merge(base, overlay)["exposes"][0]["contract"]["schema"] assert schema == [{"name": "id", "type": "string"}, {"name": "v"}] @@ -158,15 +187,18 @@ def test_before_0_7_6_a_reordered_base_retargets_the_patch_and_warns( # ...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 + 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"}]} + overlay = { + "builds": [{"id": "backfill"}, {"id": "ingest", "engine": "debezium"}] + } builds = _merge(_base("0.7.5"), overlay)["builds"] assert builds[1]["engine"] == "debezium" assert _events(caplog) == [] @@ -175,7 +207,9 @@ def test_a_patch_in_the_base_order_does_not_warn( "overlay, problem", [ pytest.param( - {"builds": [{"properties": {"executors": 8}}]}, "builds[0] names no id", id="no-id" + {"builds": [{"properties": {"executors": 8}}]}, + "builds[0] names no id", + id="no-id", ), pytest.param( {"builds": [{"id": "transform"}, {"id": "transform"}]}, @@ -195,13 +229,21 @@ 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"}]} + 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() + 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: + 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"}]}) @@ -227,13 +269,18 @@ def test_a_retarget_on_part_of_a_compound_key_warns( caplog.set_level(logging.WARNING, logger="fluid.loader") base = { "fluidVersion": "0.7.5", - "consumes": [{"productId": "a", "exposeId": "e"}, {"productId": "b", "exposeId": "e"}], + "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: + 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) @@ -241,7 +288,9 @@ def test_each_warning_is_logged_once(self, caplog: pytest.LogCaptureFixture) -> assert _events(caplog) == ["overlay_patch_retargeted"] -@pytest.mark.parametrize("version, executors", [("0.7.6", [2, None, 8]), ("0.7.5", [8, None, 2])]) +@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: @@ -249,22 +298,32 @@ def test_load_contract_with_overlay_applies_the_rule( 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 / "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") + (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) + 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 / "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"): + 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") @@ -280,13 +339,19 @@ def test_the_contract_api_reports_an_unmatched_entry_as_a_load_error() -> None: 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 / "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")) + 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( @@ -303,7 +368,9 @@ def test_a_warning_names_its_contract_and_each_contract_gets_its_own( assert "in p2, the overlay's builds[0]" in messages[1] -def test_an_added_entry_names_its_contract_too(caplog: pytest.LogCaptureFixture) -> None: +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") From ccefe0548833f78edc46463ff4d66f0508e10975 Mon Sep 17 00:00:00 2001 From: Speculator55005 <50082482+fas89@users.noreply.github.com> Date: Mon, 19 Oct 2026 17:20:55 +0200 Subject: [PATCH 4/4] style(tests): black 24.10.0 at the repo's line length --- tests/test_overlay_keyed_merge.py | 92 ++++++++----------------------- 1 file changed, 23 insertions(+), 69 deletions(-) diff --git a/tests/test_overlay_keyed_merge.py b/tests/test_overlay_keyed_merge.py index 8769887f..81531d66 100644 --- a/tests/test_overlay_keyed_merge.py +++ b/tests/test_overlay_keyed_merge.py @@ -69,9 +69,7 @@ def _merge(base: Dict[str, Any], overlay: Dict[str, Any]) -> Dict[str, Any]: def _events(caplog: pytest.LogCaptureFixture) -> list: - return [ - getattr(r, "event", None) for r in caplog.records if r.name == "fluid.loader" - ] + return [getattr(r, "event", None) for r in caplog.records if r.name == "fluid.loader"] class TestKeyedMerge: @@ -91,9 +89,7 @@ def test_an_overlay_entry_matching_nothing_is_appended_and_named( ) -> 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]] - } + 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", @@ -108,14 +104,10 @@ def test_an_overlay_entry_matching_nothing_is_appended_and_named( @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 - } + 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: + 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} @@ -128,9 +120,7 @@ def test_exposes_match_by_expose_id(self) -> None: {"exposeId": "b", "binding": {"location": {"topic": "b.dev"}}}, ], } - overlay = { - "exposes": [{"exposeId": "b", "binding": {"location": {"topic": "b.prod"}}}] - } + 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" @@ -143,19 +133,13 @@ def test_consumes_match_on_product_and_expose_together(self) -> None: {"productId": "p", "exposeId": "y", "versionConstraint": "^1.0"}, ], } - overlay = { - "consumes": [ - {"productId": "p", "exposeId": "y", "versionConstraint": "^2.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 - } + 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 = { @@ -168,9 +152,7 @@ def test_lists_inside_a_matched_entry_still_merge_by_position(self) -> None: {"exposeId": "b"}, ], } - overlay = { - "exposes": [{"exposeId": "a", "contract": {"schema": [{"type": "string"}]}}] - } + overlay = {"exposes": [{"exposeId": "a", "contract": {"schema": [{"type": "string"}]}}]} schema = _merge(base, overlay)["exposes"][0]["contract"]["schema"] assert schema == [{"name": "id", "type": "string"}, {"name": "v"}] @@ -187,18 +169,15 @@ def test_before_0_7_6_a_reordered_base_retargets_the_patch_and_warns( # ...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) + 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"}] - } + overlay = {"builds": [{"id": "backfill"}, {"id": "ingest", "engine": "debezium"}]} builds = _merge(_base("0.7.5"), overlay)["builds"] assert builds[1]["engine"] == "debezium" assert _events(caplog) == [] @@ -236,14 +215,9 @@ def test_0_7_6_merges_by_position_when_the_base_has_no_unique_ids( 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() - ) + 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: + 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"}]}) @@ -278,9 +252,7 @@ def test_a_retarget_on_part_of_a_compound_key_warns( 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: + 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) @@ -288,9 +260,7 @@ def test_each_warning_is_logged_once( assert _events(caplog) == ["overlay_patch_retargeted"] -@pytest.mark.parametrize( - "version, executors", [("0.7.6", [2, None, 8]), ("0.7.5", [8, None, 2])] -) +@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: @@ -298,32 +268,22 @@ def test_load_contract_with_overlay_applies_the_rule( 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 / "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" - ) + (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) + 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 / "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" - ): + 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") @@ -339,19 +299,13 @@ def test_the_contract_api_reports_an_unmatched_entry_as_a_load_error() -> None: 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 / "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") - ) + 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(