diff --git a/AGENTS.md b/AGENTS.md index e24e0ca37..66971f58d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -193,7 +193,7 @@ new extra installed). `uv` is dev-time only — not a runtime dep. - **Adding a provider**: Create `fluid_build/providers//`, implement `BaseProvider` interface - **Adding a template**: Create `fluid_build/templates/.j2` or add to `fluid_build/forge/templates/` - **Adding a policy rule**: Extend validators in `fluid_build/policy/` -- **Modifying the contract schema**: Update `fluid_build/schemas/` and bump `fluidVersion`. Never hardcode the new version as a literal — `SchemaManager.latest_bundled_version()` scans the schemas directory and returns the newest, so fallbacks track future bumps automatically. +- **Modifying the contract schema**: Update `fluid_build/schemas/` and bump `fluidVersion`. Never hardcode the new version as a literal — `SchemaManager.latest_bundled_version()` scans the schemas directory and returns the newest, so fallbacks track future bumps automatically. A `description` on a `not: {required: [field]}` rule is the error message `fluid validate` prints for it (`schema_manager.format_schema_error`), so write it as one: what is refused and what to do instead. - **Data-product type vocabulary** (v0.7.3+): contracts can use medallion (`metadata.layer ∈ {Bronze, Silver, Gold, Platinum}`) or Data Mesh (`metadata.productType ∈ {SDP, ADP, CDP}`) — they're equivalent. Canonical mapping: **Bronze↔SDP** (Source-Aligned Data Product), **Silver↔ADP** (Aggregated), **Gold↔CDP** (Consumption-Aligned). Either-or-both is accepted; when both are set the validator enforces consistency. Cross-check is duplicated inline at three call sites (`fluid_build/schema.py::_check_metadata`, `fluid_build/schema_manager.py::_validate_with_jsonschema`, `fluid_build/cli/contract_validation.py::_validate_metadata`) — keep them in lockstep when changing the mapping. Provider tag/label propagation in `providers/{aws,gcp,snowflake}` and `forge/core/provider_actions.py` emits both as separate cloud labels (`fluid_layer` + `fluid_product_type`). - **Adding a new apply mode**: extend `fluid_build/forge/core/apply_modes.py::ApplyMode` enum + `CANONICAL_CHOICES` list + predicates (`is_destructive`, `needs_build`, etc.) + `check_data_loss_gate` if the mode is destructive. The mode must map to a provider-native DDL flavor in `providers//actions/*.py`. - **Changing what `fluid plan` emits**: `plan.json` must carry both `planDigest` and `bundleDigest` (re-run `inject_digests` after any mutation to the dict). Actions must carry BOTH `op` and `action_type` fields — apply.py's provider dispatcher reads `op`, display/viz reads `action_type`. Dropping either silently breaks the pipeline. diff --git a/CHANGELOG.md b/CHANGELOG.md index 99c7f209a..b5643c62a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,51 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Upgrade notes + +- `fluid validate --env --strict` (on one contract or a whole workspace), `fluid + contract-validation --env --strict` and `fluid test --env --strict` (the + native engine; `--engine soda` does not read `--strict`) now fail when an overlay patch + lands on an entry that names another id (`overlay_patch_retargeted`), which an overlay + merged by position can do: a 0.7.5 or older contract, or a 0.7.6 base whose entries lack + unique ids. Without `--strict` these commands list it among their warnings (`fluid test` + as an "Environment overlay" row, `fluid validate --quiet` as a `WARNING:` line), and the + loader no longer also logs it there. Put the overlay's entries in the base's order, or + set `fluidVersion: "0.7.6"` with a unique id on every base entry, so overlays patch by + id. `overlay_entry_added` stays a log line and does not fail `--strict`. `fluid validate + .tgz` does not see the warning: `fluid bundle --env` merged the overlay before, + so validate the contract file with `--env`. +- `fluid validate --strict` now says `Validation failed: warnings treated as errors + (--strict)` when warnings alone fail it, after a summary that reads valid; with + `--quiet` it prints each such warning as `ERROR (--strict): `. +- A product whose every build becomes `continuous` (or runs outside forge-cli) gets no + `schedule/` artifacts, so pipeline stage 11 skips `fluid schedule-sync`, and a DAG + deployed for it earlier stays in the scheduler: remove that DAG by hand. +- An environment overlay cannot turn a scheduled build continuous: overlays cannot + delete the base's `cron` or `schedule`, which a continuous trigger refuses. Declare the + trigger in each environment's overlay instead of in the base. + +### Added + +- **`trigger.type: continuous` (0.7.6 preview).** A build whose job its platform keeps + running, such as an `engine: flink` job or a `kafka-connect` or `debezium` connector, + can say so: `execution.trigger: {type: continuous}`. No scheduler starts it: `fluid + generate schedule`, `fluid schedule-sync` and the `fluid apply` DAGs leave the build out, + where a connector used to get a nightly `0 2 * * *` task. A contract with no other build + has nothing to schedule. The schema refuses `cron`, `schedule`, `datasets`, `event` and + `timetable` next to it, as Databricks Jobs refuses a continuous trigger next to any + other, and the error names the field and the trigger type that takes it. It is not + Airflow's `@continuous`, which starts batch runs back to back. 0.7.5 is unchanged. + Declare it on a build that runs outside forge-cli or on a connector: the dbt and python + runners refuse it ("Unknown trigger type"), as they refuse `event` and `dataset`, and + `fluid scaffold-composer` still writes its nightly DAG for any contract. + +### Changed + +- A schema error from a `not: {required: [...]}` rule now reads ` is not allowed + here`, or the rule's own `description`, instead of quoting the whole object and the + subschema it must not match. + ## [0.21.0] — 2026-10-19 ### Upgrade notes diff --git a/docs/schedule.md b/docs/schedule.md index 59a5e6a3c..2d5540e10 100644 --- a/docs/schedule.md +++ b/docs/schedule.md @@ -38,6 +38,9 @@ hand-declares `orchestration.tasks` (those keep their declared operators). A build is left out when its own `execution.orchestration.engine` is `none` or not Airflow, or when its trigger is `event`, `dataset`, `schedule_and_dataset` or `timetable`, which a plain cron DAG cannot express. +A build whose trigger is `continuous` (schema 0.7.6), or that runs outside +forge-cli (`engine: flink`, `properties.language: flink_sql`), gets no DAG and no task +at all: its platform keeps the job running. Layout, one directory per product: diff --git a/fluid_build/cli/_acquisition_stage_ext.py b/fluid_build/cli/_acquisition_stage_ext.py index 208038539..936ea5343 100644 --- a/fluid_build/cli/_acquisition_stage_ext.py +++ b/fluid_build/cli/_acquisition_stage_ext.py @@ -59,6 +59,7 @@ from fluid_build.build_runners._ids import ( validate_identifier as _validate_identifier, ) +from fluid_build.util.contract import is_unscheduled_build LOG = logging.getLogger("fluid.acquire.stage_ext") @@ -481,6 +482,10 @@ def schedule_sync_acquisition( out: List[ScheduleArtifact] = [] for build in acquisition_builds(contract): bid = _validate_identifier(build.get("id", ""), kind="build.id") + # A continuous connector keeps running on its own: a schedule beside it + # is refused by schema 0.7.6, and must not start a second copy here. + if is_unscheduled_build(build): + continue execution = build.get("execution", {}) or {} trigger = execution.get("trigger", {}) or {} schedule = trigger.get("schedule") diff --git a/fluid_build/cli/contract_validation.py b/fluid_build/cli/contract_validation.py index 2525e081d..5bd0deaa7 100644 --- a/fluid_build/cli/contract_validation.py +++ b/fluid_build/cli/contract_validation.py @@ -292,11 +292,16 @@ def validate(self) -> ValidationReport: """Execute comprehensive contract validation.""" start_time = time.time() - # Load contract + # Load contract. A positional overlay patch that lands on another + # entry is only logged by the loader; collect it, so it is reported + # as a warning below and ``--strict`` fails on it. + from fluid_build.loader import collect_overlay_warnings + try: - self.contract = load_contract_with_overlay( - str(self.contract_path), self.env, self.logger - ) + with collect_overlay_warnings() as overlay_warnings: + self.contract = load_contract_with_overlay( + str(self.contract_path), self.env, self.logger + ) except FileNotFoundError: from .core import ContractNotFoundError @@ -330,6 +335,9 @@ def validate(self) -> ValidationReport: duration=0.0, data_checks_performed=self.check_data, ) + # The message already names the fix. + for warning in overlay_warnings: + self.report.add_issue("warning", "overlay", warning.message) # Step 1: Validate contract syntax against schema self._validate_contract_schema() @@ -1625,6 +1633,8 @@ def output_rich_report(report: ValidationReport, output_file: Optional[str] = No # Issues table if report.issues: + from rich.markup import escape + table = Table(title="Validation Issues", show_header=True) table.add_column("Severity", style="bold") table.add_column("Category") @@ -1638,7 +1648,11 @@ def output_rich_report(report: ValidationReport, output_file: Optional[str] = No "info": "[blue]INFO[/blue]", }.get(issue.severity, issue.severity.upper()) - table.add_row(severity_style, issue.category, issue.message, issue.path) + # Messages quote contract text (an overlay warning names the + # contract's ids), which Rich would parse as markup. + table.add_row( + severity_style, issue.category, escape(issue.message), escape(issue.path or "") + ) console.print(table) else: diff --git a/fluid_build/cli/generate_schedule.py b/fluid_build/cli/generate_schedule.py index 8a72412fe..e027a71c9 100644 --- a/fluid_build/cli/generate_schedule.py +++ b/fluid_build/cli/generate_schedule.py @@ -50,12 +50,13 @@ SUBCOMMAND = "schedule" -def _only_externally_run_builds(contract: Dict[str, Any]) -> bool: - """True when the contract has builds and every one runs outside forge-cli.""" - from fluid_build.util.contract import is_externally_run_build +def _only_unscheduled_builds(contract: Dict[str, Any]) -> bool: + """True when the contract has builds and no scheduler starts any of them: + each runs outside forge-cli, or continuously.""" + from fluid_build.util.contract import is_unscheduled_build builds = [b for b in contract.get("builds") or [] if isinstance(b, dict)] - return bool(builds) and all(is_externally_run_build(b) for b in builds) + return bool(builds) and all(is_unscheduled_build(b) for b in builds) def register_subcommand(subparsers: argparse._SubParsersAction): @@ -165,16 +166,18 @@ def run(args: Any, logger: logging.Logger) -> int: # the operator names, so defaulting WHO renders it commits nothing. scheduler_name = fluid_apply.DEFAULT_ENGINE - # A contract whose every build runs on its own platform (a Flink job) - # has nothing to schedule, which is not an error: unless it hand-declares - # orchestration tasks, or names a scheduler that does not exist. - if _only_externally_run_builds(contract) and not fluid_apply.has_explicit_tasks(contract): + # A contract whose every build runs on its own platform (a Flink job, or + # a build whose trigger is continuous) has nothing to schedule, which is + # not an error: unless it hand-declares orchestration tasks, or names a + # scheduler that does not exist. + if _only_unscheduled_builds(contract) and not fluid_apply.has_explicit_tasks(contract): from fluid_build.schedulers import has_scheduler if not scheduler_name or has_scheduler(scheduler_name): cprint( - "Every build runs outside forge-cli (a Flink job, for example), so " - "there is nothing to schedule. No files written." + "Every build runs outside forge-cli (a Flink job, for example) or " + "continuously (trigger.type: continuous), so there is nothing to " + "schedule. No files written." ) return 0 diff --git a/fluid_build/cli/test.py b/fluid_build/cli/test.py index 9ba646dcd..80c61bb39 100644 --- a/fluid_build/cli/test.py +++ b/fluid_build/cli/test.py @@ -439,6 +439,12 @@ def _warns(*categories: str) -> List: else: rows.append((_PASS, "Schema syntax", "Valid")) + # Environment overlay (only rendered when the --env overlay patched an + # entry it does not name): --strict fails on it, so it must show. + overlay_issues = [i for i in issues if i.category == "overlay"] + if overlay_issues: + rows.append((_WARN, "Environment overlay", _detail(e.message for e in overlay_issues))) + # Provider connectivity # The connection probe runs in provider detection, so it happens # under --no-data too. diff --git a/fluid_build/cli/validate.py b/fluid_build/cli/validate.py index e1398d996..2fc56268c 100644 --- a/fluid_build/cli/validate.py +++ b/fluid_build/cli/validate.py @@ -106,6 +106,9 @@ def register(subparsers: argparse._SubParsersAction): # Strict validation (warnings as errors) fluid validate contract.fluid.yaml --strict + + # Strict: also fail when an overlay patch lands on another entry + fluid validate contract.fluid.yaml --env prod --strict """, formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -302,7 +305,9 @@ def run(args, logger: logging.Logger) -> int: # Load contract with overlay try: - contract = load_contract_with_overlay(args.contract, getattr(args, "env", None), logger) + contract, overlay_warnings = _load_with_overlay_warnings( + args.contract, getattr(args, "env", None), logger + ) except Exception as e: raise CLIError(1, "contract_load_failed", {"error": str(e)}) @@ -331,6 +336,9 @@ def run(args, logger: logging.Logger) -> int: logger=logger, ) + for warning in overlay_warnings: + validation_result.add_warning(warning) + # Show schema if requested if args.show_schema: _show_schema_info(target_version, schema_manager, args, logger) @@ -355,6 +363,7 @@ def run(args, logger: logging.Logger) -> int: log_metric(logger, "validation_warnings", len(validation_result.warnings), unit="count") # Output results + _print_quiet_overlay_warnings(args, overlay_warnings) exit_code = _output_results(validation_result, args, logger) # Log operation result @@ -452,6 +461,23 @@ def run(args, logger: logging.Logger) -> int: raise CLIError(1, "cli_unhandled_exception", {"error": str(e)}) +def _load_with_overlay_warnings( + contract_path: str, env: Optional[str], logger: logging.Logger +) -> tuple[dict, list[str]]: + """Load a contract with its ``--env`` overlay, and the overlay warnings to report. + + A positional overlay patch that lands on another entry is only logged by + the loader, and ``--strict`` reads the validation result, never the log. + :func:`fluid_build.loader.collect_overlay_warnings` hands it to validate + instead, so it is reported as a warning and ``--strict`` fails on it. + """ + from fluid_build.loader import collect_overlay_warnings + + with collect_overlay_warnings() as overlay_warnings: + contract = load_contract_with_overlay(contract_path, env, logger) + return contract, [warning.message for warning in overlay_warnings] + + def _handle_list_versions(schema_manager: FluidSchemaManager, args, logger: logging.Logger) -> int: """Handle --list-versions flag.""" try: @@ -1258,11 +1284,39 @@ def _output_text_results(result: ValidationResult, args, logger: logging.Logger) treat_warnings_as_errors = args.strict and has_warnings if has_errors or treat_warnings_as_errors: + # Under --quiet the warnings were not listed, errors or not. + if treat_warnings_as_errors and (args.quiet or not has_errors): + _explain_strict_failure(result, args) return 1 else: return 0 +def _explain_strict_failure(result: ValidationResult, args: Any) -> None: + """Say why ``--strict`` exits 1: after a summary that reads valid, or under ``--quiet``. + + ``--quiet`` prints errors only, so there each warning is printed as the + error ``--strict`` made it: a count alone does not say what to fix. + """ + if args.quiet: + for warning in result.warnings: + _cprint_literal(f"ERROR (--strict): {warning}") + else: + cprint("❌ Validation failed: warnings treated as errors (--strict)") + + +def _print_quiet_overlay_warnings(args: Any, overlay_warnings: list[str]) -> None: + """Under ``--quiet``, still show an overlay patch that landed on another entry. + + ``--quiet`` prints errors only, but the loader always printed this warning, + whatever the flags. Under ``--strict`` it is printed as an error instead + (:func:`_explain_strict_failure`). + """ + if args.quiet and not args.strict and args.format == "text": + for warning in overlay_warnings: + _cprint_literal(f"WARNING: {warning}") + + # --------------------------------------------------------------------------- # Public helpers for other CLI commands (publish, apply, …) that need to # run FLUID schema validation on an already-loaded contract dict. @@ -1377,19 +1431,17 @@ def _try_workspace_validate( failed = 0 for product in products: try: - contract = _load_contract_for_workspace(product.contract_path, args, logger) + contract, overlay_warnings = _load_with_overlay_warnings( + str(product.contract_path), getattr(args, "env", None), logger + ) result = schema_manager.validate_contract( contract, strict=args.strict, offline_only=getattr(args, "offline", False), ) - if result.is_valid: - cprint(f" ✅ {product.name:<20} valid ({product.fluid_version or '?'})") + if _report_workspace_product(product, result, overlay_warnings, args.strict): passed += 1 else: - cprint(f" ❌ {product.name:<20} {len(result.errors)} error(s)") - for err in result.errors[:3]: - _cprint_literal(f" {err}") failed += 1 except Exception as exc: # noqa: BLE001 _cprint_literal(f" ❌ {product.name:<20} load error: {exc}") @@ -1403,6 +1455,30 @@ def _try_workspace_validate( return 0 +def _report_workspace_product( + product: Any, result: ValidationResult, overlay_warnings: list[str], strict: bool +) -> bool: + """Print one workspace product's line, and whether it passed. + + ``--strict`` fails a product on an overlay patch that lands on another + entry; its other warnings are not reported here. + """ + if not result.is_valid: + cprint(f" ❌ {product.name:<20} {len(result.errors)} error(s)") + for err in result.errors[:3]: + _cprint_literal(f" {err}") + elif overlay_warnings and strict: + cprint( + f" ❌ {product.name:<20} {len(overlay_warnings)} overlay warning(s), " + "treated as errors (--strict)" + ) + else: + cprint(f" ✅ {product.name:<20} valid ({product.fluid_version or '?'})") + for warning in overlay_warnings: + _cprint_literal(f" ⚠️ {warning}") + return result.is_valid and not (overlay_warnings and strict) + + # --------------------------------------------------------------------------- # Bundle (.tgz) validation — stage 2 of the 11-stage pipeline # --------------------------------------------------------------------------- @@ -1603,13 +1679,3 @@ def _bundle_contract_rule_issues( for msg in result.warnings ) return issues - - -def _load_contract_for_workspace( - contract_path: Path, - args: Any, - logger: logging.Logger, -) -> dict: - """Load a contract, applying environment overlay if requested.""" - env = getattr(args, "env", None) - return load_contract_with_overlay(str(contract_path), env, logger) diff --git a/fluid_build/loader.py b/fluid_build/loader.py index 2cc227b27..bc0866180 100644 --- a/fluid_build/loader.py +++ b/fluid_build/loader.py @@ -15,12 +15,26 @@ # fluid_build/loader.py from __future__ import annotations +import contextlib +import contextvars import json import logging import os import threading from pathlib import Path -from typing import Any, Dict, List, Mapping, NamedTuple, Optional, Set, Tuple, Union +from typing import ( + AbstractSet, + Any, + Dict, + Iterator, + List, + Mapping, + NamedTuple, + Optional, + Set, + Tuple, + Union, +) from fluid_build.util.ref_confinement import ( REF_ROOT_ENV, @@ -404,15 +418,66 @@ def _key_fields(item: Any, keys: Tuple[str, ...]) -> Dict[str, str]: _NOTED_OVERLAY_MERGE_WARNINGS_LOCK = threading.Lock() +class OverlayWarning(NamedTuple): + """An overlay merge warning: its event name and its message.""" + + event: str + message: str + + +#: Overlay warnings ``fluid validate`` and ``fluid contract-validation`` report +#: as their own warnings, so that ``--strict`` fails on them: a positional patch +#: that landed on another entry. ``overlay_entry_added`` is often intended (a +#: build only prod runs) and ``--strict`` could not be silenced on it, so it +#: stays a log line. +STRICT_OVERLAY_WARNINGS = frozenset({"overlay_patch_retargeted"}) + +#: The events and the list of the :func:`collect_overlay_warnings` block that +#: is running, ``None`` outside one. A context variable, as +#: ``warnings.catch_warnings`` is under Python 3.14's ``context_aware_warnings``, +#: so a load in another thread or task does not add to this block's list. +_OVERLAY_WARNING_SINK: contextvars.ContextVar[ + Optional[Tuple[AbstractSet[str], List[OverlayWarning]]] +] = contextvars.ContextVar("overlay_warning_sink", default=None) + + +@contextlib.contextmanager +def collect_overlay_warnings( + events: AbstractSet[str] = STRICT_OVERLAY_WARNINGS, +) -> Iterator[List[OverlayWarning]]: + """Hand the overlay merge warnings named in ``events`` to the caller. + + The loader logs its warnings, and a command's ``--strict`` reads its own + findings, never the log. Inside this block a warning named in ``events`` is + appended to the yielded list instead, once, and logged at DEBUG only: the + caller reports it, so the terminal does not show it twice. A warning the + block takes is not marked as said, so a later load outside a block still + logs it. + """ + sink: List[OverlayWarning] = [] + token = _OVERLAY_WARNING_SINK.set((events, sink)) + try: + yield sink + finally: + _OVERLAY_WARNING_SINK.reset(token) + + def _warn_once(event: str, name: str, index: Optional[int], message: str, *args: Any) -> None: text = message % args + extra: Dict[str, Any] = {"event": event, "list": name} + if index is not None: + extra["index"] = index + collecting = _OVERLAY_WARNING_SINK.get() + if collecting is not None and event in collecting[0]: + warning = OverlayWarning(event, text) + if warning not in collecting[1]: + collecting[1].append(warning) + LOG.debug("%s", text, extra=extra) + return 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) diff --git a/fluid_build/schedulers/synthesis.py b/fluid_build/schedulers/synthesis.py index 3bd681f9d..739161190 100644 --- a/fluid_build/schedulers/synthesis.py +++ b/fluid_build/schedulers/synthesis.py @@ -25,7 +25,7 @@ import re from typing import Any, Dict, List, Optional -from fluid_build.util.contract import is_externally_run_build +from fluid_build.util.contract import is_unscheduled_build # Provider + engine → scheduler action mapping. # Keys are (provider, engine) tuples; values are the action string used @@ -114,12 +114,13 @@ def synthesize_orchestration_from_builds( The returned dict is suitable for injection into the contract before calling a scheduler's ``validate()`` and ``generate()`` methods. """ - # A Flink job runs continuously on its own platform: there is nothing to - # schedule for it, and a nightly task that "runs" it would only log its SQL. + # A Flink job, or any build whose trigger is continuous, runs on its own + # platform: there is nothing to schedule for it, and a nightly task that + # "runs" it would at best log its SQL and at worst start a second copy. builds: List[Dict[str, Any]] = [ b for b in contract.get("builds", []) or [] - if not (isinstance(b, dict) and is_externally_run_build(b)) + if not (isinstance(b, dict) and is_unscheduled_build(b)) ] if not builds: return {} diff --git a/fluid_build/schema_manager.py b/fluid_build/schema_manager.py index 173c14418..4f3a483eb 100644 --- a/fluid_build/schema_manager.py +++ b/fluid_build/schema_manager.py @@ -72,17 +72,24 @@ def format_schema_error(error: Any) -> str: Every path that reports a FLUID-contract schema error formats it here (``fluid validate`` on a contract and on a bundle), so a secret pasted in place of a ``sink.catalogAuth`` variable name is not printed or saved. + + A ``not: {required: [field]}`` rule, a field the object's ``type`` does not + take, reads as `` is not allowed here``, or as the rule's own + ``description`` when it carries one: jsonschema's message would quote the + whole object and the subschema it must not match. """ + value = error.validator_value + if error.validator == "not" and isinstance(value, dict) and list(value) == ["required"]: + why = error.schema.get("description") if isinstance(error.schema, dict) else None + if isinstance(why, str) and why: + return why + return f"{', '.join(map(str, value['required']))} is not allowed here" if not any(part in _UNQUOTED_VALUE_PATHS for part in error.absolute_path): return str(error.message) if error.validator in _KEY_ONLY_VALIDATORS: return str(error.message) if error.validator == "pattern": return f"does not match {error.validator_value!r} (value not shown)" - value = error.validator_value - if error.validator == "not" and isinstance(value, dict) and list(value) == ["required"]: - # ``not: {required: [field]}``: a field the object's ``type`` does not take. - return f"{', '.join(map(str, value['required']))} is not allowed here" if error.validator == "not" and isinstance(value, dict) and list(value) == ["pattern"]: # ``not: {pattern: ...}``: a token-shaped variable name, or a ``${``. return f"must not match {value['pattern']!r} (value not shown)" diff --git a/fluid_build/schemas/fluid-schema-0.7.6.json b/fluid_build/schemas/fluid-schema-0.7.6.json index 3168d0746..bb76e772d 100644 --- a/fluid_build/schemas/fluid-schema-0.7.6.json +++ b/fluid_build/schemas/fluid-schema-0.7.6.json @@ -4316,8 +4316,10 @@ "dependency", "dataset", "schedule_and_dataset", - "timetable" - ] + "timetable", + "continuous" + ], + "description": "NEW in v0.7.6: continuous, for a build that runs as a long-lived job on its own platform, such as an engine: flink job on the Flink Kubernetes Operator or AWS Managed Service for Apache Flink, or a kafka-connect or debezium connector. The platform keeps the job running, so no scheduler starts runs: fluid generate schedule, fluid schedule-sync and the fluid apply DAGs leave the build out. A continuous trigger takes no other trigger, so cron, schedule, datasets, event and timetable are refused next to it, as Databricks Jobs refuses a continuous trigger next to any other." }, "schedule": { "type": "string" @@ -4382,7 +4384,66 @@ "type": "object", "description": "Custom timetable configuration" } - } + }, + "allOf": [ + { + "if": { + "type": "object", + "required": [ + "type" + ], + "properties": { + "type": { + "const": "continuous" + } + } + }, + "then": { + "allOf": [ + { + "description": "cron is not allowed with type continuous: a continuous job runs all the time, so its trigger takes no other trigger. Remove cron, or set type: schedule.", + "not": { + "required": [ + "cron" + ] + } + }, + { + "description": "schedule is not allowed with type continuous: a continuous job runs all the time, so its trigger takes no other trigger. Remove schedule, or set type: schedule.", + "not": { + "required": [ + "schedule" + ] + } + }, + { + "description": "datasets is not allowed with type continuous: a continuous job runs all the time, so its trigger takes no other trigger. Remove datasets, or set type: dataset.", + "not": { + "required": [ + "datasets" + ] + } + }, + { + "description": "event is not allowed with type continuous: a continuous job runs all the time, so its trigger takes no other trigger. Remove event, or set type: event.", + "not": { + "required": [ + "event" + ] + } + }, + { + "description": "timetable is not allowed with type continuous: a continuous job runs all the time, so its trigger takes no other trigger. Remove timetable, or set type: timetable.", + "not": { + "required": [ + "timetable" + ] + } + } + ] + } + } + ] }, "runtime": { "type": "object", diff --git a/fluid_build/util/contract.py b/fluid_build/util/contract.py index d3b30e0a0..fd9922bfd 100644 --- a/fluid_build/util/contract.py +++ b/fluid_build/util/contract.py @@ -158,6 +158,24 @@ def describe_externally_run_build(build: Mapping[str, Any]) -> str: return f"language '{props.get('language') if isinstance(props, Mapping) else None}'" +def is_continuous_build(build: Mapping[str, Any]) -> bool: + """True when the build's ``execution.trigger.type`` is ``continuous`` (new in + schema 0.7.6): its platform keeps the job running, so no scheduler starts + its runs. Named after Databricks Jobs' continuous trigger, not Airflow's + ``@continuous``, which starts batch runs back to back.""" + execution = build.get("execution") + trigger = execution.get("trigger") if isinstance(execution, Mapping) else None + kind = trigger.get("type") if isinstance(trigger, Mapping) else None + return isinstance(kind, str) and kind.strip().lower() == "continuous" + + +def is_unscheduled_build(build: Mapping[str, Any]) -> bool: + """True when no scheduler starts the build's runs: its job runs outside + forge-cli (:func:`is_externally_run_build`), or continuously + (:func:`is_continuous_build`).""" + return is_externally_run_build(build) or is_continuous_build(build) + + def get_contract_version(contract: Mapping[str, Any]) -> Optional[str]: """ Get the FLUID schema version from a contract. diff --git a/tests/test_continuous_trigger.py b/tests/test_continuous_trigger.py new file mode 100644 index 000000000..2d5e5cc2e --- /dev/null +++ b/tests/test_continuous_trigger.py @@ -0,0 +1,238 @@ +# 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. + +"""``trigger.type: continuous`` (schema 0.7.6): a job its platform keeps running. + +A long-running Flink job, or a Kafka Connect or Debezium connector, has no +trigger type: schedule, event and the rest all start runs. ``continuous`` says +no scheduler starts this build, so every path that turns builds into scheduled +runs leaves it out, and the schema refuses another trigger next to it, as +Databricks Jobs refuses a continuous trigger next to any other. +""" + +from __future__ import annotations + +import copy +import json +import os +import subprocess +import sys +from pathlib import Path +from typing import Any, Dict + +import jsonschema +import pytest +import yaml + +from fluid_build.schema_manager import format_schema_error +from fluid_build.util.contract import is_continuous_build, is_unscheduled_build + +SCHEMAS = Path(__file__).resolve().parent.parent / "fluid_build" / "schemas" + + +def _connector(trigger: Dict[str, Any]) -> Dict[str, Any]: + return { + "id": "ingest", + "engine": "kafka-connect", + "pattern": "acquisition", + "execution": {"trigger": trigger}, + } + + +def _contract(fluid_version: str = "0.7.6", **trigger: Any) -> Dict[str, Any]: + """The requester's Flink job, with a continuous trigger.""" + return { + "fluidVersion": fluid_version, + "kind": "DataProduct", + "id": "network.service-instances-full", + "name": "Service instances (reconciled)", + "metadata": {"owner": {"team": "data-engineering"}}, + "builds": [ + { + "id": "reconcile", + "engine": "flink" if fluid_version == "0.7.6" else "sql", + "pattern": "embedded-logic", + "properties": { + "language": "flink_sql", + "sql": "INSERT INTO service_instances_full SELECT * FROM service_instances", + "parameters": {"parallelism": 2}, + }, + "execution": { + "trigger": {"type": "continuous", **trigger}, + "runtime": {"platform": "kubernetes"}, + }, + } + ], + "exposes": [ + { + "exposeId": "service_instances_full", + "kind": "topic", + "binding": { + "platform": "kafka", + "format": "kafka_topic", + "location": {"topic": "network.service-instances-full"}, + }, + "contract": {"schema": [{"name": "id", "type": "string"}]}, + } + ], + } + + +def _errors(version: str, contract: Dict[str, Any]) -> list: + schema = json.loads((SCHEMAS / f"fluid-schema-{version}.json").read_text(encoding="utf-8")) + validator = jsonschema.validators.validator_for(schema)(schema) + return [format_schema_error(e) for e in validator.iter_errors(contract)] + + +class TestSchema: + def test_0_7_6_accepts_a_continuous_trigger(self) -> None: + assert _errors("0.7.6", _contract()) == [] + + def test_0_7_5_is_unchanged(self) -> None: + # 0.7.6 is the preview; the stable version does not move. + assert any("'continuous'" in msg for msg in _errors("0.7.5", _contract("0.7.5"))) + + @pytest.mark.parametrize( + "field, value, remedy", + [ + ("cron", "0 2 * * *", "set type: schedule"), + ("schedule", "@daily", "set type: schedule"), + ("datasets", ["dataset://kafka/network.service-instances"], "set type: dataset"), + ("event", "inventory-updated", "set type: event"), + ("timetable", {"kind": "business-hours"}, "set type: timetable"), + ], + ) + def test_another_trigger_beside_it_is_refused( + self, field: str, value: Any, remedy: str + ) -> None: + # Left in, a scheduler that reads the field alone would start a + # second copy of a job that is already running. + assert _errors("0.7.6", _contract(**{field: value})) == [ + f"{field} is not allowed with type continuous: a continuous job runs all the " + f"time, so its trigger takes no other trigger. Remove {field}, or {remedy}." + ] + + @pytest.mark.parametrize("trigger", [None, "@daily", ["0 2 * * *"]]) + def test_a_trigger_that_is_not_an_object_gets_one_error(self, trigger: Any) -> None: + # A blank `trigger:` in YAML is null; it used to draw five refusals + # of a continuous trigger nobody declared. + contract = _contract() + contract["builds"][0]["execution"]["trigger"] = trigger + assert [m for m in _errors("0.7.6", contract) if "continuous" in m] == [] + assert len(_errors("0.7.6", contract)) == 1 + + @pytest.mark.parametrize( + "trigger", + [ + pytest.param({"type": "schedule", "cron": "0 2 * * *"}, id="schedule-cron"), + pytest.param({"cron": "0 2 * * *"}, id="no-type-cron"), + pytest.param({"type": "dataset", "datasets": ["dataset://x"]}, id="dataset"), + ], + ) + def test_other_trigger_types_keep_their_fields(self, trigger: Dict[str, Any]) -> None: + contract = _contract() + contract["builds"][0]["execution"]["trigger"] = trigger + assert _errors("0.7.6", contract) == [] + + +@pytest.mark.parametrize( + "build, continuous, unscheduled", + [ + pytest.param(_connector({"type": "continuous"}), True, True, id="continuous"), + pytest.param(_connector({"type": " Continuous "}), True, True, id="case-and-space"), + pytest.param(_connector({"type": "schedule"}), False, False, id="schedule"), + pytest.param({"id": "x", "engine": "flink"}, False, True, id="flink-no-trigger"), + pytest.param({"id": "x", "execution": None}, False, False, id="null-execution"), + pytest.param({"id": "x", "execution": {"trigger": "continuous"}}, False, False, id="str"), + ], +) +def test_classification(build: Dict[str, Any], continuous: bool, unscheduled: bool) -> None: + assert is_continuous_build(build) is continuous + assert is_unscheduled_build(build) is unscheduled + + +def test_a_scheduler_gets_no_task_for_a_continuous_connector() -> None: + # A Kafka Connect connector used to get a nightly 02:00 task. + from fluid_build.schedulers.synthesis import synthesize_orchestration_from_builds + + contract = {"id": "p", "builds": [_connector({"type": "continuous"})]} + assert synthesize_orchestration_from_builds(copy.deepcopy(contract), "airflow") == {} + contract["builds"].append({"id": "daily", "engine": "sql", "properties": {"sql": "SELECT 1"}}) + tasks = synthesize_orchestration_from_builds(contract, "airflow")["tasks"] + assert [t["taskId"] for t in tasks] == ["daily"] + + +@pytest.mark.parametrize("kind, scheduled", [("continuous", False), ("schedule", True)]) +def test_a_continuous_build_gets_no_fluid_apply_dag(kind: str, scheduled: bool) -> None: + # With a cron beside it, as a contract that skipped validation can have. + from fluid_build.schedulers.airflow.fluid_apply import has_scheduled_builds + + contract = {"builds": [_connector({"type": kind, "cron": "0 2 * * *"})]} + assert has_scheduled_builds(contract) is scheduled + + +def test_schedule_sync_writes_nothing_for_a_continuous_connector(tmp_path: Path) -> None: + # This emitter reads ``schedule`` alone; the schema refuses the pair, but + # a contract that skipped validation must still not start a second copy. + from fluid_build.cli._acquisition_stage_ext import schedule_sync_acquisition + + contract = {"id": "p", "builds": [_connector({"type": "continuous", "schedule": "0 2 * * *"})]} + assert schedule_sync_acquisition(contract, tmp_path) == [] + contract["builds"][0]["execution"]["trigger"]["type"] = "schedule" + assert schedule_sync_acquisition(contract, tmp_path) + + +def _run_cli(*argv: str, cwd: Path) -> "subprocess.CompletedProcess[str]": + return subprocess.run( + [sys.executable, "-m", "fluid_build.cli", *argv], + cwd=str(cwd), + capture_output=True, + text=True, + env={k: v for k, v in os.environ.items() if not k.endswith("_API_KEY")}, + ) + + +class TestCli: + def test_validate_accepts_a_continuous_trigger(self, tmp_path: Path) -> None: + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(_contract()), "utf-8") + result = _run_cli("validate", "contract.fluid.yaml", "--offline", "--strict", cwd=tmp_path) + assert result.returncode == 0, result.stdout + result.stderr + + def test_validate_names_the_field_and_the_fix(self, tmp_path: Path) -> None: + (tmp_path / "contract.fluid.yaml").write_text( + yaml.safe_dump(_contract(cron="0 2 * * *")), "utf-8" + ) + result = _run_cli("validate", "contract.fluid.yaml", "--offline", cwd=tmp_path) + assert result.returncode == 1 + out = " ".join(result.stdout.split()) + assert "builds[0].execution.trigger: cron is not allowed with type continuous" in out + assert "Remove cron, or set type: schedule." in out + + def test_generate_schedule_has_nothing_to_schedule(self, tmp_path: Path) -> None: + contract = _contract() + contract["builds"] = [_connector({"type": "continuous"})] + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(contract), "utf-8") + result = _run_cli( + "generate", + "schedule", + "contract.fluid.yaml", + "-o", + "sched", + "--scheduler", + "airflow", + cwd=tmp_path, + ) + assert result.returncode == 0, result.stdout + result.stderr + assert "continuously (trigger.type: continuous)" in " ".join(result.stdout.split()) + assert not (tmp_path / "sched").exists() diff --git a/tests/test_overlay_keyed_merge.py b/tests/test_overlay_keyed_merge.py index 81531d66e..5c11e0dc2 100644 --- a/tests/test_overlay_keyed_merge.py +++ b/tests/test_overlay_keyed_merge.py @@ -25,8 +25,13 @@ from __future__ import annotations import copy +import json import logging +import os import re +import subprocess +import sys +from datetime import datetime from pathlib import Path from typing import Any, Dict @@ -335,3 +340,227 @@ def test_an_added_entry_names_its_contract_too( "overlay_entry_added: the overlay for p1", "overlay_entry_added: the overlay for p2", ] + + +# --strict: ``fluid validate`` and ``fluid contract-validation`` take a +# retargeted patch from the loader and report it as their own warning, so +# ``--strict`` fails on it. The loader used to log it, which --strict never read. + + +class TestCollectOverlayWarnings: + def test_the_block_takes_a_retargeted_patch_instead_of_logging_it( + self, caplog: pytest.LogCaptureFixture + ) -> None: + from fluid_build.loader import OverlayWarning, collect_overlay_warnings + + caplog.set_level(logging.DEBUG, logger="fluid.loader") + with collect_overlay_warnings() as taken: + _merge(_base("0.7.5"), PROD) + assert [w.event for w in taken] == ["overlay_patch_retargeted"] + assert isinstance(taken[0], OverlayWarning) + assert "names id transform but patches the base's builds[0], id backfill" in ( + taken[0].message + ) + # Said once, by the command: the log keeps it at DEBUG. + levels = [r.levelno for r in caplog.records if r.name == "fluid.loader"] + assert levels == [logging.DEBUG] + + def test_an_added_entry_stays_a_log_line(self, caplog: pytest.LogCaptureFixture) -> None: + # Adding a build only prod runs is often intended, and --strict could + # not be silenced on it. + from fluid_build.loader import collect_overlay_warnings + + caplog.set_level(logging.WARNING, logger="fluid.loader") + with collect_overlay_warnings() as taken: + _merge(_base("0.7.6"), {"builds": [{"id": "reindex", "engine": "spark"}]}) + assert taken == [] + assert _events(caplog) == ["overlay_entry_added"] + + def test_a_warning_the_block_took_is_still_logged_by_a_later_load( + self, caplog: pytest.LogCaptureFixture + ) -> None: + from fluid_build.loader import collect_overlay_warnings + + caplog.set_level(logging.WARNING, logger="fluid.loader") + with collect_overlay_warnings() as taken: + _merge(_base("0.7.5"), PROD) + _merge(_base("0.7.5"), PROD) + assert len(taken) == 1 + assert _events(caplog) == [] + _merge(_base("0.7.5"), PROD) + assert _events(caplog) == ["overlay_patch_retargeted"] + + def test_outside_a_block_nothing_is_collected(self) -> None: + from fluid_build.loader import _OVERLAY_WARNING_SINK, collect_overlay_warnings + + with collect_overlay_warnings(): + pass + assert _OVERLAY_WARNING_SINK.get() is None + + +def _write_product(tmp_path: Path, version: str, overlay: Dict[str, Any]) -> None: + contract = { + "fluidVersion": version, + "kind": "DataProduct", + "id": "network.service-instances", + "name": "Service instances", + "metadata": {"owner": {"team": "data-engineering"}}, + "builds": [{"id": b["id"], "engine": "custom", "description": b["id"]} for b in BUILDS], + "exposes": [ + { + "exposeId": "service_instances", + "kind": "topic", + "binding": { + "platform": "kafka", + "format": "kafka_topic", + "location": {"topic": "network.service-instances"}, + }, + "contract": {"schema": [{"name": "id", "type": "string"}]}, + } + ], + } + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(contract), encoding="utf-8") + (tmp_path / "overlays").mkdir() + (tmp_path / "overlays" / "prod.yaml").write_text(yaml.safe_dump(overlay), encoding="utf-8") + + +#: Written against ``[ingest, transform]``: its second entry patches transform. +PROD_BY_POSITION = {"builds": [{}, {"id": "transform", "description": "8 executors"}]} + + +def _validate(tmp_path: Path, *argv: str) -> "subprocess.CompletedProcess[str]": + return subprocess.run( + [sys.executable, "-m", "fluid_build.cli", "validate", "contract.fluid.yaml"] + + ["--env", "prod", "--offline", *argv], + cwd=str(tmp_path), + capture_output=True, + text=True, + env={k: v for k, v in os.environ.items() if not k.endswith("_API_KEY")}, + ) + + +class TestValidateStrict: + def test_the_retargeted_patch_is_a_warning(self, tmp_path: Path) -> None: + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + result = _validate(tmp_path) + assert result.returncode == 0, result.stdout + result.stderr + out = result.stdout + result.stderr + assert "Validation Warnings:" in result.stdout + # Reported by validate, not also logged by the loader. + assert out.count("overlay_patch_retargeted") == 1 + assert "the overlay's builds[1] names id transform but patches the base's builds[1], " in ( + " ".join(result.stdout.split()) + ) + + def test_strict_fails_on_it_and_says_why(self, tmp_path: Path) -> None: + # The summary above reads "Valid FLUID contract". + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + result = _validate(tmp_path, "--strict") + assert result.returncode == 1 + assert "Validation failed: warnings treated as errors (--strict)" in result.stdout + + def test_quiet_still_shows_it(self, tmp_path: Path) -> None: + # --quiet prints errors only, but the loader always printed this one. + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + result = _validate(tmp_path, "--quiet") + assert result.returncode == 0, result.stdout + result.stderr + assert result.stdout.startswith("WARNING: overlay_patch_retargeted: in ") + assert (result.stdout + result.stderr).count("overlay_patch_retargeted") == 1 + + def test_quiet_strict_names_it_as_the_error(self, tmp_path: Path) -> None: + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + result = _validate(tmp_path, "--quiet", "--strict") + assert result.returncode == 1 + assert result.stdout.startswith("ERROR (--strict): overlay_patch_retargeted: in ") + assert (result.stdout + result.stderr).count("overlay_patch_retargeted") == 1 + + def test_quiet_strict_names_it_beside_other_errors(self, tmp_path: Path) -> None: + # --strict must not show less than --quiet alone. + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + contract = yaml.safe_load((tmp_path / "contract.fluid.yaml").read_text("utf-8")) + contract["bogusTopLevel"] = 1 + (tmp_path / "contract.fluid.yaml").write_text(yaml.safe_dump(contract), "utf-8") + result = _validate(tmp_path, "--quiet", "--strict") + assert result.returncode == 1 + assert "ERROR: root: Additional properties" in result.stdout + assert "ERROR (--strict): overlay_patch_retargeted: in " in result.stdout + + def test_strict_json_lists_it(self, tmp_path: Path) -> None: + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + result = _validate(tmp_path, "--strict", "--format", "json") + assert result.returncode == 1 + warnings = json.loads(result.stdout)["warnings"] + assert [w.split(":")[0] for w in warnings] == ["overlay_patch_retargeted"] + + def test_keyed_merge_has_nothing_to_warn_about(self, tmp_path: Path) -> None: + overlay = {"builds": [{"id": "transform", "description": "8 executors"}]} + _write_product(tmp_path, "0.7.6", overlay) + result = _validate(tmp_path, "--strict") + assert result.returncode == 0, result.stdout + result.stderr + + def test_an_added_build_does_not_fail_strict(self, tmp_path: Path) -> None: + overlay = {"builds": [{"id": "reindex", "engine": "custom", "description": "reindex"}]} + _write_product(tmp_path, "0.7.6", overlay) + result = _validate(tmp_path, "--strict") + assert result.returncode == 0, result.stdout + result.stderr + + +def test_workspace_validate_strict_fails_on_it(tmp_path: Path) -> None: + # `fluid validate --env prod --strict` with no contract validates every + # product in the workspace. + (tmp_path / "fluid.workspace.yaml").write_text( + yaml.safe_dump({"products_dir": "products"}), encoding="utf-8" + ) + product = tmp_path / "products" / "svc" + product.mkdir(parents=True) + _write_product(product, "0.7.5", PROD_BY_POSITION) + run = [sys.executable, "-m", "fluid_build.cli", "validate", "--env", "prod", "--offline"] + env = {k: v for k, v in os.environ.items() if not k.endswith("_API_KEY")} + lenient = subprocess.run(run, cwd=tmp_path, capture_output=True, text=True, env=env) + strict = subprocess.run( + run + ["--strict"], cwd=tmp_path, capture_output=True, text=True, env=env + ) + assert lenient.returncode == 0, lenient.stdout + lenient.stderr + assert "overlay_patch_retargeted" in lenient.stdout + assert strict.returncode == 1 + assert "1 overlay warning(s), treated as errors (--strict)" in strict.stdout + + +def test_fluid_test_shows_it_as_a_row() -> None: + # `fluid test --strict` failed on it without printing it. + from fluid_build.cli.contract_validation import ValidationReport + from fluid_build.cli.test import _build_check_rows + + report = ValidationReport("c.yaml", "p", "1", datetime.now(), 0.0) + report.add_issue("warning", "overlay", "overlay_patch_retargeted: in p, the overlay's ...") + rows = {name: (status, detail) for status, name, detail in _build_check_rows(report)} + status, detail = rows["Environment overlay"] + assert status != rows["Schema syntax"][0] + assert detail.startswith("overlay_patch_retargeted: in p") + report.issues.clear() + assert "Environment overlay" not in [name for _, name, _ in _build_check_rows(report)] + + +def test_an_id_that_reads_as_markup_does_not_break_either_table(capsys) -> None: + # The message quotes the contract's ids; "[/]" made Rich raise MarkupError. + from fluid_build.cli.contract_validation import ValidationReport, output_rich_report + from fluid_build.cli.test import _output_rich + + message = "overlay_patch_retargeted: in p, the overlay's builds[1] names id [/] but ..." + report = ValidationReport("c.yaml", "p", "1", datetime.now(), 0.0) + report.add_issue("warning", "overlay", message) + output_rich_report(report) + _output_rich(report) + # Each table shows the id as written (the tables wrap the rest of the line). + assert capsys.readouterr().out.count("[/]") == 2 + + +def test_contract_validation_reports_it_as_a_warning(tmp_path: Path) -> None: + from fluid_build.cli.contract_validation import ContractValidator + + _write_product(tmp_path, "0.7.5", PROD_BY_POSITION) + report = ContractValidator( + tmp_path / "contract.fluid.yaml", env="prod", check_data=False, track_history=False + ).validate() + overlay = [w for w in report.get_warnings() if w.category == "overlay"] + assert [w.message.split(":")[0] for w in overlay] == ["overlay_patch_retargeted"]