You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by ep...@apache.org on 2026/09/23 20:10:56 UTC
(airflow) branch main updated: Add serialized Dag version diff engine (#69864)
This is an automated email from the ASF dual-hosted git repository.
ephraimbuddy pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 06377eb38e5 Add serialized Dag version diff engine (#69864)
06377eb38e5 is described below
commit 06377eb38e531621424d5cde1f2f24e1be84721d
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Wed Sep 23 21:10:47 2026 +0100
Add serialized Dag version diff engine (#69864)
* Add serialized Dag version diff engine
Dag authors and operators need a reliable way to understand how two stored
Dag versions differ without downloading serialized payloads or manually
comparing implementation details. This adds the shared comparison engine
that later API and CLI surfaces build on.
Historical serialized Dags can use different schema shapes, so comparisons
normalize task, schedule, dependency, and metadata state before calculating
changes. Malformed legacy payloads and missing historical data degrade to
unavailable results instead of failing the whole request.
* Prevent unauthorized Dag version diff disclosure
* Keep redacted Dag version diffs non-disclosing
Users with version access need a useful structural comparison without learning identifiers or values derived from Dag code. The diff contract also needs to remain stable as serializer fields evolve.
* Harden Dag version diff authorization handling
Authorization state crosses an internal boundary before API integration, so invalid inputs must fail closed. Consistent not-found semantics also ensure future callers can preserve the required HTTP 404 behavior.
* Fix Dag version diff accuracy and change classification
Equivalent stored Dag definitions can appear different across serialization
formats, while scheduling and authorization changes can be misclassified.
Diffs need to reflect effective settings and preserve definition metadata
without exposing redacted values or silently drifting from the serializer.
* Report mapped-task changes shadowed by a client default
A mapped task whose payload carries no partial_kwargs key resolves its
effective values from the top-level fields: populate_operator only folds
client_defaults.tasks into partial_kwargs from inside the branch that
handles a present key, so an absent one leaves nothing to shadow them.
Seeding the effective partial kwargs from the client defaults regardless
inverted that, and the setdefault meant to prefer the explicit value
could never replace the default it had already put there. Two versions
differing only in such a field compared equal, so the diff dropped a
change that changes how the task runs.
Both directions now have a test, because the suite could not tell the
inverted behaviour from either correct or over-eager alternatives: the
mapped-default cases only ever shadowed through partial_kwargs itself.
Each asserts what populate_operator actually resolves before asserting
the diff, so the engine stays pinned to runtime rather than to whichever
precedence looked right at the time.
* Withhold Dag version diff source and values until a caller asks
The two statuses carry an authorization decision the caller has already
made, and reading an absent one as permission put the burden the wrong
way round: an API handler that wired a query parameter to the request
but never computed the status would hand Dag source to whoever could
reach the endpoint, and nothing here would notice. Defaulting both to
withholding makes silence safe, so a forgotten argument can no longer
disclose anything and needs no guard of its own.
That leaves the include_source and include_values booleans with nothing
to say: withholding already had a spelling in the status, the two forms
produced identical output, and keeping both only allowed a caller to
state two things at once while the status quietly decided. The values
section is now emitted whatever the outcome as well, so a client reading
its status does not have to know which request shape produced the
response to avoid a KeyError under a declared diff schema version.
An out-of-range change bound is also rejected up front rather than after
the version lookup and both payload loads, which matters once the bound
arrives from a request.
* Improve serialized Dag version diff accuracy
Equivalent stored representations should not produce execution-impacting changes, and malformed payloads should yield an unavailable comparison.
The existing source endpoint and UI already handle code comparisons, so the structured diff does not need a second source-data contract.
* Fix Dag version comparisons for mapped task defaults
Mapped defaults and template values can share the same numeric representation while resolving to different runtime types. Consumers need accurate comparisons and diagnosable failures before relying on this helper.
* Cover mapped pool precedence in Dag version diff tests
Changes to an explicit mapped-task pool must remain visible when a client default conflicts with it. Regression coverage also needs to preserve the different behavior when partial_kwargs is present but empty.
* Fix the issues found in reviews and more
Comparisons need to preserve hydrated task behavior and report the same change set regardless of value visibility.
* Explain SHA-256 use in Dag version diffs
Serialized content can contain credentials, so the distinction between
content identity and password protection needs to be explicit.
* Name changed task fields in Dag version diffs
The serialized operator schema accepts undeclared fields, so a field the
serializer writes but the schema never names was indistinguishable from
any other unnamed change: nothing told a reviewer that a task's callable
or run-as user had changed.
A redacted diff of a large Dag also spent its whole change budget
repeating one path, and a payload nested deeper than the comparison can
walk surfaced as an error instead of the unavailable result every other
failure produces.
* Fix Dag version diff counts, classification and failure handling
A diff is only worth reading if a reviewer can trust what it says. Repeated
changes at one path reported a truncated result with an understated count even
when every path was present, so an ordinary Dag-level edit on a large Dag looked
like a partial answer. Fields the serializer writes but the schema never declares
were reported as one unnamed change, hiding a swapped callable or run-as user
behind an opaque record, and renaming a task's display name read as a dependency
being removed and added rather than as metadata.
Stored payloads that cannot be decompressed or parsed, and values carrying an
unpaired surrogate, reached the caller as an error instead of the unavailable
result every other failure mode produces. An out-of-range version number
surfaced as a database driver error rather than invalid input.
The wire format this returns had no written contract, which it needs before
anything is built on it.
* Report the impact of a Dag default_args change per key
A Dag whose only edit was an owner in default_args reported that change as
execution-impacting while the task-level owner change it caused reported as
metadata, so the same edit carried two impacts and anything filtering for
execution changes saw a false positive. The keys are the operator fields they
get applied to, so each one can carry the impact of the field it sets. Keys
outside the allowlist still aggregate into one conservative record, because a
key name a Dag author chose must not appear in a redacted path.
The task group path grammar was also being re-derived in six places with two
bounds that disagreed, which is the kind of duplication that eventually drifts
into a redaction bug, and this change would have added a seventh.
* Classify a Dag's first default_args entry per key
A Dag omits default_args entirely until it has one, and the schema carries no
default to fill the gap in, so the first entry added arrived as a missing value
on one side and was reported as one opaque execution-impacting change. That is
the most common form of the edit the per-key classification was meant to
describe: setting an owner for the first time still read as changing execution.
Removing the last entry had the same shape.
Normalizing the absent side to an empty mapping during canonicalization keeps
the boundary on the same path as every other default_args edit.
* Keep Dag version diffs faithful to how stored values hydrate
Two stored shapes were compared as something other than what they resolve to.
A retry backoff named in template_fields keeps its stored value through
hydration, so a stored True and a stored 2.0 are different factors, but
normalizing both to 2.0 reported no change at all; the surrounding timedelta
and date handling already respected that bypass. Task and task group id lists
resolve to sets, so the order a producer happened to store carried no meaning
yet a reordering read as a dependency change with execution impact.
The bound's contract also promised more than it keeps: both modes reach the
same changes and report the same public paths, but a redacted record stands for
every change sharing its path, so counting continues past the bound and the
same payload can report a different truncated flag depending on disclosure.
* Keep id-list normalization off a mapped group's expansion
Sorting a task group's dependency ids matched every descendant whose last key
shared one of those names, so a mapped group's expand_input was normalized too.
An argument's order and multiplicity there decide how many task instances run,
and deduplicating it hid that a group went from two expansions to three: both
disclosure modes reported no change at all. Matching the nesting a group's own
fields sit under keeps the normalization where it belongs.
Two Dag fields also had no schema default to fill the gap in, so a producer
predating them omitted the key while the current one writes an explicit null.
Both hydrate alike, so every Dag compared across that upgrade reported an added
execution change. An empty run-type restriction hydrates to no restriction and
now compares that way too; an empty deadline list is left alone, since it
hydrates to an empty list rather than to null.
* Read partial_kwargs as mapped configuration only when mapped
An operator that is not mapped may declare partial_kwargs a template field,
which makes that mapping its own execution input. Dispatching on the path shape
alone read those keys as operator fields, so a change to user data was reported
with the impact of whichever field name a key collided with: an author's own
"owner" entry decided how the task runs, yet the diff called it metadata. It was
also named in a redacted path, disclosing a key the author chose rather than
leaving it inside the value that was withheld.
Field-by-field comparison only means something when both sides really are
mapped; a mapping that is user input on one side and configuration on the other
cannot be read either way, so it stays opaque like any other field off the
allowlist.
* Fix Dag version diff limits for repeated changes
---
airflow-core/src/airflow/exceptions.py | 2 +-
airflow-core/src/airflow/models/dag_version.py | 161 +-
.../src/airflow/serialization/dag_version_diff.py | 1281 +++++++
.../airflow/serialization/serialized_objects.py | 36 +-
airflow-core/tests/unit/models/test_dag_version.py | 424 ++-
.../unit/serialization/test_dag_serialization.py | 27 +
.../unit/serialization/test_dag_version_diff.py | 3552 ++++++++++++++++++++
7 files changed, 5460 insertions(+), 23 deletions(-)
diff --git a/airflow-core/src/airflow/exceptions.py b/airflow-core/src/airflow/exceptions.py
index cb45b050fce..fdc0408137e 100644
--- a/airflow-core/src/airflow/exceptions.py
+++ b/airflow-core/src/airflow/exceptions.py
@@ -142,7 +142,7 @@ class DagRunNotFound(AirflowNotFoundException):
class DagVersionNotFound(AirflowNotFoundException):
- """Raised when a DagVersion for the given dag_id / bundle_version is not found."""
+ """Raised when a requested DagVersion is not found."""
class DagNotPartitionedError(ValueError):
diff --git a/airflow-core/src/airflow/models/dag_version.py b/airflow-core/src/airflow/models/dag_version.py
index 602364e71b6..c2abb6e3a26 100644
--- a/airflow-core/src/airflow/models/dag_version.py
+++ b/airflow-core/src/airflow/models/dag_version.py
@@ -17,9 +17,11 @@
from __future__ import annotations
+import json
import logging
+import zlib
from datetime import datetime
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal, get_args
from uuid import UUID
import sqlalchemy as sa
@@ -29,6 +31,7 @@ from sqlalchemy.orm import Mapped, joinedload, mapped_column, relationship
from airflow._shared.timezones import timezone
from airflow.dag_processing.bundles.manager import DagBundlesManager
+from airflow.exceptions import DagVersionNotFound
from airflow.models.base import Base, StringID
from airflow.utils.session import NEW_SESSION, provide_session
from airflow.utils.sqlalchemy import UtcDateTime, with_row_locks
@@ -37,9 +40,16 @@ if TYPE_CHECKING:
from sqlalchemy.orm import Session
from sqlalchemy.sql import Select
+ from airflow.models.serialized_dag import SerializedDagModel
+
log = logging.getLogger(__name__)
+ValuesStatus = Literal["available", "unavailable"]
+_VALID_VALUES_STATUSES = get_args(ValuesStatus)
+# version_number is a 32-bit column, and an out-of-range bind parameter makes drivers raise.
+_MAX_VERSION_NUMBER = 2**31 - 1
+
class DagVersion(Base):
"""Model to track the versions of DAGs in the database."""
@@ -245,6 +255,155 @@ class DagVersion(Base):
"""A human-friendly representation of the version."""
return f"{self.dag_id}-{self.version_number}"
+ @classmethod
+ @provide_session
+ def get_diff(
+ cls,
+ dag_id: str,
+ base_version_number: int,
+ target_version_number: int,
+ *,
+ values_status: ValuesStatus = "unavailable",
+ max_changes: int | None = None,
+ session: Session = NEW_SESSION,
+ ) -> dict[str, Any]:
+ """
+ Compare the stored state and provenance of two versions of a Dag.
+
+ ``values_status`` carries the caller's authorization decision. See
+ :func:`~airflow.serialization.dag_version_diff.build_serialized_dag_diff`
+ for disclosure and truncation rules.
+
+ :param dag_id: The Dag ID.
+ :param base_version_number: The version number to compare from.
+ :param target_version_number: The version number to compare to.
+ :param values_status: Whether the caller has authorized disclosing values and digests.
+ :param max_changes: Upper bound on output records; repeated redacted changes share a record.
+ Defaults to ``DEFAULT_MAX_CHANGES``.
+ :param session: The database session.
+ :return: The diff result, in the shape named by its ``diff_schema_version``.
+ :raises ValueError: If ``values_status``, ``max_changes``, or either version number is invalid.
+ :raises DagVersionNotFound: If either version number has no row for ``dag_id``.
+ """
+ # Keep this local to avoid the dag_version -> dag_version_diff -> serialized_objects cycle.
+ from airflow.models.serialized_dag import SerializedDagModel
+ from airflow.serialization.dag_version_diff import (
+ DEFAULT_MAX_CHANGES,
+ build_serialized_dag_diff,
+ build_unavailable_dag_diff,
+ validate_max_changes,
+ )
+
+ if values_status not in _VALID_VALUES_STATUSES:
+ raise ValueError(f"values_status must be one of {_VALID_VALUES_STATUSES}, not {values_status!r}")
+
+ if max_changes is None:
+ max_changes = DEFAULT_MAX_CHANGES
+ # Reject a bad bound before it costs the version lookup and both payload loads.
+ validate_max_changes(max_changes)
+ for version_number in (base_version_number, target_version_number):
+ if not isinstance(version_number, int) or isinstance(version_number, bool):
+ raise ValueError("Dag version numbers must be integers")
+ if not 1 <= version_number <= _MAX_VERSION_NUMBER:
+ raise ValueError(
+ f"Dag version numbers must be positive integers not exceeding {_MAX_VERSION_NUMBER}"
+ )
+
+ query = (
+ select(cls)
+ .where(
+ cls.dag_id == dag_id,
+ cls.version_number.in_((base_version_number, target_version_number)),
+ )
+ .options(joinedload(cls.serialized_dag).selectinload(SerializedDagModel.deadline_alerts))
+ )
+ versions = {version.version_number: version for version in session.scalars(query).all()}
+ missing_version = next(
+ (
+ version_number
+ for version_number in (base_version_number, target_version_number)
+ if version_number not in versions
+ ),
+ None,
+ )
+ if missing_version is not None:
+ raise DagVersionNotFound(
+ f"The DagVersion with dag_id: `{dag_id}` and version_number: `{missing_version}` was not found"
+ )
+
+ base_version = versions[base_version_number]
+ target_version = versions[target_version_number]
+ base_data, base_unavailable_reason = _get_serialized_diff_data(base_version.serialized_dag)
+ target_data, target_unavailable_reason = _get_serialized_diff_data(target_version.serialized_dag)
+ if unavailable_reason := base_unavailable_reason or target_unavailable_reason:
+ return build_unavailable_dag_diff(
+ base_data=base_data, target_data=target_data, reason=unavailable_reason
+ )
+ return build_serialized_dag_diff(
+ base_data=base_data,
+ target_data=target_data,
+ base_provenance=_get_provenance(base_version),
+ target_provenance=_get_provenance(target_version),
+ include_values=values_status == "available",
+ max_changes=max_changes,
+ )
+
+
+def _get_serialized_diff_data(
+ serialized_dag: SerializedDagModel | None,
+) -> tuple[dict[str, Any] | None, str | None]:
+ if serialized_dag is None:
+ return None, None
+ try:
+ # Reading the property decodes the compressed column, so it decodes here too.
+ data = serialized_dag.data
+ if isinstance(data, str):
+ data = json.loads(data)
+ except (ValueError, zlib.error, RecursionError):
+ return None, "serialized_dag_decode_failed"
+ if not isinstance(data, dict) or not isinstance(dag_data := data.get("dag"), dict):
+ return data, None
+ deadlines = dag_data.get("deadline")
+ if not isinstance(deadlines, list) or not any(isinstance(deadline, str) for deadline in deadlines):
+ return data, None
+
+ alerts = {alert.id: alert for alert in serialized_dag.deadline_alerts}
+ definitions = []
+ for deadline in deadlines:
+ if not isinstance(deadline, str):
+ definitions.append(deadline)
+ continue
+ alert_id = _parse_alert_id(deadline)
+ if alert_id is None or (alert := alerts.get(alert_id)) is None:
+ return data, "deadline_alert_missing"
+ definitions.append(
+ # description is left out because no writer populates it, so it is always null.
+ {
+ "name": alert.name,
+ "reference": alert.reference,
+ "interval": alert.interval,
+ "callback": alert.callback_def,
+ }
+ )
+ # Stored UUIDs differ between versions even when the alert definitions are identical.
+ return {**data, "dag": {**dag_data, "deadline": definitions}}, None
+
+
+def _parse_alert_id(deadline: str) -> UUID | None:
+ """Read a stored deadline reference as a UUID, ignoring how the writer spelled it."""
+ try:
+ return UUID(deadline)
+ except ValueError:
+ return None
+
+
+def _get_provenance(version: DagVersion) -> dict[str, Any]:
+ return {
+ "bundle_name": version.bundle_name,
+ "bundle_version": version.bundle_version,
+ "version_data": version.version_data,
+ }
+
def _resolve_version_data(
dag_version: DagVersion | None, bundle_version: str | None
diff --git a/airflow-core/src/airflow/serialization/dag_version_diff.py b/airflow-core/src/airflow/serialization/dag_version_diff.py
new file mode 100644
index 00000000000..c49c8ff56f0
--- /dev/null
+++ b/airflow-core/src/airflow/serialization/dag_version_diff.py
@@ -0,0 +1,1281 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+"""
+Observed-state diffs for serialized Dag payloads.
+
+Diff schema v1 -- the wire format every entry point here returns.
+:func:`build_serialized_dag_diff` and :func:`build_unavailable_dag_diff` both return one
+dictionary in the shape described below, and ``DIFF_SCHEMA_VERSION`` is the version of that
+shape: bump it whenever a client-observable part of this contract changes.
+
+Top-level keys, all always present:
+
+* ``diff_schema_version`` -- ``int``, the value of ``DIFF_SCHEMA_VERSION``.
+* ``serialized_dag_schema_versions`` -- ``{"base": int | None, "target": int | None}``, the
+ ``__version`` each stored payload carried. ``None`` means the key was absent or not an
+ integer; ``bool`` is rejected, so ``__version: true`` reads as ``None`` and never as
+ version 1. A version in ``SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS`` is upgraded to the
+ newest supported version before anything is compared.
+* ``mode`` -- ``"observed_state"`` when the two payloads were compared, ``"unavailable"``
+ when no comparison happened.
+* ``changes`` -- the change records described below, ordered by the deterministic walk.
+ Always empty when ``mode`` is ``"unavailable"``.
+* ``truncated`` -- ``True`` when more underlying changes exist than ``max_changes`` allowed.
+ Always ``False`` when ``mode`` is ``"unavailable"``.
+* ``values`` -- ``{"status": "available" | "unavailable"}``. ``"available"`` only for an
+ ``"observed_state"`` result whose caller authorized value disclosure, so
+ ``result["values"]["status"]`` is safe to read on every result of every entry point.
+
+``unavailable_reason`` is the one conditional key: present exactly when ``mode`` is
+``"unavailable"``. :func:`build_serialized_dag_diff` produces:
+
+* ``serialized_dag_missing`` -- one side has no stored payload.
+* ``serialized_dag_schema_version_missing`` -- one side carries no usable ``__version``.
+* ``unsupported_serialized_dag_schema_version:<n>`` -- ``<n>`` is the first unsupported
+ version found, base before target.
+* ``serialized_dag_canonicalization_failed`` -- a payload could not be normalized (malformed
+ structure, an unsupported ``client_defaults`` section, an unencodable value).
+* ``serialized_dag_recursion_limit_exceeded`` -- the comparison walk ran too deep. Defensive
+ guard; deeply nested payloads generally fail canonicalization first.
+* ``serialized_dag_json_encoding_failed`` -- a value could not be encoded as canonical JSON.
+
+:meth:`~airflow.models.dag_version.DagVersion.get_diff` reuses
+``serialized_dag_decode_failed``, for a stored payload that will not decompress or parse, and
+``deadline_alert_missing``, for a payload that references a deadline alert row that is gone.
+:func:`build_unavailable_dag_diff` passes its caller's reason through unchanged, so a new
+caller extends this list rather than inventing an undocumented value.
+
+Each change record carries these keys in both disclosure modes:
+
+* ``path`` -- a JSON-Pointer-style path into a synthetic document with two roots: ``/dag``
+ (the canonicalized serialized Dag, with ``__version`` dropped and client defaults folded
+ into the tasks) and ``/provenance`` (the provenance mapping the caller supplied --
+ ``bundle_name``, ``bundle_version`` and ``version_data`` from ``get_diff``). ``~`` and
+ ``/`` inside a component are escaped as ``~0`` and ``~1``.
+* ``operation`` -- ``"added"``, ``"removed"`` or ``"changed"``.
+* ``category`` -- a :data:`DiffCategory`: ``task`` for task definitions and the Dag fields
+ that shape task execution, ``dependency`` for ``dag_dependencies`` and downstream task
+ ids, ``schedule`` for timetable, dates and concurrency, ``param`` for params,
+ ``asset`` for task inlets and outlets, ``deadline`` for deadline alerts,
+ ``callback`` for callback presence, ``metadata`` for descriptive and display fields,
+ ``authorization`` for ``access_control``, ``provenance`` for everything under
+ ``/provenance`` plus the file and bundle locators, and ``unknown`` for anything
+ unclassified. ``_DIFF_V1_DAG_FIELD_CATEGORIES`` is the authoritative Dag-field mapping;
+ an unclassified task field falls back to ``task``, which over-reports impact.
+ ``default_args`` is the one Dag field classified per key rather than as a whole, because its
+ keys are the operator fields they will be applied to: an allowlisted key takes that field's
+ category (``owner`` is ``metadata``, ``retries`` is ``task``), and the aggregate record for
+ the rest keeps the conservative ``task``. Only a stored value that is not a readable dict
+ envelope classifies as ``task`` as a whole.
+* ``impact`` -- a :data:`DiffImpact` derived from ``category``: ``provenance``, ``metadata``
+ and ``authorization`` carry through unchanged, every operational category becomes
+ ``execution``, and ``unknown`` stays ``unknown``.
+* ``occurrence_count`` -- how many underlying changes the record stands for.
+
+A caller that authorized values (``include_values=True``, ``values.status`` ``"available"``)
+additionally gets:
+
+* ``before_digest`` / ``after_digest`` -- ``"sha256:<hex>"`` over the canonical JSON of that
+ side, or ``None`` when that side is missing. Both keys are always present.
+* ``before_value`` / ``after_value`` -- the canonicalized value, which is the stored one after
+ schema and client defaults are folded in and params are normalized. ``before_value`` is absent from
+ an ``added`` record and ``after_value`` from a ``removed`` one, which is how a missing side
+ is told apart from a stored ``null``.
+
+``path`` and ``occurrence_count`` also differ between the modes. A redacted record reports the
+public path: members of keyed collections (``/dag/tasks``, ``/dag/dag_dependencies``,
+``/dag/tags``, ``/dag/allowed_run_types`` and task group children) collapse to ``*`` instead
+of naming a task, tag or group, and records sharing a public path and operation are merged
+into one whose ``occurrence_count`` counts the merged changes. An authorized record keeps the
+identifying component and always has an ``occurrence_count`` of 1. Fields outside the v1 task,
+task group and ``default_args`` allowlists are aggregated into a single ``custom_fields`` record
+in both modes, so neither a private serializer field nor a user-chosen ``default_args`` key is
+ever named by either -- ``/dag/default_args`` names only allowlisted operator field names, which
+is why its keys can be classified at all. A top-level key outside the v1 root allowlist is
+likewise reported as ``/custom_fields`` when redacted, while an authorized record keeps its real
+root path.
+
+``max_changes`` bounds the result rather than the traversal: both modes walk the same shape and
+reach the same changes, so authorization never decides which changes exist, and the public paths
+either mode reports -- and the order of their first occurrence -- are the same. What the bound
+admits does differ. An authorized record stands for one change, so the bound caps those directly.
+A redacted record stands for every change sharing its public path and operation, and a change
+reaching a record already in the result costs no new one, so redacted counting continues past the
+bound until a change needs a record the bound will not allow. The same payload at the same bound
+can therefore report more occurrences, and a different ``truncated``, when values are withheld.
+
+``truncated`` reports that such a change was dropped. When it is ``False`` every changed path
+is present with an exact ``occurrence_count``. When it is ``True`` the walk stopped early: some
+paths are absent rather than unchanged, and every ``occurrence_count`` is a lower bound, since
+occurrences after the stop were never reached.
+
+Typed dictionaries for the result and the change record are deliberately deferred -- the
+record shape varies with disclosure mode, so modelling it belongs with the REST layer that
+exposes it. :data:`DiffCategory` and :data:`DiffImpact` are stable enough to generate enums
+from today.
+"""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+from collections.abc import Callable, Mapping, MutableMapping
+from contextlib import suppress
+from dataclasses import dataclass
+from datetime import timedelta
+from enum import Enum
+from typing import Any, Literal, NamedTuple
+
+import structlog
+
+from airflow.serialization.definitions.baseoperator import SerializedBaseOperator
+from airflow.serialization.definitions.mappedoperator import SerializedMappedOperator
+from airflow.serialization.serialized_objects import (
+ _DAG_CALLBACK_FIELDS,
+ _OPERATOR_TIMEDELTA_FIELDS,
+ DagSerialization,
+ OperatorSerialization,
+)
+
+log = structlog.get_logger(__name__)
+
+DIFF_SCHEMA_VERSION = 1
+DEFAULT_MAX_CHANGES = 500
+MAX_ALLOWED_CHANGES = 5000
+SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS = frozenset((1, 2, 3))
+
+DiffCategory = Literal[
+ "asset",
+ "authorization",
+ "callback",
+ "deadline",
+ "dependency",
+ "metadata",
+ "param",
+ "provenance",
+ "schedule",
+ "task",
+ "unknown",
+]
+DiffImpact = Literal["authorization", "execution", "metadata", "provenance", "unknown"]
+
+_ORDER_INSENSITIVE_LIST_PATHS = {
+ ("dag", "tags"),
+ ("dag", "allowed_run_types"),
+}
+_KEYED_COLLECTION_PATHS = {
+ ("dag", "tasks"),
+ ("dag", "dag_dependencies"),
+ *_ORDER_INSENSITIVE_LIST_PATHS,
+}
+# Canonicalization folds away __version and client_defaults and adds provenance, so these are the
+# only top-level sections the walk is allowed to name.
+_DIFF_V1_PUBLIC_ROOT_FIELDS = frozenset({"dag", "provenance"})
+_CUSTOM_TASK_FIELDS_PATH_COMPONENT = "custom_fields"
+# This allowlist is part of diff schema v1. Serializer schema changes must not
+# silently change the paths visible to callers of the diff API.
+_DIFF_V1_PUBLIC_TASK_FIELDS = frozenset(
+ {
+ "__type",
+ "_can_skip_downstream",
+ "_disallow_kwargs_override",
+ "_expand_input_attr",
+ "_is_empty",
+ "_is_mapped",
+ "_is_sensor",
+ "_logger_name",
+ "_needs_expansion",
+ "_operator_extra_links",
+ "_operator_name",
+ "_task_display_name",
+ "_task_module",
+ "allow_nested_operators",
+ "depends_on_past",
+ "do_xcom_push",
+ "doc",
+ "doc_json",
+ "doc_md",
+ "doc_rst",
+ "doc_yaml",
+ "downstream_task_ids",
+ "email",
+ "email_on_failure",
+ "email_on_retry",
+ "end_date",
+ "execution_timeout",
+ "executor",
+ "executor_config",
+ "expand_input",
+ "has_on_execute_callback",
+ "has_on_failure_callback",
+ "has_on_retry_callback",
+ "has_on_skipped_callback",
+ "has_on_success_callback",
+ "has_retry_policy",
+ "ignore_first_depends_on_past",
+ "inlets",
+ "is_setup",
+ "is_stub",
+ "is_teardown",
+ "map_index_template",
+ "max_active_tis_per_dag",
+ "max_active_tis_per_dagrun",
+ "max_retry_delay",
+ "multiple_outputs",
+ "on_failure_fail_dagrun",
+ "op_kwargs_expand_input",
+ "outlets",
+ "owner",
+ "params",
+ "partial_kwargs",
+ "pool",
+ "pool_slots",
+ "priority_weight",
+ "python_callable_name",
+ "queue",
+ "render_template_as_native_obj",
+ "reschedule",
+ "resources",
+ "retries",
+ "retry_delay",
+ "retry_exponential_backoff",
+ "run_as_user",
+ "start_date",
+ "start_from_trigger",
+ "start_trigger_args",
+ "task_id",
+ "task_type",
+ "template_ext",
+ "template_fields",
+ "template_fields_renderers",
+ "trigger_rule",
+ "ui_color",
+ "ui_fgcolor",
+ "wait_for_downstream",
+ "wait_for_past_depends_before_skipping",
+ "weight_rule",
+ }
+)
+_DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS = frozenset({"_arg_bindings"})
+# Task-field categories used by _get_category.
+_DIFF_V1_TASK_ASSET_FIELDS = frozenset({"inlets", "outlets"})
+_DIFF_V1_TASK_PARAM_FIELDS = frozenset({"params"})
+_DIFF_V1_TASK_DEPENDENCY_FIELDS = frozenset({"downstream_task_ids"})
+_DIFF_V1_TASK_METADATA_FIELDS = frozenset(
+ {
+ "_operator_name",
+ "doc",
+ "doc_json",
+ "doc_md",
+ "doc_rst",
+ "doc_yaml",
+ "owner",
+ "ui_color",
+ "ui_fgcolor",
+ "_task_display_name",
+ "task_display_name",
+ }
+)
+_DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS = _DIFF_V1_PUBLIC_TASK_FIELDS | {"task_display_name"}
+_DEFAULT_ARGS_PATH = ("dag", "default_args")
+_RETRY_BACKOFF_FIELD = "retry_exponential_backoff"
+# Dag fields the schema gives no default for, added after the first serializer versions.
+_ABSENT_AS_NULL_DAG_FIELDS = ("allowed_run_types", "deadline")
+# These hydrate to sets (_deserialize_operator_field for a task, TaskGroup.*_ids.update for a group),
+# so a reordered stored list means the same edges and must not read as a dependency change.
+_SET_VALUED_ID_FIELDS = frozenset(
+ {"downstream_task_ids", "upstream_task_ids", "downstream_group_ids", "upstream_group_ids"}
+)
+# A Dag's default_args holds the same operator __init__ kwargs partial_kwargs holds -- the
+# serializer rewrites _HAS_FLAG_FIELDS in both -- so it reuses that allowlist. "__type" names the
+# stored value's own dict envelope rather than a key inside it.
+_DIFF_V1_PUBLIC_DEFAULT_ARGS_FIELDS = _DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS - {"__type"}
+# Classify every Dag schema field explicitly so new fields require a policy decision.
+_DIFF_V1_DAG_FIELD_CATEGORIES: dict[str, DiffCategory] = {
+ "_concurrency": "schedule",
+ "_processor_dags_folder": "provenance",
+ "access_control": "authorization",
+ "allowed_run_types": "schedule",
+ "bundle_name": "provenance",
+ "catchup": "schedule",
+ "dag_dependencies": "dependency",
+ "dag_display_name": "metadata",
+ "dag_id": "metadata",
+ "dagrun_timeout": "schedule",
+ "deadline": "deadline",
+ # Only reached when the stored value is not the dict envelope _collect_default_args_changes
+ # walks per key; an unreadable envelope keeps the conservative impact of its widest key.
+ "default_args": "task",
+ "description": "metadata",
+ # Decides whether a run pins a bundle version, so it changes which code later runs execute
+ # and whether triggering a historical version is allowed at all.
+ "disable_bundle_versioning": "task",
+ "doc_md": "metadata",
+ "edge_info": "metadata",
+ "end_date": "schedule",
+ "fail_fast": "schedule",
+ "fileloc": "provenance",
+ "has_on_failure_callback": "callback",
+ "has_on_success_callback": "callback",
+ "is_paused_upon_creation": "schedule",
+ "max_active_runs": "schedule",
+ "max_active_tasks": "schedule",
+ "max_consecutive_failed_dag_runs": "schedule",
+ "owner_links": "metadata",
+ "params": "param",
+ "relative_fileloc": "provenance",
+ "render_template_as_native_obj": "task",
+ "rerun_with_latest_version": "task",
+ "start_date": "schedule",
+ "tags": "metadata",
+ "task_group": "task",
+ "tasks": "task",
+ "timetable": "schedule",
+ "timezone": "schedule",
+}
+# Fields dropped from the current schema that a stored payload can still carry. Every entry must
+# survive _canonicalize_payload_v1, or it classifies nothing.
+_DIFF_V1_LEGACY_DAG_FIELD_CATEGORIES: dict[str, DiffCategory] = {
+ "fail_stop": "schedule",
+ "on_failure_callback": "callback",
+ "on_success_callback": "callback",
+ "schedule": "schedule",
+}
+_RECURSIVE_MAPPING_PATHS = {
+ (),
+ ("dag",),
+ ("provenance",),
+ *_KEYED_COLLECTION_PATHS,
+}
+_DIFF_V1_TASK_GROUP_METADATA_FIELDS = frozenset(
+ {"group_display_name", "tooltip", "doc_md", "ui_color", "ui_fgcolor"}
+)
+_DIFF_V1_PUBLIC_TASK_GROUP_FIELDS = _DIFF_V1_TASK_GROUP_METADATA_FIELDS | {
+ "_group_id",
+ "prefix_group_id",
+ "children",
+ "upstream_group_ids",
+ "downstream_group_ids",
+ "upstream_task_ids",
+ "downstream_task_ids",
+ "expand_input",
+ "is_mapped",
+}
+
+
+def build_unavailable_dag_diff(
+ *,
+ base_data: dict[str, Any] | None,
+ target_data: dict[str, Any] | None,
+ reason: str,
+) -> dict[str, Any]:
+ """Report a known unavailable reason without comparing the stored payloads."""
+ return _mark_unavailable(
+ _build_diff_result(_get_schema_version(base_data), _get_schema_version(target_data)), reason
+ )
+
+
+def build_serialized_dag_diff(
+ *,
+ base_data: dict[str, Any] | None,
+ target_data: dict[str, Any] | None,
+ base_provenance: Mapping[str, Any] | None = None,
+ target_provenance: Mapping[str, Any] | None = None,
+ include_values: bool = False,
+ max_changes: int = DEFAULT_MAX_CHANGES,
+) -> dict[str, Any]:
+ """
+ Compare two serialized Dag payloads and their provenance deterministically.
+
+ Callers must authorize disclosure of the entire serialized payload, including
+ access-control roles and permissions, before setting ``include_values=True``.
+ This exposes canonicalized values, digests, and identifying path components.
+
+ ``max_changes`` limits output records. Redacted changes with the same public path
+ and operation share a record. A new record exceeding the limit stops the walk;
+ ``truncated`` is then true and occurrence counts are lower bounds.
+
+ An ``unavailable`` result includes the reason comparison failed.
+ """
+ validate_max_changes(max_changes)
+
+ base_schema_version = _get_schema_version(base_data)
+ target_schema_version = _get_schema_version(target_data)
+ result = _build_diff_result(base_schema_version, target_schema_version)
+
+ if base_data is None or target_data is None:
+ return _mark_unavailable(result, "serialized_dag_missing")
+
+ if base_schema_version is None or target_schema_version is None:
+ return _mark_unavailable(result, "serialized_dag_schema_version_missing")
+
+ unsupported_versions = [
+ version
+ for version in (base_schema_version, target_schema_version)
+ if version not in SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS
+ ]
+ if unsupported_versions:
+ return _mark_unavailable(
+ result, f"unsupported_serialized_dag_schema_version:{unsupported_versions[0]}"
+ )
+
+ try:
+ base_document = _canonicalize_payload_v1(base_data)
+ target_document = _canonicalize_payload_v1(target_data)
+ base_document["provenance"] = _canonicalize_value(dict(base_provenance or {}), path=("provenance",))
+ target_document["provenance"] = _canonicalize_value(
+ dict(target_provenance or {}), path=("provenance",)
+ )
+ except (AttributeError, KeyError, OverflowError, RecursionError, TypeError, ValueError) as error:
+ log.warning(
+ "Serialized Dag diff canonicalization failed",
+ error_type=type(error).__name__,
+ reason=str(error),
+ base_schema_version=base_schema_version,
+ target_schema_version=target_schema_version,
+ )
+ return _mark_unavailable(result, "serialized_dag_canonicalization_failed")
+
+ collector = _ChangeCollector(max_changes=max_changes, include_values=include_values)
+ try:
+ _collect_changes(base_document, target_document, path=(), collector=collector)
+ except RecursionError:
+ log.warning(
+ "Serialized Dag diff recursion limit exceeded",
+ base_schema_version=base_schema_version,
+ target_schema_version=target_schema_version,
+ )
+ return _mark_unavailable(result, "serialized_dag_recursion_limit_exceeded")
+ except _JsonEncodingError:
+ log.warning(
+ "Serialized Dag diff JSON encoding failed",
+ base_schema_version=base_schema_version,
+ target_schema_version=target_schema_version,
+ )
+ return _mark_unavailable(result, "serialized_dag_json_encoding_failed")
+
+ result["changes"] = collector.changes
+ result["truncated"] = collector.is_truncated
+ if include_values:
+ result["values"] = {"status": "available"}
+ return result
+
+
+class _ChangeCollector:
+ def __init__(self, *, max_changes: int, include_values: bool) -> None:
+ self.changes: list[dict[str, Any]] = []
+ self._record_count = 0
+ self.max_changes = max_changes
+ self.include_values = include_values
+ self._truncated = False
+ self._redacted_changes: dict[tuple[tuple[str, ...], str], dict[str, Any]] = {}
+
+ @property
+ def is_truncated(self) -> bool:
+ return self._truncated
+
+ def add(
+ self,
+ *,
+ path: tuple[str, ...],
+ operation: Literal["added", "removed", "changed"],
+ before: Any,
+ after: Any,
+ ) -> None:
+ public_path = _get_public_path(path)
+ key = (public_path, operation)
+ if not self.include_values and (existing := self._redacted_changes.get(key)) is not None:
+ # Keep repeated changes from crowding out distinct paths.
+ existing["occurrence_count"] += 1
+ return
+ self._record_count += 1
+ if self._record_count > self.max_changes:
+ self._truncated = True
+ return
+
+ category = _get_category(public_path)
+ change: dict[str, Any] = {
+ "path": _format_path(path if self.include_values else public_path),
+ "operation": operation,
+ "category": category,
+ "impact": _get_impact(category),
+ "occurrence_count": 1,
+ }
+ if self.include_values:
+ change["before_digest"] = None if before is _MISSING else _get_digest(before)
+ change["after_digest"] = None if after is _MISSING else _get_digest(after)
+ if before is not _MISSING:
+ change["before_value"] = before
+ if after is not _MISSING:
+ change["after_value"] = after
+ else:
+ self._redacted_changes[key] = change
+ self.changes.append(change)
+
+
+_MISSING = object()
+
+
+def validate_max_changes(max_changes: int) -> None:
+ if not isinstance(max_changes, int) or isinstance(max_changes, bool) or max_changes < 1:
+ raise ValueError("max_changes must be a positive integer")
+ if max_changes > MAX_ALLOWED_CHANGES:
+ raise ValueError(f"max_changes must not exceed {MAX_ALLOWED_CHANGES}")
+
+
+def _get_schema_version(data: Mapping[str, Any] | None) -> int | None:
+ if not isinstance(data, Mapping):
+ return None
+ version = data.get("__version")
+ return version if isinstance(version, int) and not isinstance(version, bool) else None
+
+
+def _mark_unavailable(result: dict[str, Any], reason: str) -> dict[str, Any]:
+ result["mode"] = "unavailable"
+ result["unavailable_reason"] = reason
+ return result
+
+
+def _build_diff_result(base_schema_version: int | None, target_schema_version: int | None) -> dict[str, Any]:
+ return {
+ "diff_schema_version": DIFF_SCHEMA_VERSION,
+ "serialized_dag_schema_versions": {
+ "base": base_schema_version,
+ "target": target_schema_version,
+ },
+ "mode": "observed_state",
+ "changes": [],
+ "truncated": False,
+ # Always present so reading result["values"]["status"] is safe on every outcome.
+ "values": {"status": "unavailable"},
+ }
+
+
+def _canonicalize_payload_v1(data: dict[str, Any]) -> dict[str, Any]:
+ payload = copy.deepcopy(data)
+ version = _get_schema_version(payload)
+ if version is None:
+ raise ValueError("missing or invalid __version")
+ if version == 1:
+ DagSerialization.conversion_v1_to_v2(payload)
+ DagSerialization.conversion_v2_to_v3(payload)
+ elif version == 2:
+ DagSerialization.conversion_v2_to_v3(payload)
+ if not isinstance(payload.get("dag"), Mapping):
+ raise ValueError("missing dag object")
+ dag_defaults = {
+ field: value
+ for field, value in DagSerialization.get_schema_defaults("dag").items()
+ # Dag callback flags are enabled by their presence, even when their value is false.
+ if field not in _DAG_CALLBACK_FIELDS
+ }
+ payload["dag"] = {**dag_defaults, **payload["dag"]}
+ for field in _DAG_CALLBACK_FIELDS & payload["dag"].keys():
+ payload["dag"][field] = True
+ if "params" in payload["dag"]:
+ payload["dag"]["params"] = _normalize_params(payload["dag"]["params"])
+ # The schema carries no default, so a Dag without default_args omits the key entirely. An
+ # absent side would reach the walk as a missing value and report the whole mapping as one
+ # opaque change, which is how adding the very first entry would escape per-key classification.
+ payload["dag"].setdefault("default_args", {"__type": "dict", "__var": {}})
+ # Neither field has a schema default to fill the gap in, so a producer that predates them omits
+ # the key while the current one writes an explicit null. Both hydrate the same, and a diff that
+ # spans an upgrade would otherwise report an added execution change on every Dag.
+ for field in _ABSENT_AS_NULL_DAG_FIELDS:
+ payload["dag"].setdefault(field, None)
+ if payload["dag"].get("allowed_run_types") == []:
+ # An empty restriction list hydrates to None, the same as no restriction at all.
+ payload["dag"]["allowed_run_types"] = None
+ _apply_task_defaults(payload)
+ payload.pop("__version", None)
+ return _canonicalize_value(payload, path=())
+
+
+@dataclass(frozen=True)
+class _TaskDefaultSets:
+ """The default sets that every task in one payload is folded against."""
+
+ schema_defaults: Mapping[str, Any]
+ outer_schema_defaults: Mapping[str, Any]
+ partial_schema_defaults: Mapping[str, Any]
+ partial_fields: frozenset[str]
+ client_task_defaults: Mapping[str, Any]
+ inherited_dates: Mapping[str, Any]
+
+
+def _build_task_default_sets(
+ payload: dict[str, Any], client_task_defaults: Mapping[str, Any]
+) -> _TaskDefaultSets:
+ # get_schema_defaults is lru_cache'd and hands back the very mapping the live Dag-hydration path
+ # reads, so every use of it in this module must stay read-only. Mutating it in place would
+ # corrupt Dag hydration process-wide, not merely produce a wrong diff.
+ schema_defaults = DagSerialization.get_schema_defaults("operator")
+ partial_fields = (
+ SerializedBaseOperator.get_serialized_fields() - SerializedMappedOperator.get_serialized_fields()
+ )
+ return _TaskDefaultSets(
+ schema_defaults=schema_defaults,
+ outer_schema_defaults={
+ field: value for field, value in schema_defaults.items() if field not in partial_fields
+ },
+ # A mapped task resolves these through partial_kwargs, so their defaults have to be applied
+ # there and encoded like any other partial value rather than left at the outer level.
+ partial_schema_defaults={
+ field: value for field, value in schema_defaults.items() if field in partial_fields
+ },
+ partial_fields=partial_fields,
+ client_task_defaults=client_task_defaults,
+ # set_task_dag_references falls back to the Dag's dates, and the serializer elides a task date
+ # that already matches, so an absent task date means the Dag's date rather than a change.
+ inherited_dates={field: payload["dag"].get(field) for field in ("start_date", "end_date")},
+ )
+
+
+def _apply_task_defaults(payload: dict[str, Any]) -> None:
+ client_defaults = payload.pop("client_defaults", None)
+ if client_defaults is None:
+ client_defaults = {}
+ if not isinstance(client_defaults, Mapping):
+ raise ValueError("client_defaults is not an object")
+
+ # Fail loudly on a section this version cannot fold in: dropping it would silently
+ # compare two payloads as equal when the unhandled defaults actually differ.
+ if unknown_sections := client_defaults.keys() - {"tasks"}:
+ raise ValueError(f"unsupported client_defaults sections: {sorted(unknown_sections)}")
+
+ client_task_defaults = client_defaults.get("tasks", {})
+ if not isinstance(client_task_defaults, Mapping):
+ raise ValueError("client_defaults.tasks is not an object")
+
+ defaults = _build_task_default_sets(payload, client_task_defaults)
+ tasks = payload["dag"].get("tasks", [])
+ if not isinstance(tasks, list):
+ raise ValueError("dag.tasks is not a list")
+ for task in tasks:
+ upgraded_task = _upgrade_task_entry(task, client_defaults)
+ template_fields = upgraded_task.get("template_fields", [])
+ if upgraded_task.get("_is_mapped"):
+ task_data = _build_mapped_task_data(upgraded_task, template_fields, defaults)
+ else:
+ task_data = _build_unmapped_task_data(upgraded_task, template_fields, defaults)
+ _apply_inherited_dates(task_data, template_fields, defaults)
+ if "params" in task_data and "params" not in template_fields:
+ task_data["params"] = _normalize_params(task_data["params"])
+ for field in _SET_VALUED_ID_FIELDS & task_data.keys():
+ if field not in template_fields:
+ task_data[field] = _sort_id_list(task_data[field])
+ task["__var"] = task_data
+
+
+def _upgrade_task_entry(task: Any, client_defaults: Mapping[str, Any]) -> dict[str, Any]:
+ if not isinstance(task, dict) or not isinstance(task.get("__var"), Mapping):
+ raise ValueError("task entry is not an object")
+ encoded_task = OperatorSerialization._apply_defaults_to_encoded_op(
+ dict(task["__var"]), dict(client_defaults)
+ )
+ upgraded_task = OperatorSerialization._upgrade_encoded_operator(encoded_task)
+ # Operator identity is selected before client defaults are applied during hydration.
+ upgraded_task["_is_mapped"] = bool(task["__var"].get("_is_mapped", False))
+ return upgraded_task
+
+
+def _build_mapped_task_data(
+ upgraded_task: dict[str, Any], template_fields: Any, defaults: _TaskDefaultSets
+) -> dict[str, Any]:
+ task_data = {**defaults.outer_schema_defaults, **upgraded_task}
+ effective_partial_kwargs = _build_effective_partial_kwargs(task_data, template_fields, defaults)
+ task_data["partial_kwargs"] = effective_partial_kwargs
+ _normalize_retry_backoff(effective_partial_kwargs)
+ return task_data
+
+
+def _build_effective_partial_kwargs(
+ task_data: dict[str, Any], template_fields: Any, defaults: _TaskDefaultSets
+) -> dict[str, Any]:
+ """Resolve a mapped task's partial_kwargs, consuming the outer fields that move into them."""
+ partial_kwargs = task_data.get("partial_kwargs", {})
+ if not isinstance(partial_kwargs, Mapping):
+ raise ValueError("partial_kwargs is not an object")
+ # populate_operator only folds client defaults into partial_kwargs when the payload
+ # carries the key, so an absent one leaves the top-level value as the effective value.
+ effective_partial_kwargs = (
+ {
+ field: _encode_partial_field_value(field, value)
+ for field, value in defaults.client_task_defaults.items()
+ }
+ if "partial_kwargs" in task_data
+ else {}
+ )
+ for field, value in partial_kwargs.items():
+ effective_partial_kwargs[field] = (
+ value if _is_encoded_partial_value(value) else _encode_partial_field_value(field, value)
+ )
+ # Match populate_operator: partial values take precedence over outer task defaults.
+ for field in defaults.partial_fields & task_data.keys():
+ value = task_data.pop(field)
+ # Outer fields are already encoded unless template handling bypasses deserialization.
+ if field in template_fields:
+ value = _encode_json_value(value)
+ effective_partial_kwargs.setdefault(field, value)
+ for field, value in defaults.partial_schema_defaults.items():
+ effective_partial_kwargs.setdefault(field, _encode_partial_field_value(field, value))
+ return effective_partial_kwargs
+
+
+def _build_unmapped_task_data(
+ upgraded_task: dict[str, Any], template_fields: Any, defaults: _TaskDefaultSets
+) -> dict[str, Any]:
+ task_data = {**defaults.schema_defaults, **upgraded_task}
+ # populate_operator leaves a template field at its stored value, so converting one here would
+ # compare a stored True and a stored 2.0 as equal when they hydrate to different backoff factors.
+ if not (_RETRY_BACKOFF_FIELD in template_fields and _RETRY_BACKOFF_FIELD in upgraded_task):
+ _normalize_retry_backoff(task_data)
+ for field in _OPERATOR_TIMEDELTA_FIELDS & task_data.keys():
+ value = task_data[field]
+ task_data[field] = (
+ _encode_json_value(value)
+ if field in template_fields and field in upgraded_task
+ else _encode_partial_field_value(field, value)
+ )
+ return task_data
+
+
+def _apply_inherited_dates(
+ task_data: dict[str, Any], template_fields: Any, defaults: _TaskDefaultSets
+) -> None:
+ for field, dag_date in defaults.inherited_dates.items():
+ if task_data.get(field) is None:
+ task_data[field] = _encode_partial_field_value(field, dag_date)
+ elif field in template_fields:
+ task_data[field] = _encode_json_value(task_data[field])
+ else:
+ task_data[field] = _encode_partial_field_value(field, task_data[field])
+
+
+# These two predicates deliberately disagree on what counts as already encoded, because the
+# serializer sites they mirror disagree: keep them apart rather than "aligning" them.
+def _is_encoded_partial_value(value: Any) -> bool:
+ # _deserialize_partial_kwargs only takes the full-deserialization path with BOTH keys present.
+ return isinstance(value, Mapping) and "__type" in value and "__var" in value
+
+
+def _is_encoded_param_attribute(value: Any) -> bool:
+ # _deserialize_param detects a legacy encoding from "__type" alone, with no "__var" required.
+ if isinstance(value, Mapping):
+ return "__type" in value
+ if isinstance(value, list):
+ return all(isinstance(item, Mapping) and "__type" in item for item in value)
+ return False
+
+
+def _normalize_params(params: Any) -> Any:
+ """Normalize legacy Params without losing the order used by the trigger form."""
+ if isinstance(params, Mapping):
+ # 2.9.2 and earlier stored params as a JSON object instead of ordered pairs.
+ pairs: Any = params.items()
+ elif isinstance(params, list):
+ pairs = params
+ else:
+ return params
+
+ normalized: dict[str, Any] = {}
+ for pair in pairs:
+ if not isinstance(pair, (list, tuple)) or len(pair) != 2:
+ # Leave an unrecognised shape to the raw comparison rather than guessing at it.
+ return params
+ name, value = pair
+ if isinstance(value, Mapping) and "__class" in value:
+ value = {
+ "default": _normalize_param_attribute(value.get("default")),
+ "description": _normalize_param_attribute(value.get("description")),
+ "schema": _normalize_param_attribute(value.get("schema") or {}),
+ "source": _normalize_param_attribute(value.get("source")),
+ }
+ else:
+ value = {
+ "default": _encode_json_value(value),
+ "description": None,
+ "schema": _encode_json_value({}),
+ "source": None,
+ }
+ normalized[str(name)] = value
+ return [[name, value] for name, value in normalized.items()]
+
+
+def _normalize_param_attribute(value: Any) -> Any:
+ # Match _deserialize_param's legacy-encoding detection without hydrating user objects.
+ if _is_encoded_param_attribute(value):
+ return value
+ return _encode_json_value(value)
+
+
+def _encode_partial_field_value(field: str, value: Any) -> Any:
+ if field in _OPERATOR_TIMEDELTA_FIELDS and value is not None:
+ return {"__type": "timedelta", "__var": value}
+ if field.endswith("_date") and value is not None and not isinstance(value, str):
+ return {"__type": "datetime", "__var": value}
+ if field == "resources":
+ # Resources serialize as a raw mapping; a plain dictionary has a dict envelope instead.
+ return value
+ return _encode_json_value(value)
+
+
+def _encode_json_value(value: Any) -> Any:
+ """Encode plain JSON without invoking object serializers."""
+ if isinstance(value, Mapping):
+ return {"__type": "dict", "__var": {key: _encode_json_value(item) for key, item in value.items()}}
+ if isinstance(value, list):
+ return [_encode_json_value(item) for item in value]
+ return value
+
+
+def _is_group_dependency_path(path: tuple[str, ...]) -> bool:
+ """
+ Identify a dependency id list that belongs to a task group itself.
+
+ Canonicalization paths carry no list indices, so a group's own fields sit under repeated
+ ``children/<group_id>`` pairs. Requiring that exact shape keeps the normalization off a mapped
+ group's ``expand_input``, where a user argument's order and multiplicity decide how many task
+ instances run. A task group declares no template fields, so nothing else needs guarding.
+ """
+ if path[:2] != ("dag", "task_group") or path[-1] not in _SET_VALUED_ID_FIELDS:
+ return False
+ nesting = path[2:-1]
+ return len(nesting) % 2 == 0 and all(name == "children" for name in nesting[::2])
+
+
+def _sort_id_list(value: Any) -> Any:
+ """Order a stored id list the way its hydrated set compares, leaving any other shape alone."""
+ if isinstance(value, list) and all(isinstance(item, str) for item in value):
+ return sorted(set(value))
+ return value
+
+
+def _normalize_retry_backoff(task_fields: dict[str, Any]) -> None:
+ if _RETRY_BACKOFF_FIELD in task_fields:
+ value = task_fields[_RETRY_BACKOFF_FIELD]
+ # A non-numeric stored value stays as it is so the comparison reports just this field
+ # instead of collapsing the whole diff to a canonicalization failure.
+ with suppress(TypeError, ValueError):
+ task_fields[_RETRY_BACKOFF_FIELD] = 2.0 if value is True else float(value)
+
+
+def _canonicalize_value(value: Any, *, path: tuple[str, ...]) -> Any:
+ if isinstance(value, Mapping):
+ if path == ("dag", "deadline"):
+ if value.get("__type") == "deadline_alert":
+ value = value["__var"]
+ value = {"name": None, **value}
+ interval = value.get("interval")
+ if isinstance(interval, (int, float)) and not isinstance(interval, bool):
+ # Diff schema v1 uses the SDK's version-2 timedelta encoding for legacy seconds.
+ value["interval"] = {
+ "__classname__": "datetime.timedelta",
+ "__version__": 2,
+ "__data__": timedelta(seconds=interval).total_seconds(),
+ }
+ return {
+ canonical_key: _canonicalize_value(item, path=path + (canonical_key,))
+ for canonical_key, item in (
+ (_canonicalize_mapping_key(key), item)
+ for key, item in sorted(value.items(), key=lambda item: _canonicalize_mapping_key(item[0]))
+ )
+ }
+ if isinstance(value, (list, tuple)):
+ canonical_values = [_canonicalize_value(item, path=path) for item in value]
+ if path == ("dag", "tasks"):
+ return _canonicalize_keyed_list(canonical_values, _get_task_id, path)
+ if path == ("dag", "dag_dependencies"):
+ return _canonicalize_keyed_list(canonical_values, _get_dependency_key, path)
+ if path in _ORDER_INSENSITIVE_LIST_PATHS:
+ return _canonicalize_keyed_list(canonical_values, _get_string_key, path)
+ if _is_group_dependency_path(path):
+ return _sort_id_list(canonical_values)
+ return canonical_values
+ return value
+
+
+def _canonicalize_mapping_key(key: Any) -> str:
+ if isinstance(key, str) and isinstance(key, Enum):
+ return key.value
+ return str(key)
+
+
+def _canonicalize_keyed_list(
+ values: list[Any], key_getter: Callable[[Any], str], path: tuple[str, ...]
+) -> dict[str, Any]:
+ keyed_values: dict[str, Any] = {}
+ for value in values:
+ key = key_getter(value)
+ if key in keyed_values:
+ if path in _ORDER_INSENSITIVE_LIST_PATHS or (
+ path == ("dag", "dag_dependencies") and keyed_values[key] == value
+ ):
+ continue
+ raise ValueError(f"duplicate key {key!r} in /{'/'.join(path)}")
+ keyed_values[key] = value
+ return {key: keyed_values[key] for key in sorted(keyed_values)}
+
+
+def _get_task_id(task: Any) -> str:
+ if not isinstance(task, Mapping):
+ raise ValueError("task entry is not an object")
+ task_data = task.get("__var", task)
+ task_id = task_data.get("task_id") if isinstance(task_data, Mapping) else None
+ if not isinstance(task_id, str):
+ raise ValueError("task entry has no task_id")
+ return task_id
+
+
+def _get_string_key(value: Any) -> str:
+ if not isinstance(value, str):
+ raise ValueError("collection entry is not a string")
+ return value
+
+
+def _get_dependency_key(dependency: Any) -> str:
+ if not isinstance(dependency, MutableMapping):
+ raise ValueError("dependency entry is not an object")
+ # DagDependency.node_id identifies an edge by the components below alone, and ``label`` only
+ # mirrors state compared elsewhere (a task's _task_display_name, or an asset name already inside
+ # dependency_id). Keying on it turns a rename into one edge removed and another added; leaving
+ # it in the entry turns two entries on one edge into duplicates that fail canonicalization.
+ dependency.pop("label", None)
+ components = (
+ dependency.get("dependency_type"),
+ dependency.get("dependency_id"),
+ dependency.get("source"),
+ dependency.get("target"),
+ )
+ return json.dumps(components, ensure_ascii=False, separators=(",", ":"))
+
+
+def _collect_changes(
+ before: Any,
+ after: Any,
+ *,
+ path: tuple[str, ...],
+ collector: _ChangeCollector,
+) -> None:
+ if before is _MISSING and after is _MISSING:
+ return
+ if isinstance(before, Mapping) and isinstance(after, Mapping):
+ # Never branch on ``include_values`` here; see build_serialized_dag_diff.
+ if _is_task_mapping_path(path):
+ _collect_task_changes(before, after, path=path, collector=collector)
+ return
+ if _is_task_group_mapping_path(path):
+ _collect_public_field_changes(
+ before,
+ after,
+ path=path,
+ public_fields=_DIFF_V1_PUBLIC_TASK_GROUP_FIELDS,
+ collector=collector,
+ )
+ return
+ if path == _DEFAULT_ARGS_PATH:
+ _collect_default_args_changes(before, after, path=path, collector=collector)
+ return
+ if not _should_recurse_mapping(path):
+ if not _is_json_equal(before, after):
+ collector.add(path=path, operation="changed", before=before, after=after)
+ return
+ keys = sorted({str(key) for key in before} | {str(key) for key in after})
+ for key in keys:
+ before_value = before.get(key, _MISSING)
+ after_value = after.get(key, _MISSING)
+ _collect_changes(before_value, after_value, path=path + (key,), collector=collector)
+ if collector.is_truncated:
+ return
+ return
+
+ if isinstance(before, list) and isinstance(after, list):
+ if (
+ _is_task_group_child_path(path)
+ and len(before) == len(after) == 2
+ and before[0] == after[0] == "taskgroup"
+ and isinstance(before[1], Mapping)
+ and isinstance(after[1], Mapping)
+ ):
+ _collect_changes(before[1], after[1], path=path + ("1",), collector=collector)
+ return
+ if not _is_json_equal(before, after):
+ collector.add(path=path, operation="changed", before=before, after=after)
+ return
+
+ if before is _MISSING:
+ collector.add(path=path, operation="added", before=_MISSING, after=after)
+ elif after is _MISSING:
+ collector.add(path=path, operation="removed", before=before, after=_MISSING)
+ elif not _is_json_equal(before, after):
+ collector.add(path=path, operation="changed", before=before, after=after)
+
+
+def _collect_task_changes(
+ before: Mapping[str, Any],
+ after: Mapping[str, Any],
+ *,
+ path: tuple[str, ...],
+ collector: _ChangeCollector,
+) -> None:
+ if len(path) == 3:
+ _collect_changes(
+ before.get("__type", _MISSING),
+ after.get("__type", _MISSING),
+ path=path + ("__type",),
+ collector=collector,
+ )
+ if collector.is_truncated:
+ return
+ before, after = before["__var"], after["__var"]
+ public_fields = _DIFF_V1_PUBLIC_TASK_FIELDS - {"__type"}
+ if not (before.get("_is_mapped") and after.get("_is_mapped")):
+ # Comparing partial_kwargs field by field only means anything when both sides really are
+ # mapped. An unmapped operator may declare partial_kwargs a template field, making the
+ # mapping its own execution input: reading those keys as operator fields classified user
+ # data by whichever field a key collided with, and named a key the author chose. Keep it
+ # opaque, like any other field off the allowlist.
+ public_fields = public_fields - {"partial_kwargs"}
+ else:
+ public_fields = _DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS
+ _collect_public_field_changes(before, after, path=path, public_fields=public_fields, collector=collector)
+
+
+def _collect_default_args_changes(
+ before: Mapping[str, Any],
+ after: Mapping[str, Any],
+ *,
+ path: tuple[str, ...],
+ collector: _ChangeCollector,
+) -> None:
+ """Compare a Dag's default_args key by key so each key carries its own impact."""
+ if not (_is_encoded_default_args(before) and _is_encoded_default_args(after)):
+ # An envelope this version cannot read stays one opaque leaf: the allowlist was written for
+ # the keys of a dict envelope, so applying it to another shape could name something else.
+ if not _is_json_equal(before, after):
+ collector.add(path=path, operation="changed", before=before, after=after)
+ return
+ _collect_public_field_changes(
+ before["__var"],
+ after["__var"],
+ path=path,
+ public_fields=_DIFF_V1_PUBLIC_DEFAULT_ARGS_FIELDS,
+ collector=collector,
+ )
+
+
+def _is_encoded_default_args(value: Any) -> bool:
+ return (
+ isinstance(value, Mapping)
+ and value.get("__type") == "dict"
+ and isinstance(value.get("__var"), Mapping)
+ )
+
+
+def _collect_public_field_changes(
+ before: Mapping[str, Any],
+ after: Mapping[str, Any],
+ *,
+ path: tuple[str, ...],
+ public_fields: frozenset[str] | set[str],
+ collector: _ChangeCollector,
+) -> None:
+ """Walk allowlisted fields one by one and report everything else as a single change."""
+ keys = {str(key) for key in before} | {str(key) for key in after}
+ for key in sorted(keys & public_fields):
+ _collect_changes(
+ before.get(key, _MISSING),
+ after.get(key, _MISSING),
+ path=path + (key,),
+ collector=collector,
+ )
+ if collector.is_truncated:
+ return
+
+ before_custom_fields = {key: before[key] for key in before if key not in public_fields}
+ after_custom_fields = {key: after[key] for key in after if key not in public_fields}
+ if not _is_json_equal(before_custom_fields, after_custom_fields):
+ collector.add(
+ path=path + (_CUSTOM_TASK_FIELDS_PATH_COMPONENT,),
+ operation="changed",
+ before=before_custom_fields,
+ after=after_custom_fields,
+ )
+
+
+def _is_task_mapping_path(path: tuple[str, ...]) -> bool:
+ return path[:2] == ("dag", "tasks") and (
+ len(path) == 3 or (len(path) == 4 and path[3] == "partial_kwargs")
+ )
+
+
+class _TaskGroupPath(NamedTuple):
+ """One reading of the task group path grammar, shared by everything that walks it."""
+
+ field_index: int
+ id_positions: tuple[int, ...]
+ remainder: tuple[str, ...]
+
+
+def _classify_task_group_path(path: tuple[str, ...]) -> _TaskGroupPath | None:
+ """
+ Split a task group path into its nesting prefix and what follows the innermost group.
+
+ The grammar is ``dag/task_group ( /children/<group_id>/1 )* [ /children/<child_id> | /<field> ... ]``:
+ a child entry is the two-element ``["taskgroup", {...}]`` list the serializer writes, so the
+ literal ``1`` selects the nested group mapping inside it. ``field_index`` is where the innermost
+ group's own keys start, ``id_positions`` the group-id components a redacted path must mask, and
+ ``remainder`` the components from ``field_index`` on: ``()`` for the group mapping itself,
+ ``("children",)`` for its children mapping, and a two-element ``("children", child_id)`` for one
+ child entry. Returns ``None`` for a path that is not rooted at the Dag's task group.
+ """
+ if path[:2] != ("dag", "task_group"):
+ return None
+ index = 2
+ id_positions: list[int] = []
+ while len(path) >= index + 2 and path[index] == "children":
+ id_positions.append(index + 1)
+ if len(path) < index + 3 or path[index + 2] != "1":
+ break
+ index += 3
+ return _TaskGroupPath(index, tuple(id_positions), path[index:])
+
+
+def _is_task_group_mapping_path(path: tuple[str, ...]) -> bool:
+ task_group_path = _classify_task_group_path(path)
+ return task_group_path is not None and task_group_path.remainder == ()
+
+
+def _should_recurse_mapping(path: tuple[str, ...]) -> bool:
+ if path in _RECURSIVE_MAPPING_PATHS:
+ return True
+ task_group_path = _classify_task_group_path(path)
+ return task_group_path is not None and task_group_path.remainder == ("children",)
+
+
+def _is_task_group_child_path(path: tuple[str, ...]) -> bool:
+ task_group_path = _classify_task_group_path(path)
+ return (
+ task_group_path is not None
+ and len(task_group_path.remainder) == 2
+ and task_group_path.remainder[0] == "children"
+ )
+
+
+def _get_public_path(path: tuple[str, ...]) -> tuple[str, ...]:
+ if path and path[0] not in _DIFF_V1_PUBLIC_ROOT_FIELDS:
+ # Root is the one level no allowlisted walk covers, so an unrecognised section name would
+ # otherwise be the single path component a redacted caller receives verbatim.
+ return (_CUSTOM_TASK_FIELDS_PATH_COMPONENT,)
+ if (task_group_path := _classify_task_group_path(path)) is not None:
+ public_path = list(path)
+ for position in task_group_path.id_positions:
+ public_path[position] = "*"
+ remainder = task_group_path.remainder
+ if remainder and remainder[0] not in _DIFF_V1_PUBLIC_TASK_GROUP_FIELDS:
+ # Unreachable by construction (_collect_public_field_changes aggregates non-public group
+ # keys first), but truncating like the task branch stops a future caller leaking a key.
+ return (*public_path[: task_group_path.field_index], _CUSTOM_TASK_FIELDS_PATH_COMPONENT)
+ return tuple(public_path)
+ if len(path) >= 3 and path[:2] in _KEYED_COLLECTION_PATHS:
+ path = (*path[:2], "*", *path[3:])
+ if len(path) >= 4 and path[:2] == ("dag", "tasks"):
+ field_index = 4 if path[3] == "partial_kwargs" and len(path) >= 5 else 3
+ public_fields = (
+ _DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS if field_index == 4 else _DIFF_V1_PUBLIC_TASK_FIELDS
+ )
+ if path[field_index] not in public_fields:
+ return (*path[:field_index], _CUSTOM_TASK_FIELDS_PATH_COMPONENT)
+ return path
+
+
+def _format_path(path: tuple[str, ...]) -> str:
+ return "/" + "/".join(component.replace("~", "~0").replace("/", "~1") for component in path)
+
+
+class _JsonEncodingError(TypeError):
+ """Distinguish unsupported JSON values from comparison implementation errors."""
+
+
+def _serialize_canonical_json(value: Any) -> str:
+ try:
+ return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
+ except TypeError as error:
+ raise _JsonEncodingError from error
+
+
+def _is_json_equal(before: Any, after: Any) -> bool:
+ # Compare canonical JSON, not Python ==, so a JSON type change (1 vs 1.0, True vs 1,
+ # False vs 0) registers as a change instead of collapsing under Python equality.
+ return _serialize_canonical_json(before) == _serialize_canonical_json(after)
+
+
+def _get_digest(value: Any) -> str:
+ # SHA-256 provides a content fingerprint, not password protection.
+ # Stored payloads can carry escaped lone surrogates, which strict UTF-8 refuses to encode;
+ # surrogatepass keeps them digestible without altering the bytes of any other payload.
+ canonical_json = _serialize_canonical_json(value).encode("utf-8", errors="surrogatepass")
+ return f"sha256:{hashlib.sha256(canonical_json).hexdigest()}"
+
+
+def _get_category(path: tuple[str, ...]) -> DiffCategory:
+ # Matched case-sensitively: every classified name is a schema field name, so a differently
+ # cased key is a different key and belongs in "unknown" rather than borrowing a category.
+ if path and path[0] == "provenance":
+ return "provenance"
+ if path[:2] == ("dag", "tasks"):
+ field = path[3] if len(path) >= 4 else None
+ if field == "partial_kwargs" and len(path) >= 5:
+ field = path[4]
+ elif path[:2] == _DEFAULT_ARGS_PATH and len(path) >= 3:
+ # A default_args key is the operator field it will be applied to, so it classifies like one.
+ field = path[2]
+ elif path and path[0] == "dag":
+ field = path[1] if len(path) >= 2 else ""
+ if field == "task_group":
+ task_group_path = _classify_task_group_path(path)
+ if task_group_path is not None and task_group_path.remainder:
+ if task_group_path.remainder[0] in _DIFF_V1_TASK_GROUP_METADATA_FIELDS:
+ return "metadata"
+ return _DIFF_V1_DAG_FIELD_CATEGORIES.get(
+ field, _DIFF_V1_LEGACY_DAG_FIELD_CATEGORIES.get(field, "unknown")
+ )
+ else:
+ return "unknown"
+
+ # Unclassified task fields fall back to "task", over-reporting impact rather than under it.
+ if field is not None and "callback" in field:
+ return "callback"
+ if field in _DIFF_V1_TASK_ASSET_FIELDS:
+ return "asset"
+ if field in _DIFF_V1_TASK_PARAM_FIELDS:
+ return "param"
+ if field in _DIFF_V1_TASK_DEPENDENCY_FIELDS:
+ return "dependency"
+ if field in _DIFF_V1_TASK_METADATA_FIELDS:
+ return "metadata"
+ return "task"
+
+
+def _get_impact(category: DiffCategory) -> DiffImpact:
+ if category in {"provenance", "metadata", "authorization"}:
+ return category
+ if category in {"task", "dependency", "schedule", "param", "asset", "deadline", "callback"}:
+ return "execution"
+ return "unknown"
diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py b/airflow-core/src/airflow/serialization/serialized_objects.py
index 3a0a9278ae2..b58bb2505b0 100644
--- a/airflow-core/src/airflow/serialization/serialized_objects.py
+++ b/airflow-core/src/airflow/serialization/serialized_objects.py
@@ -131,6 +131,8 @@ log = logging.getLogger(__name__)
_CALLBACK_TYPES = ("execute", "failure", "success", "retry", "skipped")
_OPERATOR_CALLBACK_FIELDS = frozenset(f"on_{x}_callback" for x in _CALLBACK_TYPES)
_HAS_CALLBACK_FIELDS = frozenset(f"has_on_{x}_callback" for x in _CALLBACK_TYPES)
+_DAG_CALLBACK_FIELDS = frozenset({"has_on_success_callback", "has_on_failure_callback"})
+_OPERATOR_TIMEDELTA_FIELDS = frozenset({"retry_delay", "execution_timeout", "max_retry_delay"})
# Fields whose value must never be serialized: the object has no serializer, so it would
# fall back to str(obj) and leak a non-deterministic memory address (a new DagVersion every
# parse). Only a boolean ``has_<field>`` flag is stored; the live object is recovered by
@@ -1370,15 +1372,8 @@ class OperatorSerialization(DAGNode, BaseSerialization):
return op
@classmethod
- def _preprocess_encoded_operator(cls, encoded_op: dict[str, Any]) -> dict[str, Any]:
- """
- Preprocess and upgrade all field names for backward compatibility and consistency.
-
- This consolidates all field name transformations in one place:
- - Callback field renaming (on_*_callback -> has_on_*_callback)
- - Other field upgrades and renames
- - Field exclusions
- """
+ def _upgrade_encoded_operator(cls, encoded_op: dict[str, Any]) -> dict[str, Any]:
+ """Upgrade legacy field names while preserving serialized definition metadata."""
preprocessed = encoded_op.copy()
# Handle callback field renaming for backward compatibility
@@ -1402,6 +1397,13 @@ class OperatorSerialization(DAGNode, BaseSerialization):
if old_name in preprocessed:
preprocessed[new_name] = preprocessed.pop(old_name)
+ return preprocessed
+
+ @classmethod
+ def _preprocess_encoded_operator(cls, encoded_op: dict[str, Any]) -> dict[str, Any]:
+ """Upgrade operator fields and exclude metadata unused by runtime hydration."""
+ preprocessed = cls._upgrade_encoded_operator(encoded_op)
+
# Remove fields that shouldn't be processed
fields_to_exclude = {
"python_callable_name", # Only serves to detect function name changes
@@ -1630,7 +1632,7 @@ class OperatorSerialization(DAGNode, BaseSerialization):
return set(value) if value is not None else set()
elif field_name in _HAS_CALLBACK_FIELDS:
return bool(value)
- elif field_name in {"retry_delay", "execution_timeout", "max_retry_delay"}:
+ elif field_name in _OPERATOR_TIMEDELTA_FIELDS:
# Reuse existing timedelta deserialization logic
if value is not None:
return cls._deserialize_timedelta(value)
@@ -1776,10 +1778,9 @@ class DagSerialization(BaseSerialization):
serialized_dag["params"] = cls._serialize_params_dict(dag.params)
# has_on_*_callback are only stored if the value is True, as the default is False
- if dag.has_on_success_callback:
- serialized_dag["has_on_success_callback"] = True
- if dag.has_on_failure_callback:
- serialized_dag["has_on_failure_callback"] = True
+ for field in _DAG_CALLBACK_FIELDS:
+ if getattr(dag, field):
+ serialized_dag[field] = True
# TODO: Move this logic to a better place -- ideally before serializing contents of default_args.
# There is some duplication with this and SerializedBaseOperator.partial_kwargs serialization.
@@ -1902,10 +1903,9 @@ class DagSerialization(BaseSerialization):
tg.add(task)
# Set has_on_*_callbacks to True if they exist in Serialized blob as False is the default
- if "has_on_success_callback" in encoded_dag:
- dag.has_on_success_callback = True
- if "has_on_failure_callback" in encoded_dag:
- dag.has_on_failure_callback = True
+ for field in _DAG_CALLBACK_FIELDS:
+ if field in encoded_dag:
+ setattr(dag, field, True)
dag.deadline = encoded_dag.get("deadline")
diff --git a/airflow-core/tests/unit/models/test_dag_version.py b/airflow-core/tests/unit/models/test_dag_version.py
index 670a82b7a52..b6e39472fc7 100644
--- a/airflow-core/tests/unit/models/test_dag_version.py
+++ b/airflow-core/tests/unit/models/test_dag_version.py
@@ -16,23 +16,46 @@
# under the License.
from __future__ import annotations
+import copy
+import datetime
+import json
+import zlib
from datetime import timedelta
from unittest import mock
import pytest
-from sqlalchemy import func, select
+from sqlalchemy import delete, func, select, update
from airflow._shared.timezones import timezone
+from airflow.exceptions import DagVersionNotFound
from airflow.models.dag import DagModel
from airflow.models.dag_version import DagVersion
from airflow.models.dagbundle import DagBundleModel
+from airflow.models.dagrun import DagRun
+from airflow.models.deadline_alert import DeadlineAlert as DeadlineAlertModel
+from airflow.models.serialized_dag import SerializedDagModel
from airflow.providers.standard.operators.empty import EmptyOperator
-
+from airflow.sdk import DAG
+from airflow.sdk.definitions.callback import AsyncCallback
+from airflow.sdk.definitions.deadline import DeadlineAlert, DeadlineReference
+from airflow.serialization.dag_version_diff import DEFAULT_MAX_CHANGES, MAX_ALLOWED_CHANGES
+from airflow.serialization.encoders import encode_deadline_alert
+from airflow.utils.state import DagRunState
+from airflow.utils.types import DagRunTriggeredByType, DagRunType
+
+from tests_common.test_utils.asserts import assert_queries_count
from tests_common.test_utils.dag import sync_dag_to_db
-from tests_common.test_utils.db import clear_db_dag_bundles, clear_db_dags
+from tests_common.test_utils.db import clear_db_dag_bundles, clear_db_dags, clear_db_serialized_dags
pytestmark = pytest.mark.db_test
+# Valid JSON nested past the interpreter's recursion limit, so json.loads raises RecursionError.
+_DEEPLY_NESTED_JSON = "[" * 20_000 + "]" * 20_000
+
+
+async def _handle_deadline(context, **kwargs):
+ pass
+
class TestDagVersion:
def setup_method(self):
@@ -249,3 +272,398 @@ class TestResolveVersionData:
from airflow.models.dag_version import _resolve_version_data
assert _resolve_version_data(dag_version, bundle_version) == expected
+
+
+class TestDagVersionGetDiff:
+ @pytest.fixture
+ def dag_id(self, dag_maker, session):
+ clear_db_dags()
+ clear_db_serialized_dags()
+ dag_id = "version_diff_dag"
+ for version_number in range(1, 3):
+ with dag_maker(dag_id, session=session, bundle_version=f"commit{version_number}"):
+ for task_number in range(version_number):
+ EmptyOperator(task_id=f"task{task_number + 1}")
+ dag_maker.create_dagrun(
+ run_id=f"run{version_number}",
+ logical_date=datetime.datetime(2020, 1, version_number, tzinfo=datetime.timezone.utc),
+ session=session,
+ )
+ session.commit()
+ # Read versions back from the DB the way the API and CLI callers do, so
+ # serialized payloads carry plain JSON keys rather than in-memory Encoding enums.
+ session.expunge_all()
+ return dag_id
+
+ @pytest.fixture
+ def create_deadline_versions(self, session):
+ dag_id = "version_diff_deadlines"
+ bundle_name = "version_diff_deadlines"
+
+ def create_versions(
+ *, name="completion", target_interval=timedelta(minutes=5), target_callback_kwargs=None
+ ):
+ definitions = []
+ for version_number in (1, 2):
+ deadline = DeadlineAlert(
+ name=name,
+ reference=DeadlineReference.DAGRUN_QUEUED_AT,
+ interval=timedelta(minutes=5) if version_number == 1 else target_interval,
+ callback=AsyncCallback(
+ _handle_deadline,
+ kwargs=(
+ target_callback_kwargs
+ if version_number == 2 and target_callback_kwargs is not None
+ else {"message": "original"}
+ ),
+ ),
+ )
+ definitions.append(encode_deadline_alert(deadline))
+ with DAG(dag_id, deadline=deadline) as dag:
+ for task_number in range(version_number):
+ EmptyOperator(task_id=f"task{task_number + 1}")
+ scheduler_dag = sync_dag_to_db(dag, bundle_name=bundle_name, session=session)
+ logical_date = datetime.datetime(2020, 1, version_number, tzinfo=datetime.timezone.utc)
+ scheduler_dag.create_dagrun(
+ run_id=f"run{version_number}",
+ run_after=logical_date,
+ state=DagRunState.QUEUED,
+ logical_date=logical_date,
+ data_interval=(logical_date, logical_date),
+ triggered_by=DagRunTriggeredByType.TEST,
+ run_type=DagRunType.MANUAL,
+ session=session,
+ )
+ session.commit()
+ session.expunge_all()
+ return dag_id, definitions
+
+ yield create_versions
+
+ session.rollback()
+ session.execute(delete(DagRun).where(DagRun.dag_id == dag_id))
+ session.execute(delete(DagModel).where(DagModel.dag_id == dag_id))
+ session.execute(delete(DagBundleModel).where(DagBundleModel.name == bundle_name))
+ session.commit()
+
+ def test_reports_observed_changes_between_versions(self, dag_id, session):
+ result = DagVersion.get_diff(dag_id, 1, 2, session=session)
+
+ assert result["mode"] == "observed_state"
+ paths = {change["path"]: change for change in result["changes"]}
+ assert paths["/dag/tasks/*"]["operation"] == "added"
+ assert "after_digest" not in paths["/dag/tasks/*"]
+ assert paths["/provenance/bundle_version"]["category"] == "provenance"
+ assert "source" not in result
+ assert result["values"] == {"status": "unavailable"}
+
+ @pytest.mark.parametrize("values_status", ["unavailable", "available"])
+ def test_reports_no_changes_comparing_a_version_with_itself(self, dag_id, session, values_status):
+ result = DagVersion.get_diff(dag_id, 2, 2, values_status=values_status, session=session)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+ assert result["truncated"] is False
+
+ def test_raises_for_missing_version(self, dag_id, session):
+ with pytest.raises(DagVersionNotFound, match="version_number: `3`"):
+ DagVersion.get_diff(dag_id, 1, 3, session=session)
+
+ @pytest.mark.parametrize("version_numbers", [(0, 2), (1, 0)])
+ def test_rejects_non_positive_version_numbers(self, dag_id, session, version_numbers):
+ with pytest.raises(ValueError, match="Dag version numbers must be positive integers"):
+ DagVersion.get_diff(dag_id, *version_numbers, session=session)
+
+ @pytest.mark.parametrize(
+ ("version_numbers", "expected_message"),
+ [
+ (("1", 2), "Dag version numbers must be integers"),
+ ((1, 1.5), "Dag version numbers must be integers"),
+ ((True, 2), "Dag version numbers must be integers"),
+ ((1, 10**30), "Dag version numbers must be positive integers not exceeding"),
+ ],
+ )
+ def test_rejects_unusable_version_numbers_before_querying(
+ self, dag_id, session, version_numbers, expected_message
+ ):
+ with assert_queries_count(0):
+ with pytest.raises(ValueError, match=expected_message):
+ DagVersion.get_diff(dag_id, *version_numbers, session=session)
+
+ @pytest.mark.parametrize(
+ ("max_changes", "expected_message"),
+ [
+ (0, "max_changes must be a positive integer"),
+ (1.5, "max_changes must be a positive integer"),
+ (MAX_ALLOWED_CHANGES + 1, f"max_changes must not exceed {MAX_ALLOWED_CHANGES}"),
+ ],
+ )
+ def test_rejects_invalid_change_bound_before_querying(
+ self, dag_id, session, max_changes, expected_message
+ ):
+ with assert_queries_count(0):
+ with pytest.raises(ValueError, match=expected_message):
+ DagVersion.get_diff(dag_id, 1, 2, max_changes=max_changes, session=session)
+
+ @mock.patch("airflow.serialization.dag_version_diff.build_serialized_dag_diff", autospec=True)
+ def test_applies_the_default_change_bound_when_unset(self, mock_build_diff, dag_id, session):
+ DagVersion.get_diff(dag_id, 1, 2, max_changes=None, session=session)
+
+ assert mock_build_diff.call_args.kwargs["max_changes"] == DEFAULT_MAX_CHANGES
+
+ def test_marks_values_available_when_allowed(self, dag_id, session):
+ result = DagVersion.get_diff(dag_id, 1, 2, values_status="available", session=session)
+
+ assert result["values"] == {"status": "available"}
+ assert "source" not in result
+ assert any("after_value" in change for change in result["changes"])
+ assert any(change["path"] == "/dag/tasks/task2" for change in result["changes"])
+
+ def test_marks_values_unavailable_when_status_denied(self, dag_id, session):
+ result = DagVersion.get_diff(dag_id, 1, 2, values_status="unavailable", session=session)
+
+ assert result["mode"] == "observed_state"
+ assert any(change["path"] == "/dag/tasks/*" for change in result["changes"])
+ assert all(
+ "before_digest" not in change
+ and "after_digest" not in change
+ and "before_value" not in change
+ and "after_value" not in change
+ for change in result["changes"]
+ )
+ assert result["values"] == {"status": "unavailable"}
+
+ @pytest.mark.parametrize("version_number", [1, 2])
+ def test_compares_json_string_serialized_data(self, dag_id, session, version_number):
+ expected = DagVersion.get_diff(dag_id, 1, 2, values_status="available", session=session)
+ serialized_dag = DagVersion.get_version(dag_id, version_number, session=session).serialized_dag
+ session.execute(
+ update(SerializedDagModel)
+ .where(SerializedDagModel.id == serialized_dag.id)
+ .values(_data=json.dumps(serialized_dag.data), _data_compressed=None)
+ )
+ session.commit()
+ session.expunge_all()
+
+ stored_data = DagVersion.get_version(dag_id, version_number, session=session).serialized_dag.data
+ assert isinstance(stored_data, str)
+ assert DagVersion.get_diff(dag_id, 1, 2, values_status="available", session=session) == expected
+
+ @mock.patch("airflow.serialization.dag_version_diff.build_serialized_dag_diff", autospec=True)
+ def test_marks_diff_unavailable_for_invalid_serialized_json(self, mock_build_diff, dag_id, session):
+ base = DagVersion.get_version(dag_id, 1, session=session).serialized_dag
+ session.execute(
+ update(SerializedDagModel)
+ .where(SerializedDagModel.id == base.id)
+ .values(_data="invalid JSON", _data_compressed=None)
+ )
+ session.commit()
+ session.expunge_all()
+
+ result = DagVersion.get_diff(dag_id, 1, 2, values_status="available", session=session)
+
+ mock_build_diff.assert_not_called()
+ assert result["mode"] == "unavailable"
+ assert result["unavailable_reason"] == "serialized_dag_decode_failed"
+ assert result["changes"] == []
+ assert result["values"] == {"status": "unavailable"}
+
+ @pytest.mark.parametrize(
+ "stored_columns",
+ [
+ pytest.param(
+ {"_data": None, "_data_compressed": zlib.compress(b"invalid JSON")},
+ id="compressed-invalid-json",
+ ),
+ pytest.param(
+ {"_data": None, "_data_compressed": b"not a zlib stream"},
+ id="compressed-undecompressable",
+ ),
+ pytest.param(
+ {"_data": None, "_data_compressed": zlib.compress(_DEEPLY_NESTED_JSON.encode())},
+ id="compressed-deeply-nested",
+ ),
+ pytest.param(
+ {"_data": _DEEPLY_NESTED_JSON, "_data_compressed": None},
+ id="uncompressed-deeply-nested",
+ ),
+ ],
+ )
+ @mock.patch("airflow.serialization.dag_version_diff.build_serialized_dag_diff", autospec=True)
+ def test_marks_diff_unavailable_for_undecodable_stored_payload(
+ self, mock_build_diff, stored_columns, dag_id, session
+ ):
+ base = DagVersion.get_version(dag_id, 1, session=session).serialized_dag
+ session.execute(
+ update(SerializedDagModel).where(SerializedDagModel.id == base.id).values(**stored_columns)
+ )
+ session.commit()
+ session.expunge_all()
+
+ result = DagVersion.get_diff(dag_id, 1, 2, values_status="available", session=session)
+
+ mock_build_diff.assert_not_called()
+ assert result["mode"] == "unavailable"
+ assert result["unavailable_reason"] == "serialized_dag_decode_failed"
+ assert result["changes"] == []
+ assert result["values"] == {"status": "unavailable"}
+
+ @pytest.mark.parametrize("missing_version", [1, 2])
+ def test_marks_diff_unavailable_when_serialized_dag_missing(self, dag_id, session, missing_version):
+ version = DagVersion.get_version(dag_id, missing_version, session=session)
+ session.execute(delete(SerializedDagModel).where(SerializedDagModel.dag_version_id == version.id))
+ session.commit()
+ session.expunge_all()
+
+ result = DagVersion.get_diff(dag_id, 1, 2, values_status="available", session=session)
+
+ assert result["mode"] == "unavailable"
+ assert result["unavailable_reason"] == "serialized_dag_missing"
+ assert result["serialized_dag_schema_versions"] == {
+ "base": None if missing_version == 1 else 3,
+ "target": None if missing_version == 2 else 3,
+ }
+ assert result["changes"] == []
+ assert result["values"] == {"status": "unavailable"}
+
+ @pytest.mark.parametrize("values_status", ["unavailable", "available"])
+ @pytest.mark.parametrize("changed_field", [None, "interval", "callback"])
+ def test_compares_stored_deadline_definitions(
+ self, create_deadline_versions, session, values_status, changed_field
+ ):
+ dag_id, definitions = create_deadline_versions(
+ target_interval=timedelta(minutes=10 if changed_field == "interval" else 5),
+ target_callback_kwargs={"message": "changed"} if changed_field == "callback" else None,
+ )
+ versions = [DagVersion.get_version(dag_id, number, session=session) for number in (1, 2)]
+ stored_payloads = [copy.deepcopy(version.serialized_dag.data) for version in versions]
+ assert stored_payloads[0]["dag"]["deadline"] != stored_payloads[1]["dag"]["deadline"]
+ assert all(isinstance(payload["dag"]["deadline"][0], str) for payload in stored_payloads)
+
+ with assert_queries_count(2):
+ result = DagVersion.get_diff(dag_id, 1, 2, values_status=values_status, session=session)
+
+ assert result["mode"] == "observed_state"
+ deadline_changes = [change for change in result["changes"] if change["path"] == "/dag/deadline"]
+ if changed_field is None:
+ assert deadline_changes == []
+ else:
+ assert len(deadline_changes) == 1
+ change = deadline_changes[0]
+ assert change["category"] == "deadline"
+ assert change["impact"] == "execution"
+ if values_status == "available":
+ assert change["before_value"] == [definitions[0]]
+ assert change["after_value"] == [definitions[1]]
+ assert [version.serialized_dag.data for version in versions] == stored_payloads
+
+ @pytest.mark.parametrize(
+ ("spelling", "expected_reason"),
+ [
+ ("upper", None),
+ ("unhyphenated", None),
+ ("urn", None),
+ ("malformed", "deadline_alert_missing"),
+ ],
+ )
+ def test_resolves_deadline_alerts_whatever_the_stored_uuid_spelling(
+ self, create_deadline_versions, session, spelling, expected_reason
+ ):
+ dag_id, _ = create_deadline_versions()
+ base = DagVersion.get_version(dag_id, 1, session=session).serialized_dag
+ stored_data = copy.deepcopy(base.data)
+ (alert_id,) = stored_data["dag"]["deadline"]
+ stored_data["dag"]["deadline"] = [
+ {
+ "upper": alert_id.upper(),
+ "unhyphenated": alert_id.replace("-", ""),
+ "urn": f"urn:uuid:{alert_id}",
+ "malformed": "not-a-uuid",
+ }[spelling]
+ ]
+ session.execute(
+ update(SerializedDagModel)
+ .where(SerializedDagModel.id == base.id)
+ .values(_data=stored_data, _data_compressed=None)
+ )
+ session.commit()
+ session.expunge_all()
+
+ result = DagVersion.get_diff(dag_id, 1, 2, values_status="available", session=session)
+
+ if expected_reason is None:
+ assert result["mode"] == "observed_state"
+ assert [change for change in result["changes"] if change["path"] == "/dag/deadline"] == []
+ else:
+ assert result["mode"] == "unavailable"
+ assert result["unavailable_reason"] == expected_reason
+
+ @pytest.mark.parametrize("inline_format", ["plain", "wrapped", "legacy"])
+ @pytest.mark.parametrize("changed_field", [None, "interval", "callback"])
+ def test_compares_inline_and_referenced_deadline_definitions(
+ self, create_deadline_versions, session, inline_format, changed_field
+ ):
+ dag_id, definitions = create_deadline_versions(
+ name=None if inline_format == "legacy" else "completion",
+ target_interval=timedelta(minutes=10 if changed_field == "interval" else 5),
+ target_callback_kwargs={"message": "changed"} if changed_field == "callback" else None,
+ )
+ base = DagVersion.get_version(dag_id, 1, session=session).serialized_dag
+ inline_data = copy.deepcopy(base.data)
+ definition = copy.deepcopy(definitions[0])
+ if inline_format == "legacy":
+ definition.pop("name")
+ definition["interval"] = 300.0
+ if inline_format != "plain":
+ definition = {"__type": "deadline_alert", "__var": definition}
+ inline_data["dag"]["deadline"] = [definition]
+ session.execute(
+ update(SerializedDagModel)
+ .where(SerializedDagModel.id == base.id)
+ .values(_data=inline_data, _data_compressed=None)
+ )
+ session.commit()
+ session.expunge_all()
+
+ result = DagVersion.get_diff(dag_id, 1, 2, values_status="available", session=session)
+
+ assert result["mode"] == "observed_state"
+ deadline_changes = [change for change in result["changes"] if change["path"] == "/dag/deadline"]
+ if changed_field is None:
+ assert deadline_changes == []
+ else:
+ assert len(deadline_changes) == 1
+ assert deadline_changes[0]["before_value"] == [definitions[0]]
+ assert deadline_changes[0]["after_value"] == [definitions[1]]
+ assert DagVersion.get_version(dag_id, 1, session=session).serialized_dag.data == inline_data
+
+ @pytest.mark.parametrize("missing_version", [1, 2])
+ @pytest.mark.parametrize("same_version", [False, True])
+ @mock.patch("airflow.serialization.dag_version_diff.build_serialized_dag_diff", autospec=True)
+ def test_marks_diff_unavailable_when_deadline_missing(
+ self, mock_build_diff, create_deadline_versions, session, missing_version, same_version
+ ):
+ dag_id, _ = create_deadline_versions()
+ serialized_dag = DagVersion.get_version(dag_id, missing_version, session=session).serialized_dag
+ session.execute(
+ delete(DeadlineAlertModel).where(DeadlineAlertModel.serialized_dag_id == serialized_dag.id)
+ )
+ session.commit()
+ session.expunge_all()
+ version_numbers = (missing_version, missing_version) if same_version else (1, 2)
+
+ result = DagVersion.get_diff(dag_id, *version_numbers, values_status="available", session=session)
+
+ mock_build_diff.assert_not_called()
+ assert result["mode"] == "unavailable"
+ assert result["unavailable_reason"] == "deadline_alert_missing"
+ assert result["serialized_dag_schema_versions"] == {"base": 3, "target": 3}
+ assert result["changes"] == []
+ assert result["truncated"] is False
+ assert result["values"] == {"status": "unavailable"}
+ assert "source" not in result
+
+ @pytest.mark.parametrize("invalid_status", ["", "redacted"])
+ def test_rejects_invalid_values_status(self, dag_id, session, invalid_status):
+ with pytest.raises(ValueError, match="values_status must be one of"):
+ DagVersion.get_diff(dag_id, 1, 2, values_status=invalid_status, session=session)
diff --git a/airflow-core/tests/unit/serialization/test_dag_serialization.py b/airflow-core/tests/unit/serialization/test_dag_serialization.py
index cf9c5c34c17..279a7bb02a1 100644
--- a/airflow-core/tests/unit/serialization/test_dag_serialization.py
+++ b/airflow-core/tests/unit/serialization/test_dag_serialization.py
@@ -1713,6 +1713,33 @@ class TestStringifiedDAGs:
serialized_op = OperatorSerialization.deserialize_operator(blob)
assert serialized_op.downstream_task_ids == {"foo"}
+ def test_operator_upgrade_preserves_definition_metadata(self):
+ encoded_op = {
+ "task_id": "extract",
+ "_downstream_task_ids": ["load"],
+ "downstream_task_ids": [],
+ "on_failure_callback": False,
+ "has_on_failure_callback": True,
+ "python_callable_name": "process",
+ "label": "Extract data",
+ }
+ original = copy.deepcopy(encoded_op)
+ upgraded = {
+ "task_id": "extract",
+ "downstream_task_ids": ["load"],
+ "has_on_failure_callback": False,
+ "python_callable_name": "process",
+ "label": "Extract data",
+ }
+
+ assert OperatorSerialization._upgrade_encoded_operator(encoded_op) == upgraded
+ assert OperatorSerialization._preprocess_encoded_operator(encoded_op) == {
+ "task_id": "extract",
+ "downstream_task_ids": ["load"],
+ "has_on_failure_callback": False,
+ }
+ assert encoded_op == original
+
def test_task_resources(self):
"""
Test task resources serialization/deserialization.
diff --git a/airflow-core/tests/unit/serialization/test_dag_version_diff.py b/airflow-core/tests/unit/serialization/test_dag_version_diff.py
new file mode 100644
index 00000000000..c3b027b6e31
--- /dev/null
+++ b/airflow-core/tests/unit/serialization/test_dag_version_diff.py
@@ -0,0 +1,3552 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import sys
+from contextlib import ExitStack
+from datetime import datetime, timedelta, timezone
+from typing import Any
+from unittest import mock
+
+import pytest
+from jsonschema import Draft7Validator
+
+from airflow.providers.standard.operators.bash import BashOperator
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.providers.standard.operators.python import BranchPythonOperator, PythonOperator
+from airflow.providers.standard.operators.trigger_dagrun import TriggerDagRunOperator
+from airflow.providers.standard.sensors.date_time import DateTimeSensor, DateTimeSensorAsync
+from airflow.providers.standard.sensors.external_task import ExternalTaskSensor
+from airflow.sdk import DAG, Asset, ExceptionRetryPolicy, Param, TaskGroup, task as sdk_task, task_group
+from airflow.sdk.bases.operator import BaseOperator
+from airflow.sdk.definitions.operator_resources import Resources
+from airflow.serialization import dag_version_diff
+from airflow.serialization.dag_version_diff import (
+ _DIFF_V1_DAG_FIELD_CATEGORIES,
+ _DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS,
+ _DIFF_V1_PUBLIC_ROOT_FIELDS,
+ _DIFF_V1_PUBLIC_TASK_FIELDS,
+ _DIFF_V1_PUBLIC_TASK_GROUP_FIELDS,
+ _DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS,
+ _DIFF_V1_TASK_GROUP_METADATA_FIELDS,
+ MAX_ALLOWED_CHANGES,
+ SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS,
+ build_serialized_dag_diff,
+ build_unavailable_dag_diff,
+)
+from airflow.serialization.json_schema import load_dag_schema_dict
+from airflow.serialization.serialized_objects import (
+ _DAG_CALLBACK_FIELDS,
+ _OPERATOR_TIMEDELTA_FIELDS,
+ BaseSerialization,
+ DagSerialization,
+ OperatorSerialization,
+)
+
+
+def _build_payload(
+ *,
+ tasks: list[dict],
+ tags: list[str] | None = None,
+ schedule: str = "daily",
+ dependencies: list[dict] | None = None,
+) -> dict:
+ return {
+ "__version": 3,
+ "dag": {
+ "dag_id": "example",
+ "schedule": schedule,
+ "tags": tags or [],
+ "tasks": [
+ {
+ "__type": "airflow.providers.standard.operators.empty.EmptyOperator",
+ "__var": task,
+ }
+ for task in tasks
+ ],
+ "dag_dependencies": dependencies or [],
+ },
+ }
+
+
+def _serialize_dag(dag: DAG) -> dict[str, Any]:
+ return json.loads(json.dumps(DagSerialization.to_dict(dag)))
+
+
+def _get_task(payload: dict[str, Any]) -> dict[str, Any]:
+ return payload["dag"]["tasks"][0]["__var"]
+
+
+def _build_field_sweep_payload() -> dict[str, Any]:
+ """Serialize one Dag covering the operator shapes that emit distinct task field sets."""
+ # The drift guard below is only as complete as the shapes built here: a new serializer field
+ # emitted solely by a shape this sweep does not construct still lands in custom_fields
+ # unnoticed, so a new operator shape needs a task here, not just a new allowlist entry.
+ with DAG("field_sweep", schedule=None) as dag:
+ EmptyOperator(task_id="empty")
+ BranchPythonOperator(task_id="branch", python_callable=_return_private_before)
+ PythonOperator(task_id="python", python_callable=_return_private_before, op_kwargs={"a": 1})
+ DateTimeSensor(task_id="rescheduling", target_time="{{ ts }}", mode="reschedule")
+ DateTimeSensorAsync(task_id="deferring", target_time="{{ ts }}")
+ BashOperator(
+ task_id="configured",
+ bash_command="echo unchanged",
+ run_as_user="service",
+ email="owner@example.org",
+ retry_policy=ExceptionRetryPolicy(rules=[]),
+ executor_config={"key": "value"},
+ resources={"cpus": 1},
+ outlets=[Asset("s3://sweep")],
+ )
+ BashOperator(task_id="setup", bash_command="echo unchanged").as_setup()
+ BashOperator(task_id="teardown", bash_command="echo unchanged").as_teardown()
+ BashOperator.partial(task_id="mapped", run_as_user="service").expand(bash_command=["one", "two"])
+ sdk_task(task_id="decorated")(_return_private_before).expand(private_argument=["one", "two"])
+ sdk_task.stub(task_id="stubbed")(_declare_private_argument)("private_value")
+ with TaskGroup("group"):
+ BashOperator(task_id="grouped", bash_command="echo unchanged")
+ return _serialize_dag(dag)
+
+
+def _collect_emitted_task_fields(payload: dict[str, Any]) -> tuple[set[str], set[str], set[str]]:
+ emitted: set[str] = set()
+ emitted_partial: set[str] = set()
+ template_fields: set[str] = set()
+ for entry in payload["dag"]["tasks"]:
+ task_fields = entry["__var"]
+ partial_kwargs = task_fields.get("partial_kwargs", {})
+ emitted |= task_fields.keys()
+ emitted_partial |= partial_kwargs.keys()
+ template_fields |= set(task_fields.get("template_fields", []))
+ template_fields |= set(partial_kwargs.get("template_fields", []))
+ return emitted, emitted_partial, template_fields
+
+
+def test_diff_v1_task_field_policy_tracks_serializer_schema() -> None:
+ schema_fields = frozenset(load_dag_schema_dict()["definitions"]["operator"]["properties"])
+ emitted, emitted_partial, template_fields = _collect_emitted_task_fields(_build_field_sweep_payload())
+ # definitions.operator is additionalProperties: true, so comparing the allowlist against the
+ # schema alone cannot see a field the serializer writes and the schema never declares. Derive
+ # those from a real payload instead of listing them, or they get buried in custom_fields unnoticed.
+ serializer_only_fields = emitted - schema_fields - template_fields
+ # Template field names are operator-defined and unbounded, so the allowlist cannot enumerate
+ # them; naming one in a redacted path would also disclose what template_fields itself withholds.
+ exempt_fields = _DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS | template_fields
+
+ assert emitted <= (_DIFF_V1_PUBLIC_TASK_FIELDS - {"__type"}) | exempt_fields
+ assert emitted_partial <= _DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS | exempt_fields
+ assert schema_fields == (
+ (_DIFF_V1_PUBLIC_TASK_FIELDS - {"__type"} - serializer_only_fields)
+ | _DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS
+ )
+ assert _DIFF_V1_PUBLIC_TASK_FIELDS - schema_fields == {"__type"} | serializer_only_fields
+ assert not (_DIFF_V1_PUBLIC_TASK_FIELDS & _DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS)
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_names_sensor_reschedule_mode_changes(include_values):
+ payloads = []
+ for mode in ("poke", "reschedule"):
+ with DAG("sensor_mode", schedule=None) as dag:
+ DateTimeSensor(task_id="private_task", target_time="{{ ts }}", mode=mode)
+ payloads.append(_serialize_dag(dag))
+ hydrated = [
+ DagSerialization.from_dict(copy.deepcopy(payload)).task_dict["private_task"].reschedule
+ for payload in payloads
+ ]
+ assert hydrated == [False, True]
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ task_id = "private_task" if include_values else "*"
+ assert [change["path"] for change in result["changes"]] == [f"/dag/tasks/{task_id}/reschedule"]
+ assert result["changes"][0]["impact"] == "execution"
+
+
+def _return_private_before(private_argument=None):
+ return private_argument
+
+
+def _return_private_after(private_argument=None):
+ return private_argument
+
+
+def _declare_private_argument(private_argument: str): ...
+
+
+def _build_serializer_only_field_payload(field: str, is_mapped: bool, *, after: bool) -> dict[str, Any]:
+ with DAG("serializer_only_fields", schedule=None) as dag:
+ if field == "python_callable_name":
+ python_callable = _return_private_after if after else _return_private_before
+ if is_mapped:
+ PythonOperator.partial(task_id="private_task", python_callable=python_callable).expand(
+ op_kwargs=[{"private_argument": "private_value"}]
+ )
+ else:
+ PythonOperator(task_id="private_task", python_callable=python_callable)
+ elif field == "expand_input":
+ BashOperator.partial(task_id="private_task").expand(
+ bash_command=["private_before", "private_after"] if after else ["private_before"]
+ )
+ elif field == "op_kwargs_expand_input":
+ sdk_task(task_id="private_task")(_return_private_before).expand(
+ private_argument=["private_before", "private_after"] if after else ["private_before"]
+ )
+ else:
+ values: dict[str, tuple[Any, Any]] = {
+ "resources": ({"cpus": 1}, {"cpus": 2}),
+ "run_as_user": ("private_before", "private_after"),
+ "email": ("private_before@example.org", "private_after@example.org"),
+ "has_retry_policy": (None, ExceptionRetryPolicy(rules=[])),
+ }
+ constructor_field = "retry_policy" if field == "has_retry_policy" else field
+ kwargs = {} if field == "_operator_name" else {constructor_field: values[field][after]}
+ operator = (
+ BashOperator.partial(task_id="private_task", **kwargs).expand(bash_command=["echo unchanged"])
+ if is_mapped
+ else BashOperator(task_id="private_task", bash_command="echo unchanged", **kwargs)
+ )
+ if field == "_operator_name":
+ setattr(
+ operator,
+ "_operator_name" if is_mapped else "custom_operator_name",
+ "private_after" if after else "private_before",
+ )
+ return _serialize_dag(dag)
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize(
+ ("field", "is_mapped"),
+ [
+ ("python_callable_name", False),
+ ("python_callable_name", True),
+ ("expand_input", True),
+ ("op_kwargs_expand_input", True),
+ ("resources", False),
+ ("resources", True),
+ ("run_as_user", False),
+ ("run_as_user", True),
+ ("email", False),
+ ("email", True),
+ ("_operator_name", False),
+ ("_operator_name", True),
+ ("has_retry_policy", False),
+ ("has_retry_policy", True),
+ ],
+)
+def test_build_diff_exposes_serializer_only_field_names_without_private_values(
+ field, is_mapped, include_values
+):
+ base = _build_serializer_only_field_payload(field, is_mapped, after=False)
+ target = _build_serializer_only_field_payload(field, is_mapped, after=True)
+ for payload in (base, target):
+ emitted_fields = _get_task(payload)
+ redacted_fields = _DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS | set(
+ emitted_fields.get("template_fields", [])
+ )
+ assert emitted_fields.keys() <= (_DIFF_V1_PUBLIC_TASK_FIELDS - {"__type"}) | redacted_fields
+ assert emitted_fields.get("partial_kwargs", {}).keys() <= (
+ _DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS | redacted_fields
+ )
+ task_fields = _get_task(target)
+ in_partial_kwargs = is_mapped and field in {
+ "python_callable_name",
+ "resources",
+ "run_as_user",
+ "email",
+ "has_retry_policy",
+ }
+ stored_fields = task_fields["partial_kwargs"] if in_partial_kwargs else task_fields
+ assert field in stored_fields
+ originals = copy.deepcopy((base, target))
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ task_id = "private_task" if include_values else "*"
+ field_path = f"partial_kwargs/{field}" if in_partial_kwargs else field
+ assert change["path"] == f"/dag/tasks/{task_id}/{field_path}"
+ expected_category = "metadata" if field == "_operator_name" else "task"
+ assert change["category"] == expected_category
+ assert change["impact"] == ("metadata" if field == "_operator_name" else "execution")
+ if include_values:
+ assert change["after_value"] == stored_fields[field]
+ assert change["before_digest"] != change["after_digest"]
+ else:
+ assert "private_" not in json.dumps(result)
+ assert not {"before_value", "after_value", "before_digest", "after_digest"} & change.keys()
+ assert (base, target) == originals
+
+
+def test_diff_v1_task_group_field_policy_tracks_serializer_schema() -> None:
+ schema_fields = set(load_dag_schema_dict()["definitions"]["task_group"]["properties"])
+
+ # A mapped group serializes expand_input, which the task_group definition does not declare.
+ assert schema_fields == _DIFF_V1_PUBLIC_TASK_GROUP_FIELDS - {"expand_input"}
+ assert _DIFF_V1_TASK_GROUP_METADATA_FIELDS <= _DIFF_V1_PUBLIC_TASK_GROUP_FIELDS
+
+
+def test_diff_v1_dag_field_policy_tracks_serializer_schema() -> None:
+ schema_fields = set(load_dag_schema_dict()["definitions"]["dag"]["properties"])
+
+ assert schema_fields == _DIFF_V1_DAG_FIELD_CATEGORIES.keys()
+
+
+def test_diff_v1_root_field_policy_tracks_serializer_schema() -> None:
+ schema_fields = set(load_dag_schema_dict()["allOf"][0]["properties"])
+
+ # __version and client_defaults are folded in by canonicalization, which adds provenance.
+ assert (schema_fields - {"__version", "client_defaults"}) | {"provenance"} == _DIFF_V1_PUBLIC_ROOT_FIELDS
+
+
+def test_dag_callback_fields_track_serializer_schema() -> None:
+ schema_fields = load_dag_schema_dict()["definitions"]["dag"]["properties"]
+
+ assert {
+ field for field in schema_fields if field.startswith("has_on_") and field.endswith("_callback")
+ } == _DAG_CALLBACK_FIELDS
+
+
+def test_operator_timedelta_fields_track_serializer_schema() -> None:
+ schema_fields = load_dag_schema_dict()["definitions"]["operator"]["properties"]
+
+ assert {
+ field for field, schema in schema_fields.items() if schema.get("$ref") == "#/definitions/timedelta"
+ } == _OPERATOR_TIMEDELTA_FIELDS
+
+
+def test_json_value_encoding_tracks_serializer() -> None:
+ value = {"nested": [None, True, 7, 1.5, "plain", {"__type": "dict", "__var": {"cpu": 1}}]}
+ assert dag_version_diff._encode_json_value(value) == BaseSerialization.serialize(value)
+
+
+def test_supported_schema_versions_track_serializer() -> None:
+ # A serializer bump without a matching conversion in _canonicalize_payload_v1 would otherwise
+ # turn every stored Dag in a deployment into unsupported_serialized_dag_schema_version.
+ supported_by_serializer = frozenset(range(1, DagSerialization.SERIALIZER_VERSION + 1))
+
+ assert supported_by_serializer == SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS
+
+
+def test_task_field_categories_are_reachable() -> None:
+ # _get_category only reaches these after a non-public field has been aggregated under
+ # custom_fields, so a name outside the allowlist would be dead.
+ classified = (
+ dag_version_diff._DIFF_V1_TASK_ASSET_FIELDS
+ | dag_version_diff._DIFF_V1_TASK_PARAM_FIELDS
+ | dag_version_diff._DIFF_V1_TASK_DEPENDENCY_FIELDS
+ | dag_version_diff._DIFF_V1_TASK_METADATA_FIELDS
+ )
+
+ assert classified <= _DIFF_V1_PUBLIC_PARTIAL_TASK_FIELDS
+
+
+@pytest.mark.parametrize("field", sorted(dag_version_diff._DIFF_V1_LEGACY_DAG_FIELD_CATEGORIES))
+def test_legacy_dag_field_categories_are_reachable(field) -> None:
+ # The v1 conversion rewrites some retired fields away, and a category for a field that never
+ # survives canonicalization classifies nothing.
+ payload = _build_v1_payload()
+ payload["dag"][field] = "@daily"
+
+ canonical = dag_version_diff._canonicalize_payload_v1(payload)
+
+ assert field in canonical["dag"]
+
+
+def _build_params_payload(params: dict) -> dict:
+ with DAG("params_example", schedule=None, params=params) as dag:
+ BashOperator(task_id="extract", bash_command="echo hello")
+ return _serialize_dag(dag)
+
+
+def _get_param_container(payload: dict[str, Any], location: str) -> dict[str, Any]:
+ return payload["dag"] if location == "dag" else _get_task(payload)
+
+
+def _get_hydrated_params(payload: dict[str, Any], location: str):
+ dag = DagSerialization.from_dict(copy.deepcopy(payload))
+ return dag.params if location == "dag" else dag.task_dict["extract"].params
+
+
+def _rewrite_params(payload: dict, rewrite) -> dict:
+ payload = copy.deepcopy(payload)
+ for data in [payload["dag"], *(task["__var"] for task in payload["dag"]["tasks"])]:
+ if "params" in data:
+ data["params"] = rewrite(data["params"])
+ return payload
+
+
+def _as_legacy_class(params: list) -> list:
+ return [[name, {**param, "__class": "airflow.models.param.Param"}] for name, param in params]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize(
+ ("rewrite", "shape"),
+ [
+ (dict, "mapping stored by 2.9.2 and earlier"),
+ (_as_legacy_class, "pre-SDK Param class identifier"),
+ ],
+)
+def test_build_diff_normalizes_equivalent_param_shapes(rewrite, shape, include_values):
+ current = _build_params_payload({"x": 1, "y": 2})
+ stored = _rewrite_params(current, rewrite)
+ hydrated = [
+ list(DagSerialization.from_dict(copy.deepcopy(payload)).params.items())
+ for payload in (stored, current)
+ ]
+ assert [name for name, _ in hydrated[0]] == [name for name, _ in hydrated[1]], shape
+ assert [value.dump() for _, value in hydrated[0]] == [value.dump() for _, value in hydrated[1]], shape
+
+ result = build_serialized_dag_diff(base_data=stored, target_data=current, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize("location", ["dag", "task"])
+@pytest.mark.parametrize(
+ "shape",
+ [
+ "primitive",
+ "minimal_param",
+ "missing_source",
+ "missing_description",
+ "missing_schema",
+ "plain_schema",
+ "plain_default",
+ ],
+)
+def test_diff_normalizes_equivalent_param_contents(include_values, location, shape):
+ value = {"nested": [1, {"value": True}]} if shape == "plain_default" else "same"
+ schema = {"type": "string"} if shape == "plain_schema" else {}
+ current = _build_params_payload({"x": Param(value, schema=schema)})
+ _get_param_container(current, location)["params"] = copy.deepcopy(current["dag"]["params"])
+ stored = copy.deepcopy(current)
+ container = _get_param_container(stored, location)
+ param = container["params"][0][1]
+ if shape == "primitive":
+ container["params"] = {"x": value}
+ elif shape == "minimal_param":
+ container["params"] = {"x": {"__class": "airflow.models.param.Param", "default": value}}
+ elif shape.startswith("missing_"):
+ param.pop(shape.removeprefix("missing_"))
+ elif shape == "plain_schema":
+ param["schema"] = schema
+ else:
+ param["default"] = value
+ DagSerialization.validate_schema(current)
+ # Hydration still accepts the legacy mappings excluded by the current schema.
+ if shape not in {"primitive", "minimal_param"}:
+ DagSerialization.validate_schema(stored)
+ assert (
+ _get_hydrated_params(stored, location).get_param("x").dump()
+ == _get_hydrated_params(current, location).get_param("x").dump()
+ )
+
+ result = build_serialized_dag_diff(base_data=stored, target_data=current, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+def test_build_diff_still_reports_renamed_params() -> None:
+ result = build_serialized_dag_diff(
+ base_data=_build_params_payload({"x": 1}),
+ target_data=_build_params_payload({"renamed": 1}),
+ )
+
+ assert [change["path"] for change in result["changes"]] == [
+ "/dag/params",
+ "/dag/tasks/*/params",
+ ]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_diff_preserves_param_definition_order(include_values):
+ base = _build_params_payload({"first": 1, "second": 2})
+ target = _build_params_payload({"second": 2, "first": 1})
+ for location in ("dag", "task"):
+ assert list(_get_hydrated_params(base, location)) == ["first", "second"]
+ assert list(_get_hydrated_params(target, location)) == ["second", "first"]
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ task_id = "extract" if include_values else "*"
+ assert [change["path"] for change in result["changes"]] == ["/dag/params", f"/dag/tasks/{task_id}/params"]
+ if include_values:
+ for change in result["changes"]:
+ assert [pair[0] for pair in change["before_value"]] == ["first", "second"]
+ assert [pair[0] for pair in change["after_value"]] == ["second", "first"]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize("location", ["dag", "task"])
+@pytest.mark.parametrize(
+ ("field", "before", "after"),
+ [
+ ("default", "old", "new"),
+ ("schema", {"minLength": 1}, {"minLength": 2}),
+ ("source", "dag", "task"),
+ ],
+)
+def test_diff_preserves_param_value_schema_and_source_changes(include_values, location, field, before, after):
+ base = _build_params_payload({"x": "same"})
+ target = copy.deepcopy(base)
+ for payload, value in ((base, before), (target, after)):
+ _get_param_container(payload, location)["params"][0][1][field] = value
+ assert (
+ _get_hydrated_params(base, location).get_param("x").dump()
+ != _get_hydrated_params(target, location).get_param("x").dump()
+ )
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert len(result["changes"]) == 1
+ task_id = "extract" if include_values else "*"
+ change = result["changes"][0]
+ assert change["path"] == ("/dag/params" if location == "dag" else f"/dag/tasks/{task_id}/params")
+ if include_values:
+ assert change["before_digest"] != change["after_digest"]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_diff_preserves_literal_encoding_keys_in_param_defaults(include_values):
+ base = _build_params_payload({"x": {"nested": {"__type": "dict", "__var": {"value": "same"}}}})
+ target = _build_params_payload({"x": {"nested": {"value": "same"}}})
+ assert _get_hydrated_params(base, "dag")["x"] != _get_hydrated_params(target, "dag")["x"]
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ task_id = "extract" if include_values else "*"
+ assert [change["path"] for change in result["changes"]] == ["/dag/params", f"/dag/tasks/{task_id}/params"]
+ if include_values:
+ assert all(change["before_digest"] != change["after_digest"] for change in result["changes"])
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize("mapped", [False, True])
+@pytest.mark.parametrize("change", ["class", "description"])
+def test_diff_preserves_templated_params(include_values, mapped, change):
+ with DAG("diff_regression", schedule=None, params={"x": "same"}) as dag:
+ if mapped:
+ BashOperator.partial(task_id="extract").expand(bash_command=["echo hello"])
+ else:
+ BashOperator(task_id="extract", bash_command="echo hello")
+ base = _serialize_dag(dag)
+ task = _get_task(base)
+ task["template_fields"].append("params")
+ operator_schema = {"$ref": "#/definitions/operator", "definitions": load_dag_schema_dict()["definitions"]}
+ schema_defaults = DagSerialization.get_schema_defaults("operator")
+ for field in operator_schema["definitions"]["operator"]["required"]:
+ if field not in task:
+ task[field] = schema_defaults[field]
+ target = copy.deepcopy(base)
+ param = _get_task(target)["params"][0][1]
+ if change == "class":
+ param["__class"] = "airflow.models.param.Param"
+ else:
+ param.pop("description")
+ originals = copy.deepcopy((base, target))
+ for payload in (base, target):
+ DagSerialization.validate_schema(payload)
+ Draft7Validator(operator_schema).validate(_get_task(payload))
+ hydrated_params = _get_hydrated_params(payload, "task")
+ assert isinstance(hydrated_params, list)
+ assert hydrated_params == _get_task(payload)["params"]
+ assert _get_task(base)["params"] != _get_task(target)["params"]
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ task_id = "extract" if include_values else "*"
+ assert change["path"] == f"/dag/tasks/{task_id}/params"
+ assert change["category"] == "param"
+ assert change["impact"] == "execution"
+ if include_values:
+ assert change["before_value"] == _get_task(base)["params"]
+ assert change["after_value"] == _get_task(target)["params"]
+ assert change["before_digest"] != change["after_digest"]
+ else:
+ assert "before_value" not in change
+ assert "after_value" not in change
+ assert (base, target) == originals
+
+
+def _build_dated_payloads(field: str, task_date: datetime | None) -> list[dict]:
+ payloads = []
+ for day in (2, 1):
+ dag_kwargs: dict[str, Any] = {field: datetime(2024, 1, day, tzinfo=timezone.utc)}
+ task_kwargs: dict[str, Any] = {field: task_date} if task_date else {}
+ with DAG("dates_example", schedule=None, **dag_kwargs) as dag:
+ BashOperator(task_id="extract", bash_command="echo hello", **task_kwargs)
+ payloads.append(json.loads(json.dumps(DagSerialization.to_dict(dag))))
+ return payloads
+
+
+def _hydrated_task_dates(payloads: list[dict], field: str) -> list[Any]:
+ return [
+ getattr(DagSerialization.from_dict(copy.deepcopy(payload)).task_dict["extract"], field)
+ for payload in payloads
+ ]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize(
+ ("field", "pinned_day", "stored_dates"),
+ [("start_date", 2, [False, True]), ("end_date", 1, [True, False])],
+)
+def test_build_diff_ignores_task_dates_matching_the_dag(include_values, field, pinned_day, stored_dates):
+ pinned = datetime(2024, 1, pinned_day, tzinfo=timezone.utc)
+ payloads = _build_dated_payloads(field, pinned)
+ assert [field in payload["dag"]["tasks"][0]["__var"] for payload in payloads] == stored_dates
+ assert set(_hydrated_task_dates(payloads, field)) == {pinned}
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ assert result["mode"] == "observed_state"
+ assert [change["path"] for change in result["changes"]] == [f"/dag/{field}"]
+
+
+@pytest.mark.parametrize("field", ["start_date", "end_date"])
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_reports_inherited_task_date_changes(field, include_values):
+ payloads = _build_dated_payloads(field, task_date=None)
+ # Neither payload stores the task date, but each task still inherits its own Dag's.
+ assert not any(field in payload["dag"]["tasks"][0]["__var"] for payload in payloads)
+ base_date, target_date = _hydrated_task_dates(payloads, field)
+ assert base_date != target_date
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ assert result["mode"] == "observed_state"
+ assert [change["path"] for change in result["changes"]] == [
+ f"/dag/{field}",
+ f"/dag/tasks/{'extract' if include_values else '*'}/{field}",
+ ]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_reports_values_status_from_every_entry_point(include_values):
+ base = _build_payload(tasks=[{"task_id": "extract", "retries": 1}])
+ target = _build_payload(tasks=[{"task_id": "extract", "retries": 2}])
+ expected = "available" if include_values else "unavailable"
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+ unavailable = build_unavailable_dag_diff(
+ base_data=base, target_data=None, reason="serialized_dag_missing"
+ )
+
+ assert result["values"] == {"status": expected}
+ assert unavailable["values"] == {"status": "unavailable"}
+
+
+def test_build_diff_pins_diff_schema_version_and_top_level_keys() -> None:
+ base = _build_payload(tasks=[{"task_id": "extract", "retries": 1}])
+ target = _build_payload(tasks=[{"task_id": "extract", "retries": 2}])
+
+ observed = build_serialized_dag_diff(base_data=base, target_data=target)
+ unavailable = build_unavailable_dag_diff(
+ base_data=base, target_data=None, reason="serialized_dag_missing"
+ )
+
+ assert observed == {
+ "diff_schema_version": 1,
+ "serialized_dag_schema_versions": {"base": 3, "target": 3},
+ "mode": "observed_state",
+ "changes": [
+ {
+ "path": "/dag/tasks/*/retries",
+ "operation": "changed",
+ "category": "task",
+ "impact": "execution",
+ "occurrence_count": 1,
+ }
+ ],
+ "truncated": False,
+ "values": {"status": "unavailable"},
+ }
+ assert unavailable == {
+ "diff_schema_version": 1,
+ "serialized_dag_schema_versions": {"base": 3, "target": None},
+ "mode": "unavailable",
+ "unavailable_reason": "serialized_dag_missing",
+ "changes": [],
+ "truncated": False,
+ "values": {"status": "unavailable"},
+ }
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_marks_values_unavailable_when_comparison_fails(include_values):
+ base = _build_payload(tasks=[])
+ target = copy.deepcopy(base)
+ target["dag"]["tasks"] = "not a list"
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "unavailable"
+ assert result["values"] == {"status": "unavailable"}
+
+
+def test_build_diff_rejects_unsupported_client_defaults_sections(caplog):
+ base = _build_payload(tasks=[{"task_id": "extract"}])
+ target = copy.deepcopy(base)
+ base["client_defaults"] = {"tasks": {}, "dags": {"catchup": True}}
+ target["client_defaults"] = {"tasks": {}, "dags": {"catchup": False}}
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=True)
+
+ assert result["mode"] == "unavailable"
+ assert result["unavailable_reason"] == "serialized_dag_canonicalization_failed"
+ assert result["changes"] == []
+ assert {
+ "event": "Serialized Dag diff canonicalization failed",
+ "error_type": "ValueError",
+ } in caplog
+
+
+@pytest.mark.parametrize("max_changes", [1, 9, 10, 11])
+def test_build_diff_bound_withholds_records_not_public_paths(max_changes):
+ def payload(value):
+ return _build_payload(
+ tasks=[
+ {
+ "task_id": f"task{index}",
+ "executor_config": {"__type": "dict", "__var": {f"k{key}": value for key in range(20)}},
+ }
+ for index in range(10)
+ ]
+ )
+
+ base, target = payload("old"), payload("new")
+ results = {
+ include_values: build_serialized_dag_diff(
+ base_data=copy.deepcopy(base),
+ target_data=copy.deepcopy(target),
+ include_values=include_values,
+ max_changes=max_changes,
+ )
+ for include_values in (False, True)
+ }
+
+ redacted, valued = results[False], results[True]
+ included = min(10, max_changes)
+ assert len(valued["changes"]) == included
+ assert valued["truncated"] is (max_changes < 10)
+ assert {change["path"] for change in valued["changes"]} == {
+ f"/dag/tasks/task{index}/executor_config" for index in range(included)
+ }
+ # The sole public path is disclosed by the first change, so no bound can withhold a path
+ # from the caller or cost an occurrence.
+ assert redacted["truncated"] is False
+ assert redacted["changes"] == [
+ {
+ "path": "/dag/tasks/*/executor_config",
+ "operation": "changed",
+ "category": "task",
+ "impact": "execution",
+ "occurrence_count": 10,
+ }
+ ]
+ assert all(change["occurrence_count"] == 1 for change in valued["changes"])
+ assert all("after_value" in change for change in valued["changes"])
+
+
+@pytest.mark.parametrize("max_changes", [2, 3])
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_excludes_new_group_beyond_change_limit(max_changes, include_values):
+ base = _build_payload(tasks=[{"task_id": name, "retries": 0} for name in ("a", "b", "c")])
+ target = _build_payload(
+ tasks=[
+ {"task_id": "a", "retries": 1},
+ {"task_id": "b", "retries": 1},
+ {"task_id": "c", "queue": "new"},
+ ]
+ )
+
+ result = build_serialized_dag_diff(
+ base_data=base, target_data=target, max_changes=max_changes, include_values=include_values
+ )
+
+ if include_values:
+ assert result["truncated"] is (max_changes == 2)
+ expected_paths = ["/dag/tasks/a/retries", "/dag/tasks/b/retries", "/dag/tasks/c/queue"]
+ assert [change["path"] for change in result["changes"]] == expected_paths[:max_changes]
+ else:
+ assert result["truncated"] is False
+ assert [(change["path"], change["occurrence_count"]) for change in result["changes"]] == [
+ ("/dag/tasks/*/retries", 2),
+ ("/dag/tasks/*/queue", 1),
+ ]
+
+
+@pytest.mark.parametrize("max_changes", [3, 2])
+@pytest.mark.parametrize("repeated_path", [False, True])
+def test_build_diff_truncation_boundary_is_exact_in_both_modes(repeated_path, max_changes):
+ """One change over the bound truncates only when it would disclose an unseen path."""
+ if repeated_path:
+ base_tasks = [{"task_id": name, "owner": "alice"} for name in ("a", "b", "c")]
+ target_tasks = [{"task_id": name, "owner": "bob"} for name in ("a", "b", "c")]
+ redacted_paths = ["/dag/tasks/*/owner"]
+ else:
+ base_tasks = [{"task_id": "a", "doc_md": "old", "owner": "alice", "queue": "old"}]
+ target_tasks = [{"task_id": "a", "doc_md": "new", "owner": "bob", "queue": "new"}]
+ redacted_paths = ["/dag/tasks/*/doc_md", "/dag/tasks/*/owner", "/dag/tasks/*/queue"]
+
+ redacted, valued = [
+ build_serialized_dag_diff(
+ base_data=_build_payload(tasks=copy.deepcopy(base_tasks)),
+ target_data=_build_payload(tasks=copy.deepcopy(target_tasks)),
+ include_values=include_values,
+ max_changes=max_changes,
+ )
+ for include_values in (False, True)
+ ]
+
+ assert len(valued["changes"]) == max_changes
+ assert valued["truncated"] is (max_changes < 3)
+ assert [change["path"] for change in redacted["changes"]] == redacted_paths[: len(redacted["changes"])]
+ if repeated_path:
+ assert redacted["truncated"] is False
+ assert [change["occurrence_count"] for change in redacted["changes"]] == [3]
+ else:
+ assert redacted["truncated"] is (max_changes < 3)
+ assert [change["occurrence_count"] for change in redacted["changes"]] == [1] * max_changes
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_keeps_operations_separate_when_grouping(include_values):
+ base = _build_payload(
+ tasks=[
+ {"task_id": "a", "doc_md": "before"},
+ {"task_id": "b"},
+ {"task_id": "c", "doc_md": "before"},
+ {"task_id": "d", "doc_md": "before"},
+ ]
+ )
+ target = _build_payload(
+ tasks=[
+ {"task_id": "a"},
+ {"task_id": "b", "doc_md": "after"},
+ {"task_id": "c", "doc_md": "after"},
+ {"task_id": "d", "doc_md": "different"},
+ ]
+ )
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["truncated"] is False
+ assert [(change["operation"], change["occurrence_count"]) for change in result["changes"]] == (
+ [("removed", 1), ("added", 1), ("changed", 1), ("changed", 1)]
+ if include_values
+ else [("removed", 1), ("added", 1), ("changed", 2)]
+ )
+ assert all(change["category"] == change["impact"] == "metadata" for change in result["changes"])
+ if not include_values:
+ assert {change["path"] for change in result["changes"]} == {"/dag/tasks/*/doc_md"}
+ assert all(value not in json.dumps(result) for value in ("before", "after", "different"))
+ for payload in (base, target):
+ payload["dag"]["tasks"].reverse()
+ payload["dag"] = dict(reversed(payload["dag"].items()))
+ assert result == build_serialized_dag_diff(
+ base_data=base, target_data=target, include_values=include_values
+ )
+
+
+@pytest.mark.parametrize(("max_changes", "valued_truncated"), [(500, True), (1201, False)])
+def test_build_diff_groups_repeated_changes_in_large_dag(max_changes, valued_truncated):
+ payloads = []
+ for day, retries in ((1, 0), (2, 1)):
+ with DAG(
+ "grouped_diff", schedule=None, start_date=datetime(2025, 1, day, tzinfo=timezone.utc)
+ ) as dag:
+ for index in range(600):
+ BashOperator(task_id=f"task_{index:04d}", bash_command="echo hello", retries=retries)
+ payloads.append(_serialize_dag(dag))
+
+ redacted, valued = [
+ build_serialized_dag_diff(
+ base_data=payloads[0],
+ target_data=payloads[1],
+ include_values=include_values,
+ max_changes=max_changes,
+ )
+ for include_values in (False, True)
+ ]
+
+ assert valued["truncated"] is valued_truncated
+ # All three public paths appear within the first few changes, so the redacted diff stays
+ # complete and exact at a bound the authorized diff exhausts.
+ assert redacted["truncated"] is False
+ assert [(change["path"], change["occurrence_count"]) for change in redacted["changes"]] == [
+ ("/dag/start_date", 1),
+ ("/dag/tasks/*/retries", 600),
+ ("/dag/tasks/*/start_date", 600),
+ ]
+ assert len(valued["changes"]) == len({change["path"] for change in valued["changes"]}) == max_changes
+ assert sum(change["occurrence_count"] for change in redacted["changes"]) == 1201
+
+
+@pytest.mark.parametrize("max_changes", [100, 200])
+def test_build_diff_counts_every_occurrence_at_a_disclosed_path(max_changes):
+ """A Dag-level edit fanning out to every task stays complete and exactly counted."""
+ payloads = []
+ for owner in ("alice", "bob"):
+ with DAG(
+ "fan_out",
+ schedule=None,
+ start_date=datetime(2025, 1, 1, tzinfo=timezone.utc),
+ default_args={"owner": owner},
+ ) as dag:
+ for index in range(120):
+ EmptyOperator(task_id=f"task_{index:04d}")
+ payloads.append(_serialize_dag(dag))
+
+ redacted = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], max_changes=max_changes
+ )
+
+ assert redacted["truncated"] is False
+ assert [(change["path"], change["occurrence_count"]) for change in redacted["changes"]] == [
+ ("/dag/default_args/owner", 1),
+ ("/dag/tasks/*/owner", 120),
+ ]
+
+
+def test_build_diff_reports_no_execution_impact_for_a_default_args_owner_edit() -> None:
+ """The inherited task change and the default_args key it came from agree on impact."""
+ payloads = []
+ for owner in ("alice", "bob"):
+ with DAG(
+ "owner_only",
+ schedule=None,
+ start_date=datetime(2025, 1, 1, tzinfo=timezone.utc),
+ default_args={"owner": owner},
+ ) as dag:
+ EmptyOperator(task_id="extract")
+ payloads.append(_serialize_dag(dag))
+
+ result = build_serialized_dag_diff(base_data=payloads[0], target_data=payloads[1])
+
+ assert [(change["path"], change["category"], change["impact"]) for change in result["changes"]] == [
+ ("/dag/default_args/owner", "metadata", "metadata"),
+ ("/dag/tasks/*/owner", "metadata", "metadata"),
+ ]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize("operation", ["added", "removed"])
+def test_build_diff_classifies_the_first_default_args_entry_per_key(operation, include_values):
+ """A Dag omits default_args entirely until it has one, so the boundary must not skip per-key."""
+ payloads = []
+ for default_args in (None, {"owner": "bob"}):
+ keywords = {} if default_args is None else {"default_args": default_args}
+ with DAG("first_entry", schedule=None, **keywords) as dag:
+ EmptyOperator(task_id="extract")
+ payloads.append(_serialize_dag(dag))
+ assert "default_args" not in payloads[0]["dag"], "a Dag without default_args omits the key"
+ base, target = payloads if operation == "added" else payloads[::-1]
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ default_args_changes = [
+ (change["path"], change["operation"], change["category"], change["impact"])
+ for change in result["changes"]
+ if "default_args" in change["path"]
+ ]
+ assert default_args_changes == [("/dag/default_args/owner", operation, "metadata", "metadata")]
+
+
+def test_build_diff_reports_nothing_when_neither_version_sets_default_args() -> None:
+ payloads = []
+ for task_id in ("extract", "extract"):
+ with DAG("no_default_args", schedule=None) as dag:
+ EmptyOperator(task_id=task_id)
+ payloads.append(_serialize_dag(dag))
+
+ result = build_serialized_dag_diff(base_data=payloads[0], target_data=payloads[1])
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize(
+ ("max_changes", "expected_counts", "truncated"),
+ [(1, [1], True), (3, [1, 1, 1], True), (5, [1, 1, 4, 1], False), (500, [1, 1, 4, 1], False)],
+)
+def test_build_diff_redacted_records_mirror_authorized_paths(max_changes, expected_counts, truncated):
+ """Redacted records are the authorized paths mapped through the public-path projection."""
+ base = _build_payload(
+ tasks=[{"task_id": name, "owner": "alice"} for name in ("a", "b", "c", "d")], tags=["old"]
+ )
+ base["dag"]["tasks"][0]["__var"]["queue"] = "old"
+ target = _build_payload(
+ tasks=[{"task_id": name, "owner": "bob"} for name in ("a", "b", "c", "d")], tags=["new"]
+ )
+ target["dag"]["tasks"][0]["__var"]["queue"] = "new"
+
+ projection = dag_version_diff._get_public_path
+ with mock.patch.object(dag_version_diff, "_get_public_path", wraps=projection, spec=True) as spy:
+ valued = build_serialized_dag_diff(
+ base_data=copy.deepcopy(base),
+ target_data=copy.deepcopy(target),
+ include_values=True,
+ max_changes=max_changes,
+ )
+ redacted = build_serialized_dag_diff(
+ base_data=copy.deepcopy(base), target_data=copy.deepcopy(target), max_changes=max_changes
+ )
+
+ # add() projects each change it sees, so the leading calls carry the raw authorized paths.
+ raw_paths = [call.args[0] for call in spy.call_args_list][: len(valued["changes"])]
+ assert [(change["path"], change["operation"]) for change in redacted["changes"]] == list(
+ dict.fromkeys(
+ (dag_version_diff._format_path(projection(raw_path)), change["operation"])
+ for raw_path, change in zip(raw_paths, valued["changes"], strict=True)
+ )
+ )
+ assert [change["occurrence_count"] for change in redacted["changes"]] == expected_counts
+ assert redacted["truncated"] is truncated
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize(
+ ("field", "before", "after", "category", "impact"),
+ [
+ ("start_date", 1704067200, 1704153600, "schedule", "execution"),
+ ("end_date", 1704153600, 1704240000, "schedule", "execution"),
+ ("timezone", "UTC", "Europe/Paris", "schedule", "execution"),
+ ("_concurrency", 1, 2, "schedule", "execution"),
+ ("max_consecutive_failed_dag_runs", 1, 2, "schedule", "execution"),
+ ("is_paused_upon_creation", False, True, "schedule", "execution"),
+ ("render_template_as_native_obj", False, True, "task", "execution"),
+ ("disable_bundle_versioning", False, True, "task", "execution"),
+ ("rerun_with_latest_version", False, True, "task", "execution"),
+ # Only a value that is not a readable dict envelope still classifies as a whole; the
+ # per-key classification of a real default_args mapping is covered separately.
+ (
+ "default_args",
+ {"owner": "first"},
+ {"owner": "second"},
+ "task",
+ "execution",
+ ),
+ ("dag_id", "old", "new", "metadata", "metadata"),
+ ("dag_display_name", "Old name", "New name", "metadata", "metadata"),
+ ("owner_links", {"owner": "old"}, {"owner": "new"}, "metadata", "metadata"),
+ (
+ "edge_info",
+ {"extract": {"load": {"label": "old"}}},
+ {"extract": {"load": {"label": "new"}}},
+ "metadata",
+ "metadata",
+ ),
+ ("fileloc", "/old.py", "/new.py", "provenance", "provenance"),
+ ("relative_fileloc", "old.py", "new.py", "provenance", "provenance"),
+ ("bundle_name", "old", "new", "provenance", "provenance"),
+ ("_processor_dags_folder", "/old", "/new", "provenance", "provenance"),
+ (
+ "access_control",
+ {"private-role": ["can_read"]},
+ {"private-role": ["can_edit"]},
+ "authorization",
+ "authorization",
+ ),
+ ("custom_callback_setting", "old", "new", "unknown", "unknown"),
+ # A differently cased schema name is a different key, not the schema field it resembles.
+ ("TAGS", ["old"], ["new"], "unknown", "unknown"),
+ ],
+)
+def test_build_diff_classifies_dag_fields(include_values, field, before, after, category, impact):
+ base = _build_payload(tasks=[])
+ target = _build_payload(tasks=[])
+ base["dag"][field] = before
+ target["dag"][field] = after
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ assert change["category"] == category
+ assert change["impact"] == impact
+ if not include_values:
+ assert change == {
+ "path": f"/dag/{field}",
+ "operation": "changed",
+ "category": category,
+ "impact": impact,
+ "occurrence_count": 1,
+ }
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize(
+ ("key", "before", "after", "category", "impact"),
+ [
+ ("owner", "alice", "bob", "metadata", "metadata"),
+ ("retries", 3, 5, "task", "execution"),
+ ("queue", "old", "new", "task", "execution"),
+ ("has_on_failure_callback", False, True, "callback", "execution"),
+ ("outlets", [], ["asset"], "asset", "execution"),
+ ("downstream_task_ids", [], ["load"], "dependency", "execution"),
+ ],
+)
+def test_build_diff_classifies_default_args_per_key(include_values, key, before, after, category, impact):
+ base = _build_payload(tasks=[])
+ target = _build_payload(tasks=[])
+ base["dag"]["default_args"] = {"__type": "dict", "__var": {key: before}}
+ target["dag"]["default_args"] = {"__type": "dict", "__var": {key: after}}
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ # An allowlisted key is a fixed operator field name, so both modes report the same path.
+ assert change["path"] == f"/dag/default_args/{key}"
+ assert (change["operation"], change["category"], change["impact"]) == ("changed", category, impact)
+
+
+def _build_task_group_payload(*, field: str, value: str, depth: int) -> tuple[dict, list[str]]:
+ group_ids = []
+ with DAG("task_group_diff", schedule=None) as dag, ExitStack() as stack:
+ group = dag.task_group
+ for index in range(depth):
+ group = stack.enter_context(TaskGroup(group_id=f"secret_group_{index}"))
+ group_id = group.group_id
+ assert group_id is not None
+ group_ids.append(group_id)
+ setattr(group, field, value)
+ BashOperator(task_id="secret_task", bash_command="echo hello")
+ return DagSerialization.to_dict(dag), group_ids
+
+
+def _build_group_path(group_ids: list[str], *, include_values: bool) -> str:
+ return "/dag/task_group" + "".join(
+ f"/children/{group_id if include_values else '*'}/1" for group_id in group_ids
+ )
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize(
+ ("depth", "json_roundtrip"), [(0, False), (1, False), (1, True), (2, False), (2, True)]
+)
+@pytest.mark.parametrize("field", ["group_display_name", "tooltip", "doc_md", "ui_color", "ui_fgcolor"])
+def test_build_diff_classifies_task_group_display_fields(include_values, depth, json_roundtrip, field):
+ base, group_ids = _build_task_group_payload(field=field, value="old", depth=depth)
+ target, _ = _build_task_group_payload(field=field, value="new", depth=depth)
+ if json_roundtrip:
+ base, target = json.loads(json.dumps([base, target]))
+ originals = copy.deepcopy((base, target))
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ assert change["path"] == f"{_build_group_path(group_ids, include_values=include_values)}/{field}"
+ assert change["operation"] == "changed"
+ assert change["category"] == change["impact"] == "metadata"
+ if include_values:
+ assert (change["before_value"], change["after_value"]) == ("old", "new")
+ else:
+ assert "secret_" not in json.dumps(result)
+ assert (base, target) == originals
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize("nested", [False, True])
+def test_build_diff_classifies_mapped_group_inputs_as_execution(include_values, nested):
+ @task_group(group_id="secret_inner")
+ def mapped_group(command):
+ BashOperator(task_id="secret_task", bash_command=command)
+
+ payloads = []
+ for commands in (["first"], ["first", "second"]):
+ with DAG("task_group_diff", schedule=None) as dag, ExitStack() as stack:
+ if nested:
+ stack.enter_context(TaskGroup(group_id="secret_outer"))
+ mapped_group.expand(command=commands)
+ payloads.append(DagSerialization.to_dict(dag))
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ assert result["mode"] == "observed_state"
+ group_changes = [change for change in result["changes"] if change["path"].startswith("/dag/task_group/")]
+ assert len(group_changes) == 1
+ change = group_changes[0]
+ group_ids = ["secret_outer", "secret_outer.secret_inner"] if nested else ["secret_inner"]
+ path = _build_group_path(group_ids, include_values=include_values)
+ task_id = f"{group_ids[-1]}.secret_task" if include_values else "*"
+ assert {change["path"] for change in result["changes"]} == {
+ f"{path}/expand_input",
+ f"/dag/tasks/{task_id}/custom_fields",
+ }
+ assert all(change["impact"] == "execution" for change in result["changes"])
+ assert change["path"] == f"{path}/expand_input"
+ assert change["category"] == "task"
+ if not include_values:
+ assert "secret_" not in json.dumps(result)
+
+
+@pytest.mark.parametrize("operation", ["added", "removed"])
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_redacts_task_group_membership_changes(operation, include_values):
+ payloads = []
+ for include_group in (False, True):
+ with DAG("task_group_diff", schedule=None) as dag:
+ if include_group:
+ TaskGroup(group_id="secret_group")
+ payloads.append(DagSerialization.to_dict(dag))
+ if operation == "removed":
+ payloads.reverse()
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ group_id = "secret_group" if include_values else "*"
+ assert change["path"] == f"/dag/task_group/children/{group_id}"
+ assert change["operation"] == operation
+ assert change["category"] == "task"
+ assert change["impact"] == "execution"
+ if not include_values:
+ assert "secret_group" not in json.dumps(result)
+
+
+def test_build_diff_bounds_nested_group_changes():
+ base, group_ids = _build_task_group_payload(field="tooltip", value="old", depth=2)
+ target, _ = _build_task_group_payload(field="tooltip", value="new", depth=2)
+ group = target["dag"]["task_group"]
+ for group_id in group_ids:
+ group = group["children"][group_id][1]
+ group["ui_color"] = "changed"
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, max_changes=1)
+
+ assert len(result["changes"]) == 1
+ assert result["changes"][0]["path"] == f"{_build_group_path(group_ids, include_values=False)}/tooltip"
+ assert result["truncated"] is True
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_collapses_unknown_nested_group_fields(include_values):
+ base, group_ids = _build_task_group_payload(field="tooltip", value="same", depth=2)
+ target = copy.deepcopy(base)
+ for payload, value in ((base, "old"), (target, "new")):
+ group = payload["dag"]["task_group"]
+ for group_id in group_ids:
+ group = group["children"][group_id][1]
+ group["secret_custom_field"] = value
+ group["another_secret_field"] = value
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ group_path = _build_group_path(group_ids, include_values=include_values)
+ assert [change["path"] for change in result["changes"]] == [f"{group_path}/custom_fields"]
+ change = result["changes"][0]
+ assert (change["operation"], change["category"], change["impact"]) == (
+ "changed",
+ "task",
+ "execution",
+ )
+ if include_values:
+ assert change["before_value"] == {
+ "secret_custom_field": "old",
+ "another_secret_field": "old",
+ }
+ assert change["after_value"] == {
+ "secret_custom_field": "new",
+ "another_secret_field": "new",
+ }
+ else:
+ assert "secret_" not in json.dumps(result)
+
+
+def test_public_group_path_truncates_non_public_group_fields() -> None:
+ # No walk reaches this: _collect_public_field_changes aggregates non-public group keys into
+ # custom_fields first, so only the masking itself can be pinned against a future caller.
+ assert dag_version_diff._get_public_path(
+ ("dag", "task_group", "children", "secret_group", "1", "secret_field", "secret_inner_key")
+ ) == ("dag", "task_group", "children", "*", "1", "custom_fields")
+
+
+def test_build_diff_reports_known_nested_group_fields_beside_collapsed_ones():
+ base, group_ids = _build_task_group_payload(field="tooltip", value="old", depth=2)
+ target, _ = _build_task_group_payload(field="tooltip", value="new", depth=2)
+ for payload, value in ((base, "old"), (target, "new")):
+ group = payload["dag"]["task_group"]
+ for group_id in group_ids:
+ group = group["children"][group_id][1]
+ group["secret_custom_field"] = value
+ group["another_secret_field"] = value
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ group_path = _build_group_path(group_ids, include_values=False)
+ assert [change["path"] for change in result["changes"]] == [
+ f"{group_path}/tooltip",
+ f"{group_path}/custom_fields",
+ ]
+ assert [change["category"] for change in result["changes"]] == ["metadata", "task"]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize(
+ ("max_changes", "expected_counts", "redacted_truncated"),
+ [
+ (1, [("tooltip", 1)], True),
+ (3, [("tooltip", 2), ("custom_fields", 2)], False),
+ (4, [("tooltip", 2), ("custom_fields", 2)], False),
+ ],
+)
+def test_build_diff_groups_nested_task_group_changes(
+ include_values, max_changes, expected_counts, redacted_truncated
+):
+ with DAG("grouped_diff", schedule=None) as dag:
+ with TaskGroup("secret_outer"):
+ for index in range(2):
+ TaskGroup(f"secret_inner_{index}", tooltip="before")
+ base = _serialize_dag(dag)
+ target = copy.deepcopy(base)
+ for payload, value in ((base, "before"), (target, "after")):
+ children = payload["dag"]["task_group"]["children"]["secret_outer"][1]["children"]
+ for _, group in children.values():
+ group.update(tooltip=value, secret_field=value, another_secret_field=value)
+
+ result = build_serialized_dag_diff(
+ base_data=base, target_data=target, include_values=include_values, max_changes=max_changes
+ )
+
+ if include_values:
+ assert result["truncated"] is (max_changes < 4)
+ assert sum(change["occurrence_count"] for change in result["changes"]) == max_changes
+ expected_paths = [
+ f"/dag/task_group/children/secret_outer/1/children/secret_outer.secret_inner_{index}/1/{field}"
+ for index in range(2)
+ for field in ("tooltip", "custom_fields")
+ ]
+ assert [change["path"] for change in result["changes"]] == expected_paths[:max_changes]
+ assert all(change["occurrence_count"] == 1 for change in result["changes"])
+ else:
+ assert result["truncated"] is redacted_truncated
+ assert [(change["path"], change["occurrence_count"]) for change in result["changes"]] == [
+ (f"/dag/task_group/children/*/1/children/*/1/{field}", count) for field, count in expected_counts
+ ]
+ assert "secret_" not in json.dumps(result)
+ for payload in (base, target):
+ outer = payload["dag"]["task_group"]["children"]["secret_outer"][1]
+ outer["children"] = dict(reversed(outer["children"].items()))
+ assert result == build_serialized_dag_diff(
+ base_data=base, target_data=target, include_values=include_values, max_changes=max_changes
+ )
+
+
+def test_build_diff_is_deterministic_and_normalizes_order() -> None:
+ base = _build_payload(
+ tasks=[{"task_id": "extract", "retries": 1}, {"task_id": "load", "retries": 1}],
+ tags=["one", "two"],
+ dependencies=[
+ {
+ "dependency_type": "task",
+ "dependency_id": "extract-load",
+ "source": "extract",
+ "target": "load",
+ "label": "extract-load",
+ }
+ ],
+ )
+ target = _build_payload(
+ tasks=[{"task_id": "load", "retries": 1}, {"task_id": "extract", "retries": 1}],
+ tags=["two", "one"],
+ dependencies=[
+ {
+ "label": "extract-load",
+ "target": "load",
+ "source": "extract",
+ "dependency_id": "extract-load",
+ "dependency_type": "task",
+ }
+ ],
+ )
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ assert result["mode"] == "observed_state"
+ assert result["serialized_dag_schema_versions"] == {"base": 3, "target": 3}
+ assert result["changes"] == []
+ assert result["truncated"] is False
+
+
+def test_build_diff_reports_categories_digests_and_values() -> None:
+ base = _build_payload(tasks=[{"task_id": "extract", "retries": 1}], tags=["old"])
+ target = _build_payload(
+ tasks=[
+ {"task_id": "extract", "retries": 2},
+ {"task_id": "load", "retries": 1},
+ ],
+ tags=["new"],
+ schedule="hourly",
+ )
+
+ result = build_serialized_dag_diff(
+ base_data=base,
+ target_data=target,
+ base_provenance={"bundle_version": "one"},
+ target_provenance={"bundle_version": "two"},
+ include_values=True,
+ )
+
+ changes = {change["path"]: change for change in result["changes"]}
+ assert changes["/dag/tasks/extract/retries"]["category"] == "task"
+ assert changes["/dag/tasks/extract/retries"]["impact"] == "execution"
+ assert changes["/dag/tasks/extract/retries"]["before_value"] == 1
+ assert changes["/dag/tasks/extract/retries"]["after_value"] == 2
+ assert changes["/dag/tasks/extract/retries"]["before_digest"].startswith("sha256:")
+ assert changes["/dag/tasks/load"]["operation"] == "added"
+ assert changes["/dag/schedule"]["category"] == "schedule"
+ assert changes["/dag/tags/old"]["operation"] == "removed"
+ assert changes["/dag/tags/new"]["operation"] == "added"
+ assert changes["/dag/tags/old"]["category"] == "metadata"
+ assert changes["/provenance/bundle_version"]["category"] == "provenance"
+ assert changes["/provenance/bundle_version"]["impact"] == "provenance"
+
+
+def test_build_diff_digests_escaped_surrogates() -> None:
+ base = _build_payload(tasks=[{"task_id": "extract", "doc_md": "café"}])
+ # An escaped lone surrogate survives ensure_ascii=True storage and the compressed column.
+ target = _build_payload(tasks=[{"task_id": "extract", "doc_md": "\ud800"}])
+ assert json.loads(json.dumps(target)) == target
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=True)
+
+ assert result["mode"] == "observed_state"
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ assert change["after_digest"] == "sha256:" + hashlib.sha256(b'"\xed\xa0\x80"').hexdigest()
+ # Payloads that already digested keep their digest, so the wire format is unchanged.
+ unescaped = hashlib.sha256(json.dumps("café", ensure_ascii=False).encode()).hexdigest()
+ assert change["before_digest"] == f"sha256:{unescaped}"
+
+
+def _get_expected_digest(canonical_json: bytes) -> str:
+ return f"sha256:{hashlib.sha256(canonical_json).hexdigest()}"
+
+
+def test_build_diff_pins_authorized_change_record_keys() -> None:
+ base = _build_payload(tasks=[{"task_id": "extract", "retries": 1}], tags=["old"])
+ target = _build_payload(tasks=[{"task_id": "extract", "retries": 2}], tags=["new"])
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=True)
+
+ assert result["changes"] == [
+ {
+ "path": "/dag/tags/new",
+ "operation": "added",
+ "category": "metadata",
+ "impact": "metadata",
+ "occurrence_count": 1,
+ "before_digest": None,
+ "after_digest": _get_expected_digest(b'"new"'),
+ "after_value": "new",
+ },
+ {
+ "path": "/dag/tags/old",
+ "operation": "removed",
+ "category": "metadata",
+ "impact": "metadata",
+ "occurrence_count": 1,
+ "before_digest": _get_expected_digest(b'"old"'),
+ "after_digest": None,
+ "before_value": "old",
+ },
+ {
+ "path": "/dag/tasks/extract/retries",
+ "operation": "changed",
+ "category": "task",
+ "impact": "execution",
+ "occurrence_count": 1,
+ "before_digest": _get_expected_digest(b"1"),
+ "after_digest": _get_expected_digest(b"2"),
+ "before_value": 1,
+ "after_value": 2,
+ },
+ ]
+
+
+def test_build_diff_bounds_changes_and_reports_truncation() -> None:
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[{"task_id": "extract", "retries": 1}], tags=["old"]),
+ target_data=_build_payload(tasks=[{"task_id": "extract", "retries": 2}], tags=["new"]),
+ max_changes=1,
+ )
+
+ assert len(result["changes"]) == 1
+ assert result["truncated"] is True
+
+
+def test_build_diff_reports_dependency_changes() -> None:
+ dependency = {
+ "dependency_type": "sensor",
+ "dependency_id": "upstream-task",
+ "source": "upstream-dag",
+ "target": "example",
+ "label": "upstream-task",
+ }
+
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[], dependencies=[dependency]),
+ target_data=_build_payload(tasks=[]),
+ )
+
+ assert len(result["changes"]) == 1
+ assert result["changes"][0]["category"] == "dependency"
+ assert result["changes"][0]["impact"] == "execution"
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_ignores_dependency_label_renames(include_values):
+ payloads = []
+ for display_name in ("Before", "After"):
+ with DAG("dependency_labels", schedule=None) as dag:
+ TriggerDagRunOperator(task_id="fire", trigger_dag_id="downstream", task_display_name=display_name)
+ ExternalTaskSensor(task_id="watch", external_dag_id="upstream", task_display_name=display_name)
+ payloads.append(_serialize_dag(dag))
+ base_dependencies, target_dependencies = (payload["dag"]["dag_dependencies"] for payload in payloads)
+ assert [dependency["label"] for dependency in base_dependencies] == ["Before", "Before"]
+ assert [dependency["label"] for dependency in target_dependencies] == ["After", "After"]
+ # Every edge keeps the identity DagDependency.node_id is built from.
+ assert [
+ {key: value for key, value in dependency.items() if key != "label"}
+ for dependency in base_dependencies
+ ] == [
+ {key: value for key, value in dependency.items() if key != "label"}
+ for dependency in target_dependencies
+ ]
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ task_ids = ["fire", "watch"] if include_values else ["*"]
+ assert [
+ (change["path"], change["operation"], change["category"], change["impact"])
+ for change in result["changes"]
+ ] == [
+ (f"/dag/tasks/{task_id}/_task_display_name", "changed", "metadata", "metadata")
+ for task_id in task_ids
+ ]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_ignores_downstream_id_reordering(include_values):
+ """
+ A reordered id list is the same edges, not a change.
+
+ _deserialize_operator_field turns ``downstream_task_ids`` into a set, so the order a producer
+ happened to store carries no meaning and must not read as a dependency change.
+ """
+ payloads = [
+ _build_payload(tasks=[{"task_id": "head", "downstream_task_ids": ids}])
+ for ids in (["a", "b"], ["b", "a"])
+ ]
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_reports_a_changed_downstream_id_set(include_values):
+ payloads = [
+ _build_payload(tasks=[{"task_id": "head", "downstream_task_ids": ids}])
+ for ids in (["a", "b"], ["b", "c"])
+ ]
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ task_id = "head" if include_values else "*"
+ assert [(change["path"], change["category"], change["impact"]) for change in result["changes"]] == [
+ (f"/dag/tasks/{task_id}/downstream_task_ids", "dependency", "execution")
+ ]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_ignores_task_group_id_reordering(include_values):
+ """A task group's id lists hydrate to sets too, and declare no template fields."""
+ payloads = []
+ for order in (["g2", "g3"], ["g3", "g2"]):
+ payload = _build_payload(tasks=[])
+ payload["dag"]["task_group"] = {
+ "_group_id": None,
+ "children": {"g1": ["taskgroup", {"_group_id": "g1", "downstream_group_ids": order}]},
+ }
+ payloads.append(payload)
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize(
+ ("path", "is_group_field"),
+ [
+ (("dag", "task_group", "downstream_group_ids"), True),
+ (("dag", "task_group", "children", "g1", "downstream_group_ids"), True),
+ (("dag", "task_group", "children", "g1", "children", "g2", "upstream_task_ids"), True),
+ # A group whose own id happens to be "children" is still a group.
+ (("dag", "task_group", "children", "children", "downstream_group_ids"), True),
+ # User data inside a mapped group's expansion is not a dependency list.
+ (
+ ("dag", "task_group", "children", "g1", "expand_input", "value", "__var", "downstream_task_ids"),
+ False,
+ ),
+ (
+ (
+ "dag",
+ "task_group",
+ "children",
+ "g1",
+ "expand_input",
+ "__var",
+ "children",
+ "x",
+ "upstream_task_ids",
+ ),
+ False,
+ ),
+ (("dag", "tasks", "extract", "downstream_task_ids"), False),
+ ((), False),
+ ],
+)
+def test_group_dependency_path_matches_only_a_group_field(path, is_group_field):
+ assert dag_version_diff._is_group_dependency_path(path) is is_group_field
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_reports_a_mapped_group_expand_input_change(include_values):
+ """
+ A mapped group's expand_input is user data, not a dependency list.
+
+ Its order and multiplicity decide how many task instances run, so normalizing a group's
+ dependency ids must not reach an argument that happens to share one of those names.
+ """
+ payloads = []
+ for values in (["a", "b"], ["a", "b", "a"]):
+ with DAG("mapped_group", schedule=None) as dag:
+
+ @sdk_task
+ def work():
+ return 1
+
+ @task_group
+ def grp(downstream_task_ids):
+ work()
+
+ grp.expand(downstream_task_ids=values)
+ payloads.append(_serialize_dag(dag))
+ stored = [
+ payload["dag"]["task_group"]["children"]["grp"][1]["expand_input"]["value"]["__var"][
+ "downstream_task_ids"
+ ]
+ for payload in payloads
+ ]
+ assert [len(value) for value in stored] == [2, 3], "the stored expansion count changes"
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ group_id = "grp" if include_values else "*"
+ assert [(change["path"], change["impact"]) for change in result["changes"]] == [
+ (f"/dag/task_group/children/{group_id}/1/expand_input", "execution")
+ ]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize(
+ ("field", "legacy"),
+ [
+ ("allowed_run_types", "omitted"),
+ ("allowed_run_types", "empty list"),
+ ("deadline", "omitted"),
+ ],
+)
+def test_build_diff_ignores_fields_a_legacy_producer_left_unset(field, legacy, include_values):
+ """
+ Neither field has a schema default, so an older producer omits what the current one nulls.
+
+ A diff spanning that upgrade would otherwise report an added execution change on every Dag.
+ """
+ with DAG("legacy_unset", schedule=None) as dag:
+ EmptyOperator(task_id="extract")
+ current = _serialize_dag(dag)
+ assert current["dag"][field] is None, "the current serializer writes an explicit null"
+ stored = copy.deepcopy(current)
+ if legacy == "omitted":
+ del stored["dag"][field]
+ else:
+ stored["dag"][field] = []
+ hydrated = [
+ getattr(DagSerialization.from_dict(copy.deepcopy(payload)), field) for payload in (stored, current)
+ ]
+ assert hydrated[0] == hydrated[1] is None
+
+ result = build_serialized_dag_diff(base_data=stored, target_data=current, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+def test_build_diff_classifies_fail_fast_as_schedule() -> None:
+ base = _build_payload(tasks=[])
+ target = _build_payload(tasks=[])
+ base["dag"]["fail_fast"] = False
+ target["dag"]["fail_fast"] = True
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ assert change["path"] == "/dag/fail_fast"
+ assert change["operation"] == "changed"
+ assert change["category"] == "schedule"
+ assert change["impact"] == "execution"
+
+
+@pytest.mark.parametrize(
+ ("base_data", "target_data", "reason"),
+ [
+ (None, _build_payload(tasks=[]), "serialized_dag_missing"),
+ (
+ {"__version": 99, "dag": {}},
+ _build_payload(tasks=[]),
+ "unsupported_serialized_dag_schema_version:99",
+ ),
+ ({"__version": 3, "dag": []}, _build_payload(tasks=[]), "serialized_dag_canonicalization_failed"),
+ (
+ {
+ "__version": 1,
+ "dag": {
+ "dag_id": "example",
+ "tasks": [],
+ "task_group": {},
+ "schedule_interval": 1,
+ },
+ },
+ _build_payload(tasks=[]),
+ "serialized_dag_canonicalization_failed",
+ ),
+ (
+ {
+ "__version": 1,
+ "dag": {
+ "dag_id": "example",
+ "tasks": [],
+ "task_group": {},
+ "schedule_interval": {"__type": "timedelta", "__var": 10**20},
+ },
+ },
+ _build_payload(tasks=[]),
+ "serialized_dag_canonicalization_failed",
+ ),
+ ],
+)
+def test_build_diff_returns_unavailable_for_unsafe_inputs(base_data, target_data, reason, caplog) -> None:
+ result = build_serialized_dag_diff(base_data=base_data, target_data=target_data)
+
+ assert result["mode"] == "unavailable"
+ assert result["changes"] == []
+ assert result["unavailable_reason"] == reason
+ if reason == "serialized_dag_canonicalization_failed":
+ assert {
+ "event": "Serialized Dag diff canonicalization failed",
+ "base_schema_version": base_data["__version"],
+ "target_schema_version": target_data["__version"],
+ } in caplog
+
+
+@pytest.mark.parametrize(
+ "base_data",
+ [
+ pytest.param({"dag": {}}, id="absent-version"),
+ pytest.param({"__version": True, "dag": {}}, id="boolean-version-is-not-version-1"),
+ pytest.param({"__version": "3", "dag": {}}, id="string-version"),
+ pytest.param({"__version": 3.0, "dag": {}}, id="float-version"),
+ pytest.param(["not", "a", "mapping"], id="non-mapping-payload"),
+ ],
+)
+def test_build_diff_returns_unavailable_without_a_usable_schema_version(base_data) -> None:
+ result = build_serialized_dag_diff(base_data=base_data, target_data=_build_payload(tasks=[]))
+
+ assert result["mode"] == "unavailable"
+ assert result["unavailable_reason"] == "serialized_dag_schema_version_missing"
+ assert result["serialized_dag_schema_versions"] == {"base": None, "target": 3}
+ assert result["changes"] == []
+ assert result["values"] == {"status": "unavailable"}
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize("location", ["dag", "task", "provenance"])
+def test_build_diff_discards_changes_when_json_serialization_fails(location, include_values, caplog):
+ base = _build_payload(tasks=[{"task_id": "extract"}])
+ target = copy.deepcopy(base)
+ base["dag"]["catchup"] = True
+ target["dag"]["catchup"] = False
+ base_provenance = {}
+ target_provenance = {}
+ if location == "provenance":
+ before, after = base_provenance, target_provenance
+ elif location == "task":
+ before, after = base["dag"]["tasks"][0]["__var"], target["dag"]["tasks"][0]["__var"]
+ else:
+ before, after = base["dag"], target["dag"]
+ before["opaque"] = None
+ after["opaque"] = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+ result = build_serialized_dag_diff(
+ base_data=base,
+ target_data=target,
+ base_provenance=base_provenance,
+ target_provenance=target_provenance,
+ include_values=include_values,
+ )
+
+ assert result["mode"] == "unavailable"
+ assert result["unavailable_reason"] == "serialized_dag_json_encoding_failed"
+ assert result["changes"] == []
+ assert result["truncated"] is False
+ assert {
+ "event": "Serialized Dag diff JSON encoding failed",
+ "base_schema_version": 3,
+ "target_schema_version": 3,
+ } in caplog
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize("failed_payload", ["base", "target"])
+@mock.patch.object(dag_version_diff, "_canonicalize_payload_v1", autospec=True)
+def test_build_diff_returns_unavailable_on_canonicalization_recursion(
+ canonicalize, failed_payload, include_values, caplog
+):
+ error = RecursionError("secret payload")
+ canonicalize.side_effect = [error] if failed_payload == "base" else [{}, error]
+
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[]),
+ target_data=_build_payload(tasks=[]),
+ include_values=include_values,
+ )
+
+ assert result["mode"] == "unavailable"
+ assert result["unavailable_reason"] == "serialized_dag_canonicalization_failed"
+ assert result["changes"] == []
+ assert result["truncated"] is False
+ assert result["values"]["status"] == "unavailable"
+ assert {"event": "Serialized Dag diff canonicalization failed", "error_type": "RecursionError"} in caplog
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize("after_first_change", [False, True])
+@mock.patch.object(dag_version_diff, "_collect_changes", autospec=True)
+def test_build_diff_discards_changes_on_collection_recursion(
+ collect, after_first_change, include_values, caplog
+):
+ def fail_collection(before, after, *, path, collector):
+ if after_first_change:
+ collector.add(path=("dag", "catchup"), operation="changed", before=False, after=True)
+ raise RecursionError("secret payload")
+
+ collect.side_effect = fail_collection
+
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[]),
+ target_data=_build_payload(tasks=[]),
+ include_values=include_values,
+ )
+
+ assert result["mode"] == "unavailable"
+ assert result["unavailable_reason"] == "serialized_dag_recursion_limit_exceeded"
+ assert result["changes"] == []
+ assert result["truncated"] is False
+ assert result["values"]["status"] == "unavailable"
+ assert "Serialized Dag diff recursion limit exceeded" in caplog
+
+
+@pytest.mark.parametrize("error_type", [AttributeError, KeyError, RuntimeError, TypeError, ValueError])
+@mock.patch.object(dag_version_diff, "_get_category", autospec=True)
+def test_build_diff_propagates_comparison_bugs(mock_get_category, error_type):
+ mock_get_category.side_effect = error_type("comparison bug")
+ base = _build_payload(tasks=[])
+ target = copy.deepcopy(base)
+ target["dag"]["catchup"] = False
+
+ with pytest.raises(error_type, match="comparison bug"):
+ build_serialized_dag_diff(base_data=base, target_data=target)
+
+
+@pytest.mark.parametrize(
+ ("max_changes", "expected_message"),
+ [
+ (0, "max_changes must be a positive integer"),
+ (1.5, "max_changes must be a positive integer"),
+ (1.0, "max_changes must be a positive integer"),
+ (float("nan"), "max_changes must be a positive integer"),
+ (float("inf"), "max_changes must be a positive integer"),
+ (True, "max_changes must be a positive integer"),
+ ("1", "max_changes must be a positive integer"),
+ (None, "max_changes must be a positive integer"),
+ ([], "max_changes must be a positive integer"),
+ (MAX_ALLOWED_CHANGES + 1, f"max_changes must not exceed {MAX_ALLOWED_CHANGES}"),
+ ],
+)
+def test_build_diff_rejects_invalid_change_bounds(max_changes, expected_message):
+ with pytest.raises(ValueError, match=expected_message):
+ build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[]), target_data=_build_payload(tasks=[]), max_changes=max_changes
+ )
+
+
+def test_build_diff_redacts_sensitive_data_without_values() -> None:
+ dependency = {
+ "dependency_type": "sensor",
+ "dependency_id": "secret-dependency",
+ "source": "secret-upstream",
+ "target": "example",
+ "label": "secret-label",
+ }
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(
+ tasks=[{"task_id": "secret-task", "retries": 1}],
+ tags=["secret-old-tag"],
+ dependencies=[dependency],
+ ),
+ target_data=_build_payload(
+ tasks=[
+ {"task_id": "secret-task", "retries": 2},
+ {"task_id": "secret-new-task", "retries": 1},
+ ],
+ tags=["secret-new-tag"],
+ ),
+ )
+
+ assert result["mode"] == "observed_state"
+ assert any(change["path"] == "/dag/tasks/*/retries" for change in result["changes"])
+ assert any(change["path"] == "/dag/tasks/*" for change in result["changes"])
+ assert any(change["path"] == "/dag/tags/*" for change in result["changes"])
+ assert any(change["path"] == "/dag/dag_dependencies/*" for change in result["changes"])
+ assert all(
+ "before_digest" not in change
+ and "after_digest" not in change
+ and "before_value" not in change
+ and "after_value" not in change
+ for change in result["changes"]
+ )
+ encoded_result = json.dumps(result)
+ for sensitive_value in (
+ "secret-task",
+ "secret-new-task",
+ "secret-old-tag",
+ "secret-new-tag",
+ "secret-dependency",
+ "secret-upstream",
+ "secret-label",
+ ):
+ assert sensitive_value not in encoded_result
+
+
+def test_build_diff_collapses_custom_task_fields_without_values() -> None:
+ custom_field = "secret_callback_field"
+ base_task = {
+ "task_id": "extract",
+ "template_fields": [custom_field],
+ custom_field: "old-secret",
+ }
+ target_task = {
+ "task_id": "extract",
+ "template_fields": [custom_field],
+ custom_field: "new-secret",
+ }
+
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[base_task]),
+ target_data=_build_payload(tasks=[target_task]),
+ )
+
+ assert result["changes"] == [
+ {
+ "path": "/dag/tasks/*/custom_fields",
+ "operation": "changed",
+ "category": "task",
+ "impact": "execution",
+ "occurrence_count": 1,
+ }
+ ]
+ encoded_result = json.dumps(result)
+ assert custom_field not in encoded_result
+ assert "old-secret" not in encoded_result
+ assert "new-secret" not in encoded_result
+
+
+def test_build_diff_includes_custom_task_fields_with_values() -> None:
+ custom_field = "secret_callback_field"
+ base_task = {
+ "task_id": "extract",
+ "template_fields": [custom_field],
+ custom_field: "old-secret",
+ }
+ target_task = {
+ "task_id": "extract",
+ "template_fields": [custom_field],
+ custom_field: "new-secret",
+ }
+
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[base_task]),
+ target_data=_build_payload(tasks=[target_task]),
+ include_values=True,
+ )
+
+ # Authorization decides what a change carries, not which changes exist, so custom fields stay
+ # aggregated into the one record the redacted walk produces and the values ride on it.
+ change = next(change for change in result["changes"] if change["path"].endswith("custom_fields"))
+ assert change["path"] == "/dag/tasks/extract/custom_fields"
+ assert change["category"] == "task"
+ assert change["impact"] == "execution"
+ assert change["before_value"] == {custom_field: "old-secret"}
+ assert change["after_value"] == {custom_field: "new-secret"}
+
+
+@pytest.mark.parametrize(
+ (
+ "base_data",
+ "target_data",
+ "hidden_identifier",
+ "expected_path",
+ "expected_operation",
+ "expected_category",
+ "expected_impact",
+ ),
+ [
+ pytest.param(
+ _build_payload(tasks=[{"task_id": "deadline", "retries": 1}]),
+ _build_payload(tasks=[{"task_id": "deadline", "retries": 2}]),
+ "deadline",
+ "/dag/tasks/*/retries",
+ "changed",
+ "task",
+ "execution",
+ id="task-id",
+ ),
+ pytest.param(
+ _build_payload(tasks=[]),
+ _build_payload(tasks=[], tags=["deadline"]),
+ "deadline",
+ "/dag/tags/*",
+ "added",
+ "metadata",
+ "metadata",
+ id="tag",
+ ),
+ pytest.param(
+ _build_payload(tasks=[]),
+ _build_payload(
+ tasks=[],
+ dependencies=[
+ {
+ "dependency_type": "sensor",
+ "dependency_id": "upstream-task",
+ "source": "upstream-dag",
+ "target": "example",
+ "label": "secret_callback",
+ }
+ ],
+ ),
+ "secret_callback",
+ "/dag/dag_dependencies/*",
+ "added",
+ "dependency",
+ "execution",
+ id="dependency",
+ ),
+ ],
+)
+def test_build_diff_classification_ignores_redacted_identifiers(
+ base_data,
+ target_data,
+ hidden_identifier,
+ expected_path,
+ expected_operation,
+ expected_category,
+ expected_impact,
+) -> None:
+ result = build_serialized_dag_diff(base_data=base_data, target_data=target_data)
+
+ assert result["changes"] == [
+ {
+ "path": expected_path,
+ "operation": expected_operation,
+ "category": expected_category,
+ "impact": expected_impact,
+ "occurrence_count": 1,
+ }
+ ]
+ assert hidden_identifier not in json.dumps(result)
+
+
+def test_build_diff_preserves_known_task_field_classification_without_values() -> None:
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[{"task_id": "extract", "outlets": []}]),
+ target_data=_build_payload(tasks=[{"task_id": "extract", "outlets": ["asset"]}]),
+ )
+
+ assert result["changes"] == [
+ {
+ "path": "/dag/tasks/*/outlets",
+ "operation": "changed",
+ "category": "asset",
+ "impact": "execution",
+ "occurrence_count": 1,
+ }
+ ]
+
+
+def test_build_diff_collapses_arbitrary_mappings_without_values() -> None:
+ base = _build_payload(tasks=[])
+ target = _build_payload(tasks=[])
+ base["dag"]["default_args"] = {"secret-argument": "old-secret"}
+ target["dag"]["default_args"] = {"secret-argument": "new-secret"}
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ assert result["changes"] == [
+ {
+ "path": "/dag/default_args",
+ "operation": "changed",
+ "category": "task",
+ "impact": "execution",
+ "occurrence_count": 1,
+ }
+ ]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_never_names_a_user_chosen_default_args_key(include_values):
+ base = _build_payload(tasks=[])
+ target = _build_payload(tasks=[])
+ base["dag"]["default_args"] = {"__type": "dict", "__var": {"secret_argument": "old"}}
+ target["dag"]["default_args"] = {"__type": "dict", "__var": {"secret_argument": "new"}}
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ assert change["path"] == "/dag/default_args/custom_fields"
+ assert (change["operation"], change["category"], change["impact"]) == (
+ "changed",
+ "task",
+ "execution",
+ )
+ if not include_values:
+ assert "secret_argument" not in json.dumps(result)
+
+
+def test_build_diff_keeps_default_args_conservative_for_a_mixed_edit() -> None:
+ """A metadata key and an unrecognised one are separate records, so neither impact is lost."""
+ base = _build_payload(tasks=[])
+ target = _build_payload(tasks=[])
+ base["dag"]["default_args"] = {"__type": "dict", "__var": {"owner": "alice", "secret": "old"}}
+ target["dag"]["default_args"] = {"__type": "dict", "__var": {"owner": "bob", "secret": "new"}}
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ assert [(change["path"], change["category"], change["impact"]) for change in result["changes"]] == [
+ ("/dag/default_args/owner", "metadata", "metadata"),
+ ("/dag/default_args/custom_fields", "task", "execution"),
+ ]
+ assert "secret" not in json.dumps(result)
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_collapses_unknown_root_sections(include_values):
+ base = _build_payload(tasks=[])
+ target = copy.deepcopy(base)
+ base["secret_root_section"] = {"secret-argument": "old-secret"}
+ target["secret_root_section"] = {"secret-argument": "new-secret"}
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ assert change["path"] == ("/secret_root_section" if include_values else "/custom_fields")
+ assert (change["operation"], change["category"], change["impact"]) == (
+ "changed",
+ "unknown",
+ "unknown",
+ )
+ if not include_values:
+ assert "secret" not in json.dumps(result)
+
+
+def test_build_diff_reports_unkeyed_lists_as_one_stable_change() -> None:
+ base = _build_payload(tasks=[])
+ target = _build_payload(tasks=[])
+ base["dag"]["custom_list"] = [{"name": "old"}]
+ target["dag"]["custom_list"] = [{"name": "new"}]
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ assert change["path"] == "/dag/custom_list"
+ assert change["operation"] == "changed"
+ assert change["category"] == "unknown"
+ assert change["impact"] == "unknown"
+ assert "before_digest" not in change
+ assert "after_digest" not in change
+
+
+def test_build_diff_reports_deadline_lists_as_one_stable_change() -> None:
+ base = _build_payload(tasks=[])
+ target = _build_payload(tasks=[])
+ base["dag"]["deadline"] = [{"name": "old", "interval": 60}]
+ target["dag"]["deadline"] = [{"name": "new", "interval": 60}]
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ assert result["mode"] == "observed_state"
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ assert change["path"] == "/dag/deadline"
+ assert change["operation"] == "changed"
+ assert change["category"] == "deadline"
+ assert change["impact"] == "execution"
+
+
+def test_build_diff_preserves_order_sensitive_string_lists() -> None:
+ base = _build_payload(tasks=[])
+ target = _build_payload(tasks=[])
+ base["dag"]["template_searchpath"] = ["first", "second"]
+ target["dag"]["template_searchpath"] = ["second", "first"]
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ assert change["path"] == "/dag/template_searchpath"
+ assert change["operation"] == "changed"
+
+
+@pytest.mark.parametrize(
+ ("base_task", "target_task"),
+ [
+ ({"task_id": "extract", "retries": 2}, {"task_id": "extract"}),
+ (
+ {"task_id": "extract", "_is_mapped": True, "retries": 2, "partial_kwargs": {"retries": 2}},
+ {"task_id": "extract", "_is_mapped": True, "partial_kwargs": {}},
+ ),
+ ],
+)
+def test_build_diff_normalizes_v3_client_defaults(base_task, target_task) -> None:
+ base = _build_payload(tasks=[base_task])
+ base["__version"] = 2
+ target = _build_payload(tasks=[target_task])
+ target["client_defaults"] = {"tasks": {"retries": 2}}
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ assert result["serialized_dag_schema_versions"] == {"base": 2, "target": 3}
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize("schema_version", [2, 3])
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize(
+ ("field", "value"), [("retries", 0), ("depends_on_past", False), ("ui_fgcolor", "#000")]
+)
+def test_build_diff_normalizes_schema_defaults(schema_version, include_values, field, value) -> None:
+ base = _build_payload(tasks=[{"task_id": "extract", field: value}])
+ base["__version"] = schema_version
+ target = _build_payload(tasks=[{"task_id": "extract"}])
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize("is_mapped", [False, True])
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_normalizes_explicit_schema_default_overrides(is_mapped, include_values) -> None:
+ base_task: dict[str, Any] = {"task_id": "extract"}
+ target_task: dict[str, Any] = {"task_id": "extract", "retries": 0}
+ if is_mapped:
+ base_task.update(_is_mapped=True, partial_kwargs={})
+ target_task = {"task_id": "extract", "_is_mapped": True, "partial_kwargs": {"retries": 0}}
+ base = _build_payload(tasks=[base_task])
+ target = _build_payload(tasks=[target_task])
+ target["client_defaults"] = {"tasks": {"retries": 3}}
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_diff_selects_mapping_from_the_raw_operator(include_values):
+ with DAG("diff_regression", schedule=None) as dag:
+ BashOperator.partial(task_id="extract").expand(bash_command=["first", "second"])
+ target = _serialize_dag(dag)
+ operator_schema = {"$ref": "#/definitions/operator", "definitions": load_dag_schema_dict()["definitions"]}
+ task = _get_task(target)
+ schema_defaults = DagSerialization.get_schema_defaults("operator")
+ for field in operator_schema["definitions"]["operator"]["required"]:
+ if field not in task:
+ task[field] = schema_defaults[field]
+ defaults = target.setdefault("client_defaults", {}).setdefault("tasks", {})
+ mapping_fields = ("_is_mapped", "partial_kwargs", "expand_input")
+ defaults.update({field: copy.deepcopy(task[field]) for field in mapping_fields})
+ base = copy.deepcopy(target)
+ for field in mapping_fields:
+ _get_task(base).pop(field)
+ for payload in (base, target):
+ DagSerialization.validate_schema(payload)
+ Draft7Validator(operator_schema).validate(_get_task(payload))
+ restored = [
+ DagSerialization.from_dict(copy.deepcopy(payload)).task_dict["extract"] for payload in (base, target)
+ ]
+ assert restored[0].is_mapped is False
+ assert restored[1].is_mapped is True
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ task_id = "extract" if include_values else "*"
+ change = next(
+ change for change in result["changes"] if change["path"] == f"/dag/tasks/{task_id}/_is_mapped"
+ )
+ assert change["impact"] == "execution"
+ if include_values:
+ assert change["before_value"] is False
+ assert change["after_value"] is True
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_ignores_shadowed_mapped_client_defaults(include_values) -> None:
+ task = {"task_id": "extract", "_is_mapped": True, "partial_kwargs": {"retries": 5}}
+ base = _build_payload(tasks=[task])
+ target = _build_payload(tasks=[task])
+ target["client_defaults"] = {"tasks": {"retries": 3}}
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+def test_build_diff_reports_effective_mapped_default_changes() -> None:
+ task = {"task_id": "extract", "_is_mapped": True, "partial_kwargs": {}}
+ base = _build_payload(tasks=[task])
+ base["client_defaults"] = {"tasks": {"retries": 3}}
+ target = _build_payload(tasks=[task])
+ target["client_defaults"] = {"tasks": {"retries": 5}}
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=True)
+
+ assert result["mode"] == "observed_state"
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ assert change["path"] == "/dag/tasks/extract/partial_kwargs/retries"
+ assert change["operation"] == "changed"
+ assert change["before_value"] == 3
+ assert change["after_value"] == 5
+
+
+def _build_mapped_precedence_payload(
+ *, field: str, top_level_value: int | str, client_default: int | str, partial_kwargs: dict | None
+) -> dict:
+ with DAG("mapped_precedence", schedule=None) as dag:
+ BashOperator.partial(task_id="mapped").expand(bash_command=["first", "second"])
+ payload = json.loads(json.dumps(DagSerialization.to_dict(dag)))
+ task = payload["dag"]["tasks"][0]["__var"]
+ task.pop("partial_kwargs", None)
+ task[field] = top_level_value
+ if partial_kwargs is not None:
+ task["partial_kwargs"] = partial_kwargs
+ payload["client_defaults"] = {"tasks": {field: client_default}}
+ return payload
+
+
+def _resolve_runtime_partial_value(payload: dict, field: str) -> int | str:
+ hydrated = DagSerialization.from_dict(copy.deepcopy(payload))
+ return hydrated.task_dict["mapped"].partial_kwargs[field]
+
+
+@pytest.mark.parametrize(
+ ("field", "before", "after", "client_default"),
+ [("retries", 2, 3, 5), ("pool", "base_pool", "target_pool", "client_pool")],
+)
+def test_build_diff_reports_top_level_shadowing_without_partial_kwargs(field, before, after, client_default):
+ base = _build_mapped_precedence_payload(
+ field=field, top_level_value=before, client_default=client_default, partial_kwargs=None
+ )
+ target = _build_mapped_precedence_payload(
+ field=field, top_level_value=after, client_default=client_default, partial_kwargs=None
+ )
+ # populate_operator only folds client defaults into partial_kwargs for a payload that carries
+ # the key, so without it the top-level values survive and the two versions really do differ.
+ assert (_resolve_runtime_partial_value(base, field), _resolve_runtime_partial_value(target, field)) == (
+ before,
+ after,
+ )
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=True)
+
+ assert result["mode"] == "observed_state"
+ assert [
+ (change["path"], change["before_value"], change["after_value"]) for change in result["changes"]
+ ] == [(f"/dag/tasks/mapped/partial_kwargs/{field}", before, after)]
+
+
+@pytest.mark.parametrize(
+ ("field", "before", "after", "client_default"),
+ [("retries", 2, 3, 5), ("pool", "base_pool", "target_pool", "client_pool")],
+)
+def test_build_diff_ignores_top_level_shadowing_behind_empty_partial_kwargs(
+ field, before, after, client_default
+):
+ base = _build_mapped_precedence_payload(
+ field=field, top_level_value=before, client_default=client_default, partial_kwargs={}
+ )
+ target = _build_mapped_precedence_payload(
+ field=field, top_level_value=after, client_default=client_default, partial_kwargs={}
+ )
+ # The present key makes populate_operator apply the client default instead, shadowing both
+ # top-level values, so the versions resolve identically and the diff must stay silent.
+ assert (_resolve_runtime_partial_value(base, field), _resolve_runtime_partial_value(target, field)) == (
+ client_default,
+ client_default,
+ )
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=True)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize(("field", "default_type"), [("retry_delay", timedelta), ("start_date", datetime)])
+def test_diff_distinguishes_implicit_typed_defaults_from_template_numbers(
+ include_values, field, default_type
+):
+ with DAG("diff_regression", schedule=None, start_date=datetime(2024, 1, 1, tzinfo=timezone.utc)) as dag:
+ BashOperator(task_id="extract", bash_command="echo hello")
+ base = _serialize_dag(dag)
+ _get_task(base)["template_fields"].append(field)
+ _get_task(base).pop(field, None)
+ base.get("client_defaults", {}).get("tasks", {}).pop(field, None)
+ target = copy.deepcopy(base)
+ value = (
+ base["dag"][field]
+ if field == "start_date"
+ else DagSerialization.get_schema_defaults("operator")[field]
+ )
+ _get_task(target)[field] = value
+ for payload in (base, target):
+ DagSerialization.validate_schema(payload)
+ restored = [
+ DagSerialization.from_dict(copy.deepcopy(payload)).task_dict["extract"] for payload in (base, target)
+ ]
+ assert isinstance(getattr(restored[0], field), default_type)
+ assert type(getattr(restored[1], field)) is float
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ task_id = "extract" if include_values else "*"
+ assert change["path"] == f"/dag/tasks/{task_id}/{field}"
+ assert change["impact"] == "execution"
+ if include_values:
+ assert change["before_digest"] != change["after_digest"]
+
+
+@pytest.mark.parametrize("field", sorted(_OPERATOR_TIMEDELTA_FIELDS))
+@pytest.mark.parametrize("duration_source", ["encoded_partial", "plain_partial", "client_default"])
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_distinguishes_mapped_template_numbers_from_timedeltas(
+ field, duration_source, include_values
+):
+ with DAG("example", schedule=None) as dag:
+ BashOperator.partial(task_id="extract").expand(bash_command=["echo hello"])
+ base = json.loads(json.dumps(DagSerialization.to_dict(dag)))
+ base.get("client_defaults", {}).get("tasks", {}).pop(field, None)
+ base_task = base["dag"]["tasks"][0]["__var"]
+ base_task["partial_kwargs"].pop(field, None)
+ base_task["template_fields"].append(field)
+ base_task[field] = 60.0
+ target = copy.deepcopy(base)
+ target_task = target["dag"]["tasks"][0]["__var"]
+ target_task.pop(field)
+ if duration_source == "client_default":
+ target.setdefault("client_defaults", {}).setdefault("tasks", {})[field] = 60.0
+ else:
+ target_task["partial_kwargs"][field] = (
+ {"__type": "timedelta", "__var": 60.0} if duration_source == "encoded_partial" else 60.0
+ )
+ stored_payloads = copy.deepcopy((base, target))
+ for payload, expected in ((base, 60.0), (target, timedelta(seconds=60))):
+ DagSerialization.validate_schema(payload)
+ actual = getattr(DagSerialization.from_dict(copy.deepcopy(payload)).task_dict["extract"], field)
+ assert actual == expected
+ assert type(actual) is type(expected)
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ task_id = "extract" if include_values else "*"
+ assert change["path"] == f"/dag/tasks/{task_id}/partial_kwargs/{field}"
+ assert change["operation"] == "changed"
+ if include_values:
+ assert change["before_value"] == 60.0
+ assert change["after_value"] == {"__type": "timedelta", "__var": 60.0}
+ assert (base, target) == stored_payloads
+
+
+@pytest.mark.parametrize("field", sorted(_OPERATOR_TIMEDELTA_FIELDS))
+@pytest.mark.parametrize("duration_source", ["encoded_partial", "plain_partial", "client_default"])
+def test_build_diff_normalizes_mapped_timedelta_defaults(field, duration_source) -> None:
+ with DAG("example", schedule=None) as dag:
+ BashOperator.partial(task_id="extract").expand(bash_command=["echo hello"])
+ base = json.loads(json.dumps(DagSerialization.to_dict(dag)))
+ base.get("client_defaults", {}).get("tasks", {}).pop(field, None)
+ base_task = base["dag"]["tasks"][0]["__var"]
+ base_task["template_fields"].append(field)
+ base_task[field] = 60.0
+ base_task["partial_kwargs"][field] = {"__type": "timedelta", "__var": 60.0}
+ target = copy.deepcopy(base)
+ target_task = target["dag"]["tasks"][0]["__var"]
+ target_task[field] = 120.0
+ if duration_source == "client_default":
+ target_task["partial_kwargs"].pop(field)
+ target.setdefault("client_defaults", {}).setdefault("tasks", {})[field] = 60.0
+ elif duration_source == "plain_partial":
+ target_task["partial_kwargs"][field] = 60.0
+ for payload in (base, target):
+ DagSerialization.validate_schema(payload)
+ assert getattr(
+ DagSerialization.from_dict(copy.deepcopy(payload)).task_dict["extract"], field
+ ) == timedelta(seconds=60)
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=True)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize(
+ "field",
+ sorted(_OPERATOR_TIMEDELTA_FIELDS & DagSerialization.get_schema_defaults("operator").keys()),
+)
+@pytest.mark.parametrize("stores_partial_kwargs", [True, False])
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_normalizes_mapped_timedelta_schema_defaults(field, stores_partial_kwargs, include_values):
+ default = DagSerialization.get_schema_defaults("operator")[field]
+ with DAG("example", schedule=None) as dag:
+ BashOperator.partial(task_id="extract").expand(bash_command=["echo hello"])
+ base = json.loads(json.dumps(DagSerialization.to_dict(dag)))
+ base.get("client_defaults", {}).get("tasks", {}).pop(field, None)
+ base_task = base["dag"]["tasks"][0]["__var"]
+ base_task["partial_kwargs"].pop(field, None)
+ if not stores_partial_kwargs:
+ base_task.pop("partial_kwargs")
+ target = copy.deepcopy(base)
+ target["dag"]["tasks"][0]["__var"].setdefault("partial_kwargs", {})[field] = default
+ for payload in (base, target):
+ DagSerialization.validate_schema(payload)
+ assert getattr(
+ DagSerialization.from_dict(copy.deepcopy(payload)).task_dict["extract"], field
+ ) == timedelta(seconds=default)
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_diff_preserves_mapped_resource_types(include_values):
+ resources = Resources(cpus=1, ram=512, disk=512, gpus=0)
+ payloads = []
+ for value in (resources, resources.to_dict()):
+ with DAG("diff_regression", schedule=None) as dag:
+ task = BashOperator.partial(task_id="extract", resources=value).expand(
+ bash_command=["echo hello"]
+ )
+ payloads.append(_serialize_dag(dag))
+ if isinstance(value, Resources):
+ assert isinstance(task.unmap({"bash_command": "echo hello"}).resources, Resources)
+ else:
+ with pytest.raises(TypeError):
+ task.unmap({"bash_command": "echo hello"})
+
+ restored = [
+ DagSerialization.from_dict(copy.deepcopy(payload)).task_dict["extract"] for payload in payloads
+ ]
+ assert isinstance(restored[0].resources, Resources)
+ assert isinstance(restored[1].resources, dict)
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ assert result["mode"] == "observed_state"
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ task_id = "extract" if include_values else "*"
+ assert change["path"] == f"/dag/tasks/{task_id}/partial_kwargs/resources"
+ assert change["impact"] == "execution"
+ if include_values:
+ assert change["before_digest"] != change["after_digest"]
+ else:
+ assert "before_value" not in change
+ assert "after_value" not in change
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize("encoded_location", ["partial_kwargs", "top_level"])
+@pytest.mark.parametrize(
+ "executor_config",
+ [
+ {"queue": "same"},
+ {"nested": {"limits": [{"cpu": 2}]}},
+ {"nested": {"__type": "timedelta", "__var": 60.0}},
+ ],
+)
+def test_build_diff_normalizes_mapped_plain_mappings(executor_config, encoded_location, include_values):
+ with DAG("example", schedule=None) as dag:
+ BashOperator.partial(task_id="extract", executor_config=executor_config).expand(
+ bash_command=["echo hello"]
+ )
+ base = json.loads(json.dumps(DagSerialization.to_dict(dag)))
+ if encoded_location == "top_level":
+ encoded_task = base["dag"]["tasks"][0]["__var"]
+ encoded_task["executor_config"] = encoded_task["partial_kwargs"].pop("executor_config")
+ target = copy.deepcopy(base)
+ target["dag"]["tasks"][0]["__var"]["partial_kwargs"]["executor_config"] = executor_config
+ stored_payloads = copy.deepcopy((base, target))
+ for payload in (base, target):
+ DagSerialization.validate_schema(payload)
+ assert (
+ DagSerialization.from_dict(copy.deepcopy(payload)).task_dict["extract"].executor_config
+ == executor_config
+ )
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+ assert (base, target) == stored_payloads
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_preserves_literal_mapping_client_defaults(include_values):
+ executor_config = {"__type": "dict", "__var": {"queue": "same"}}
+ with DAG("example", schedule=None) as dag:
+ BashOperator.partial(task_id="extract").expand(bash_command=["echo hello"])
+ base = json.loads(json.dumps(DagSerialization.to_dict(dag)))
+ base["dag"]["tasks"][0]["__var"]["partial_kwargs"].pop("executor_config", None)
+ base["client_defaults"] = {"tasks": {"executor_config": executor_config}}
+ target = copy.deepcopy(base)
+ target["dag"]["tasks"][0]["__var"]["partial_kwargs"]["executor_config"] = OperatorSerialization.serialize(
+ executor_config
+ )
+ for payload in (base, target):
+ DagSerialization.validate_schema(payload)
+ assert (
+ DagSerialization.from_dict(copy.deepcopy(payload)).task_dict["extract"].executor_config
+ == executor_config
+ )
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize(
+ ("field", "value", "is_template"),
+ [
+ ("resources", {"cpus": 1}, False),
+ ("execution_timeout", timedelta(seconds=60), False),
+ ("executor_config", {"pod": {"cpu": 1}}, True),
+ ],
+)
+def test_build_diff_normalizes_encoded_outer_mapped_fields(field, value, is_template, include_values):
+ with DAG("example", schedule=None) as dag:
+ BashOperator.partial(task_id="extract").expand(bash_command=["echo hello"])
+ base = json.loads(json.dumps(DagSerialization.to_dict(dag)))
+ base.get("client_defaults", {}).get("tasks", {}).pop(field, None)
+ encoded_value = json.loads(json.dumps(BaseSerialization.serialize(value)))
+ expected = encoded_value if is_template else value
+ base_task = base["dag"]["tasks"][0]["__var"]
+ if is_template:
+ base_task["template_fields"].append(field)
+ base_task["partial_kwargs"][field] = BaseSerialization.serialize(expected)
+ target = copy.deepcopy(base)
+ target_task = target["dag"]["tasks"][0]["__var"]
+ target_task["partial_kwargs"].pop(field)
+ target_task[field] = encoded_value
+ for payload in (base, target):
+ DagSerialization.validate_schema(payload)
+ assert (
+ getattr(DagSerialization.from_dict(copy.deepcopy(payload)).task_dict["extract"], field)
+ == expected
+ )
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize(
+ "base_data",
+ [
+ {"__version": 3, "dag": {"tasks": []}, "client_defaults": []},
+ {"__version": 3, "dag": {"tasks": []}, "client_defaults": {"tasks": []}},
+ {"__version": 3, "dag": {"tasks": {}}, "client_defaults": {"tasks": {}}},
+ {"__version": 3, "dag": {"tasks": [{}]}, "client_defaults": {"tasks": {}}},
+ _build_payload(tasks=[{"task_id": "extract", "_is_mapped": True, "partial_kwargs": []}]),
+ ],
+)
+def test_build_diff_rejects_malformed_client_defaults(base_data) -> None:
+ result = build_serialized_dag_diff(base_data=base_data, target_data=_build_payload(tasks=[]))
+
+ assert result["mode"] == "unavailable"
+ assert result["changes"] == []
+ assert result["unavailable_reason"] == "serialized_dag_canonicalization_failed"
+
+
+def test_build_diff_ignores_exact_duplicate_dependencies() -> None:
+ dependency = {
+ "dependency_type": "task",
+ "dependency_id": "extract-load",
+ "source": "extract",
+ "target": "load",
+ "label": "extract-load",
+ }
+
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[], dependencies=[dependency, dependency]),
+ target_data=_build_payload(tasks=[], dependencies=[dependency]),
+ )
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+def test_build_diff_collapses_dependencies_differing_only_by_label() -> None:
+ def build_dependency(label: str) -> dict:
+ return {
+ "dependency_type": "trigger",
+ "dependency_id": "fire",
+ "source": "example",
+ "target": "downstream",
+ "label": label,
+ }
+
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[], dependencies=[build_dependency("Old"), build_dependency("New")]),
+ target_data=_build_payload(tasks=[], dependencies=[build_dependency("Old")]),
+ )
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+def test_build_diff_dependency_keys_do_not_collide_on_delimiters() -> None:
+ first_dependency = {
+ "dependency_type": "trigger",
+ "dependency_id": "id",
+ "source": "a",
+ "target": "b|c",
+ "label": "d",
+ }
+ second_dependency = {
+ "dependency_type": "trigger",
+ "dependency_id": "id",
+ "source": "a|b",
+ "target": "c",
+ "label": "d",
+ }
+
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[], dependencies=[first_dependency, second_dependency]),
+ target_data=_build_payload(tasks=[], dependencies=[second_dependency, first_dependency]),
+ )
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+def test_build_diff_detects_json_type_change() -> None:
+ base = _build_payload(tasks=[])
+ target = _build_payload(tasks=[])
+ base["dag"]["flag"] = True
+ target["dag"]["flag"] = 1
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ assert change["path"] == "/dag/flag"
+ assert change["operation"] == "changed"
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_diff_preserves_task_template_named_type(include_values):
+ payloads = []
+ for value in ("old", "new"):
+ with DAG("diff_regression", schedule=None) as dag:
+ task = BashOperator(task_id="extract", bash_command="echo hello")
+ task.template_fields = (*task.template_fields, "__type")
+ setattr(task, "__type", value)
+ payload = _serialize_dag(dag)
+ assert (
+ getattr(DagSerialization.from_dict(copy.deepcopy(payload)).task_dict["extract"], "__type")
+ == value
+ )
+ payloads.append(payload)
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ assert result["mode"] == "observed_state"
+ assert len(result["changes"]) == 1
+ task_id = "extract" if include_values else "*"
+ change = result["changes"][0]
+ assert change["path"] == f"/dag/tasks/{task_id}/custom_fields"
+ assert change["impact"] == "execution"
+ if include_values:
+ assert change["before_value"]["__type"] == "old"
+ assert change["after_value"]["__type"] == "new"
+ else:
+ assert "before_value" not in change
+ assert "after_value" not in change
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize("provenance_changed", [False, True])
+def test_diff_ignores_wrapper_types_missing_on_both_sides(include_values, provenance_changed):
+ base = _build_params_payload({})
+ base["dag"]["tasks"][0].pop("__type")
+ base["dag"].pop("task_group")
+ target = copy.deepcopy(base)
+ DagSerialization.validate_schema(base)
+ assert DagSerialization.from_dict(copy.deepcopy(base)).task_dict == {}
+ target_version = "new" if provenance_changed else "old"
+
+ result = build_serialized_dag_diff(
+ base_data=base,
+ target_data=target,
+ base_provenance={"bundle_version": "old"},
+ target_provenance={"bundle_version": target_version},
+ include_values=include_values,
+ max_changes=1,
+ )
+
+ assert result["mode"] == "observed_state"
+ assert [change["path"] for change in result["changes"]] == (
+ ["/provenance/bundle_version"] if provenance_changed else []
+ )
+ assert result["truncated"] is False
+ if provenance_changed and include_values:
+ assert result["changes"][0]["before_value"] == "old"
+ assert result["changes"][0]["after_value"] == "new"
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize("operation", ["added", "removed"])
+def test_diff_preserves_one_sided_wrapper_types(include_values, operation):
+ base = _build_params_payload({})
+ base["dag"].pop("task_group")
+ target = copy.deepcopy(base)
+ (base if operation == "added" else target)["dag"]["tasks"][0].pop("__type")
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ task_id = "extract" if include_values else "*"
+ assert change["path"] == f"/dag/tasks/{task_id}/__type"
+ assert change["operation"] == operation
+ if include_values:
+ present_side = "after" if operation == "added" else "before"
+ missing_side = "before" if operation == "added" else "after"
+ assert change[f"{present_side}_value"] == "operator"
+ assert change[f"{missing_side}_digest"] is None
+ assert f"{missing_side}_value" not in change
+
+
+def test_build_diff_classifies_callback_as_execution() -> None:
+ base = _build_payload(tasks=[])
+ target = _build_payload(tasks=[])
+ base["dag"]["on_failure_callback"] = "old"
+ target["dag"]["on_failure_callback"] = "new"
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ assert change["path"] == "/dag/on_failure_callback"
+ assert change["category"] == "callback"
+ assert change["impact"] == "execution"
+
+
+def _build_v1_payload() -> dict:
+ return {
+ "__version": 1,
+ "dag": {
+ "_dag_id": "example",
+ "fileloc": "/dags/example.py",
+ "timezone": "UTC",
+ "task_group": {},
+ "tasks": [
+ {
+ "__type": "operator",
+ "__var": {
+ "task_id": "extract",
+ "_task_type": "EmptyOperator",
+ "_task_module": "airflow.providers.standard.operators.empty",
+ "template_fields": [],
+ "ui_color": "#e8f7e4",
+ },
+ }
+ ],
+ "dag_dependencies": [],
+ },
+ }
+
+
+def _build_equivalent_v3_payload() -> dict:
+ return {
+ "__version": 3,
+ "dag": {
+ "dag_id": "example",
+ "fileloc": "/dags/example.py",
+ "timezone": "UTC",
+ "timetable": {
+ "__type": "airflow.timetables.simple.NullTimetable",
+ "__var": {},
+ },
+ "task_group": {"group_display_name": ""},
+ "tasks": [
+ {
+ "__type": "operator",
+ "__var": {
+ "task_id": "extract",
+ "task_type": "EmptyOperator",
+ "_task_module": "airflow.providers.standard.operators.empty",
+ "ui_color": "#e8f7e4",
+ },
+ }
+ ],
+ "dag_dependencies": [],
+ },
+ }
+
+
+def test_build_diff_normalizes_valid_v1_payload() -> None:
+ result = build_serialized_dag_diff(
+ base_data=_build_v1_payload(), target_data=_build_equivalent_v3_payload()
+ )
+
+ assert result["mode"] == "observed_state"
+ assert result["serialized_dag_schema_versions"] == {"base": 1, "target": 3}
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize(
+ ("legacy_fields", "current_fields", "attribute"),
+ [
+ ({"_downstream_task_ids": ["load"]}, {"downstream_task_ids": ["load"]}, "downstream_task_ids"),
+ (
+ {"_downstream_task_ids": ["load"], "downstream_task_ids": []},
+ {"downstream_task_ids": ["load"]},
+ "downstream_task_ids",
+ ),
+ ({"on_failure_callback": True}, {"has_on_failure_callback": True}, "has_on_failure_callback"),
+ (
+ {"on_failure_callback": False, "has_on_failure_callback": True},
+ {"has_on_failure_callback": False},
+ "has_on_failure_callback",
+ ),
+ ({"task_display_name": "display"}, {"_task_display_name": "display"}, "_task_display_name"),
+ ],
+)
+def test_build_diff_matches_legacy_operator_aliases(legacy_fields, current_fields, attribute):
+ task = {"task_id": "extract", "task_type": "BaseOperator", "_task_module": "airflow.sdk.bases.operator"}
+ legacy_task = {**task, **legacy_fields}
+ current_task = {**task, **current_fields}
+ assert getattr(OperatorSerialization.deserialize_operator(legacy_task), attribute) == getattr(
+ OperatorSerialization.deserialize_operator(current_task), attribute
+ )
+ base = _build_payload(tasks=[legacy_task])
+ base["__version"] = 2
+
+ result = build_serialized_dag_diff(base_data=base, target_data=_build_payload(tasks=[current_task]))
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize("is_mapped", [False, True])
+@pytest.mark.parametrize(("legacy", "current"), [(True, 2.0), (False, 0.0), (2, 2.0)])
+def test_build_diff_normalizes_retry_backoff(is_mapped, legacy, current):
+ tasks = []
+ for value in (legacy, current):
+ task = {"task_id": "extract", "retry_exponential_backoff": value}
+ if is_mapped:
+ task = {
+ "task_id": "extract",
+ "_is_mapped": True,
+ "partial_kwargs": {"retry_exponential_backoff": value},
+ }
+ tasks.append(task)
+
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[tasks[0]]), target_data=_build_payload(tasks=[tasks[1]])
+ )
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_reports_a_templated_retry_backoff_change(include_values):
+ """
+ A templated backoff must not be normalized away.
+
+ populate_operator skips deserialization for a name in ``template_fields``, so a stored ``True``
+ and a stored ``2.0`` reach the scheduler as different factors and must not compare equal here.
+ """
+
+ def payload(value):
+ return _build_payload(
+ tasks=[
+ {
+ "task_id": "extract",
+ "template_fields": ["retry_exponential_backoff"],
+ "retry_exponential_backoff": value,
+ }
+ ]
+ )
+
+ result = build_serialized_dag_diff(
+ base_data=payload(True), target_data=payload(2.0), include_values=include_values
+ )
+
+ assert [(change["path"], change["category"]) for change in result["changes"]] == [
+ ("/dag/tasks/extract/retry_exponential_backoff", "task")
+ if include_values
+ else ("/dag/tasks/*/retry_exponential_backoff", "task")
+ ]
+
+
+@pytest.mark.parametrize("is_mapped", [False, True])
+@pytest.mark.parametrize("malformed", [None, "abc", {"__type": "dict", "__var": {}}])
+def test_build_diff_degrades_only_malformed_retry_backoff(is_mapped, malformed):
+ def build(value):
+ task: dict[str, Any] = {"task_id": "extract", "_is_mapped": is_mapped}
+ fields = task.setdefault("partial_kwargs", {}) if is_mapped else task
+ fields["retry_exponential_backoff"] = value
+ return _build_payload(tasks=[task, {"task_id": "load", "retries": 1}])
+
+ target = build(malformed)
+ target["dag"]["tasks"][1]["__var"]["retries"] = 2
+
+ result = build_serialized_dag_diff(base_data=build(True), target_data=target, include_values=True)
+
+ assert result["mode"] == "observed_state"
+ changes = {change["path"]: change for change in result["changes"]}
+ assert changes["/dag/tasks/load/retries"]["after_value"] == 2
+ backoff_path = (
+ "/dag/tasks/extract/partial_kwargs/retry_exponential_backoff"
+ if is_mapped
+ else "/dag/tasks/extract/retry_exponential_backoff"
+ )
+ assert changes[backoff_path]["before_value"] == 2.0
+ assert changes[backoff_path]["after_value"] == malformed
+
+
+@pytest.mark.parametrize("is_mapped", [False, True])
+def test_build_diff_reports_effective_retry_backoff_changes(is_mapped):
+ task: dict[str, Any] = {"task_id": "extract", "_is_mapped": is_mapped}
+ base = _build_payload(tasks=[task])
+ target = copy.deepcopy(base)
+ target_fields = target["dag"]["tasks"][0]["__var"]
+ if is_mapped:
+ target_fields = target_fields.setdefault("partial_kwargs", {})
+ target_fields["retry_exponential_backoff"] = True
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=True)
+
+ assert len(result["changes"]) == 1
+ assert result["changes"][0]["before_value"] == 0.0
+ assert result["changes"][0]["after_value"] == 2.0
+
+
+def test_build_diff_preserves_shadowed_mapped_backoff():
+ task = {"task_id": "extract", "_is_mapped": True, "partial_kwargs": {"retry_exponential_backoff": True}}
+ base = _build_payload(tasks=[task])
+ target = _build_payload(tasks=[task])
+ base["client_defaults"] = {"tasks": {"retry_exponential_backoff": 3.0}}
+ target["client_defaults"] = {"tasks": {"retry_exponential_backoff": 4.0}}
+ target = copy.deepcopy(target)
+ target["dag"]["tasks"][0]["__var"]["partial_kwargs"]["retry_exponential_backoff"] = 2.0
+
+ assert build_serialized_dag_diff(base_data=base, target_data=target)["changes"] == []
+
+
+@pytest.mark.parametrize("field", ["python_callable_name", "label"])
+def test_build_diff_retains_definition_metadata(field):
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[{"task_id": "extract", field: "old"}]),
+ target_data=_build_payload(tasks=[{"task_id": "extract", field: "new"}]),
+ include_values=True,
+ )
+
+ assert len(result["changes"]) == 1
+ change = result["changes"][0]
+ if field == "python_callable_name":
+ assert change["path"] == f"/dag/tasks/extract/{field}"
+ assert change["before_value"] == "old"
+ assert change["after_value"] == "new"
+ else:
+ assert change["path"] == "/dag/tasks/extract/custom_fields"
+ assert change["before_value"] == {field: "old"}
+ assert change["after_value"] == {field: "new"}
+
+
+@pytest.mark.parametrize(
+ ("field", "default"),
+ [("fail_fast", False), ("render_template_as_native_obj", False), ("rerun_with_latest_version", None)],
+)
+def test_build_diff_normalizes_dag_defaults(field, default):
+ base = _build_payload(tasks=[])
+ base["__version"] = 2
+ base["dag"][field] = default
+ target = _build_payload(tasks=[])
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+ target["dag"][field] = True
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+ assert [change["path"] for change in result["changes"]] == [f"/dag/{field}"]
+
+
+@pytest.mark.parametrize("field", sorted(_DAG_CALLBACK_FIELDS))
+def test_build_diff_preserves_dag_callback_presence(field):
+ base = _build_payload(tasks=[])
+ target = _build_payload(tasks=[])
+ target["dag"][field] = False
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ assert result["changes"] == [
+ {
+ "path": f"/dag/{field}",
+ "operation": "added",
+ "category": "callback",
+ "impact": "execution",
+ "occurrence_count": 1,
+ }
+ ]
+
+
+@pytest.mark.parametrize("field", sorted(_DAG_CALLBACK_FIELDS))
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_normalizes_present_dag_callbacks(field, include_values):
+ with DAG("example", schedule=None) as dag:
+ BashOperator(task_id="extract", bash_command="echo hello")
+ base = json.loads(json.dumps(DagSerialization.to_dict(dag)))
+ target = copy.deepcopy(base)
+ base["dag"][field] = False
+ target["dag"][field] = True
+ for payload in (base, target):
+ assert getattr(DagSerialization.from_dict(copy.deepcopy(payload)), field) is True
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+def _handle_failure(context):
+ pass
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize("is_mapped", [False, True])
+@pytest.mark.parametrize(
+ ("field", "before", "after", "category", "impact"),
+ [
+ ("owner", "old", "new", "metadata", "metadata"),
+ ("doc", "old", "new", "metadata", "metadata"),
+ ("doc_md", "old", "new", "metadata", "metadata"),
+ ("doc_rst", "old", "new", "metadata", "metadata"),
+ ("doc_json", "{}", '{"new": true}', "metadata", "metadata"),
+ ("doc_yaml", "old: true", "new: true", "metadata", "metadata"),
+ ("ui_color", "#000000", "#ffffff", "metadata", "metadata"),
+ ("ui_fgcolor", "#000000", "#ffffff", "metadata", "metadata"),
+ ("task_display_name", "old", "new", "metadata", "metadata"),
+ ("on_failure_callback", None, _handle_failure, "callback", "execution"),
+ ("outlets", [Asset("s3://bucket/old")], [Asset("s3://bucket/new")], "asset", "execution"),
+ ],
+)
+def test_build_diff_classifies_serialized_task_fields(
+ is_mapped, include_values, field, before, after, category, impact
+):
+ payloads = []
+ for value in (before, after):
+ kwargs = {} if field in {"ui_color", "ui_fgcolor"} else {field: value}
+ with DAG("classification") as dag:
+ if is_mapped:
+ task = BashOperator.partial(task_id="task", **kwargs).expand(bash_command=["echo hello"])
+ else:
+ task = BashOperator(task_id="task", bash_command="echo hello", **kwargs)
+ if field in {"ui_color", "ui_fgcolor"}:
+ setattr(task, field, value)
+ payloads.append(DagSerialization.to_dict(dag))
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ changes = [change for change in result["changes"] if change["path"].startswith("/dag/tasks/")]
+ assert len(changes) == 1
+ assert changes[0]["category"] == category
+ assert changes[0]["impact"] == impact
+ if include_values:
+ assert "before_value" in changes[0] or "after_value" in changes[0]
+ else:
+ assert "before_value" not in changes[0]
+ assert "after_value" not in changes[0]
+
+
+def _build_operator_flag_payload(shape: str) -> dict[str, Any]:
+ with DAG("operator_flags", schedule=None) as dag:
+ if shape == "branch":
+ BranchPythonOperator(task_id="secret_task", python_callable=_return_private_before)
+ elif shape == "empty":
+ EmptyOperator(task_id="secret_task")
+ elif shape == "stub":
+ sdk_task.stub(task_id="secret_task")(_declare_private_argument)("private_value")
+ else:
+ PythonOperator(task_id="secret_task", python_callable=_return_private_before)
+ return _serialize_dag(dag)
+
+
+class _TemplatedPartialOperator(BaseOperator):
+ """An unmapped operator whose own execution input happens to be called partial_kwargs."""
+
+ template_fields = ("partial_kwargs",)
+
+ def __init__(self, *, partial_kwargs, **kwargs):
+ super().__init__(**kwargs)
+ self.partial_kwargs = partial_kwargs
+
+ def execute(self, context):
+ return self.partial_kwargs["owner"]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_keeps_partial_kwargs_opaque_across_a_mapping_change(include_values):
+ """One mapped side is not enough: the unmapped side's keys are still its own input."""
+ payloads = []
+ for is_mapped, owner in ((False, "first"), (True, "second")):
+ task: dict[str, Any] = {"task_id": "extract", "partial_kwargs": {"owner": owner}}
+ if is_mapped:
+ task["_is_mapped"] = True
+ payloads.append(_build_payload(tasks=[task]))
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ task_id = "extract" if include_values else "*"
+ assert not [change for change in result["changes"] if "partial_kwargs" in change["path"]], (
+ "an unmapped side's keys must not be named"
+ )
+ assert (f"/dag/tasks/{task_id}/custom_fields", "task", "execution") in [
+ (change["path"], change["category"], change["impact"]) for change in result["changes"]
+ ]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_keeps_an_unmapped_templated_partial_kwargs_opaque(include_values):
+ """
+ partial_kwargs is only mapped configuration when the task is actually mapped.
+
+ Reading an unmapped operator's templated dictionary as operator fields classified its
+ execution input by whichever field name a key collided with, and named a key its author
+ chose rather than leaving it inside the withheld value.
+ """
+ payloads = []
+ for owner in ("first", "second"):
+ with DAG("templated_partial", schedule=None) as dag:
+ _TemplatedPartialOperator(task_id="extract", partial_kwargs={"owner": owner})
+ payloads.append(_serialize_dag(dag))
+ assert _get_task(payloads[0])["partial_kwargs"] == {"owner": "first"}
+ assert not _get_task(payloads[0]).get("_is_mapped", False), "the task is not mapped"
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ task_id = "extract" if include_values else "*"
+ assert [(change["path"], change["category"], change["impact"]) for change in result["changes"]] == [
+ (f"/dag/tasks/{task_id}/custom_fields", "task", "execution")
+ ]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize(
+ ("field", "expected_category", "expected_impact"),
+ [("pool", "task", "execution"), ("owner", "metadata", "metadata")],
+)
+def test_build_diff_classifies_a_mapped_task_partial_kwarg_per_field(
+ field, expected_category, expected_impact, include_values
+):
+ payloads = []
+ for value in ("before", "after"):
+ with DAG("mapped_partial", schedule=None) as dag:
+ BashOperator.partial(task_id="extract", **{field: value}).expand(bash_command=["a", "b"])
+ payloads.append(_serialize_dag(dag))
+ assert _get_task(payloads[0])["_is_mapped"] is True
+
+ result = build_serialized_dag_diff(
+ base_data=payloads[0], target_data=payloads[1], include_values=include_values
+ )
+
+ task_id = "extract" if include_values else "*"
+ assert [(change["path"], change["category"], change["impact"]) for change in result["changes"]] == [
+ (f"/dag/tasks/{task_id}/partial_kwargs/{field}", expected_category, expected_impact)
+ ]
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize(
+ ("field", "shape"),
+ [("_can_skip_downstream", "branch"), ("_is_empty", "empty"), ("is_stub", "stub")],
+)
+def test_build_diff_names_operator_flag_fields(field, shape, include_values):
+ base = _build_operator_flag_payload(shape)
+ target = _build_operator_flag_payload("python")
+ assert field in _get_task(base)
+ assert field not in _get_task(target)
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target, include_values=include_values)
+
+ field_path = f"/dag/tasks/{'secret_task' if include_values else '*'}/{field}"
+ assert field_path in [change["path"] for change in result["changes"]]
+ change = next(change for change in result["changes"] if change["path"] == field_path)
+ assert (change["operation"], change["category"], change["impact"]) == (
+ "removed",
+ "task",
+ "execution",
+ )
+ if not include_values:
+ assert "secret_" not in json.dumps(result)
+
+
+@pytest.mark.parametrize("field", ["_arg_bindings", "secret_callback_argument"])
+@pytest.mark.parametrize("is_mapped", [False, True])
+def test_build_diff_redacts_argument_bindings_and_partial_arguments(field, is_mapped):
+ tasks = []
+ for value in ("old-secret", "new-secret"):
+ fields = {field: [{"argument": value}]}
+ task = {"task_id": "secret-task", **fields}
+ if is_mapped:
+ task = {"task_id": "secret-task", "_is_mapped": True, "partial_kwargs": fields}
+ tasks.append(task)
+
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=[tasks[0]]), target_data=_build_payload(tasks=[tasks[1]])
+ )
+
+ path = "/dag/tasks/*/partial_kwargs/custom_fields" if is_mapped else "/dag/tasks/*/custom_fields"
+ assert result["changes"] == [
+ {
+ "path": path,
+ "operation": "changed",
+ "category": "task",
+ "impact": "execution",
+ "occurrence_count": 1,
+ }
+ ]
+ assert all(
+ value not in json.dumps(result) for value in (field, "secret-task", "old-secret", "new-secret")
+ )
+
+
+@pytest.mark.parametrize(("field", "value"), [("tags", "tag"), ("allowed_run_types", "manual")])
+def test_build_diff_normalizes_duplicate_set_entries(field, value):
+ with DAG("duplicate_sets") as dag:
+ BashOperator(task_id="task", bash_command="echo hello")
+ base = DagSerialization.to_dict(dag)
+ base["dag"][field] = [value, value]
+ target = copy.deepcopy(base)
+ target["dag"][field] = [value]
+ DagSerialization.validate_schema(base)
+ assert getattr(DagSerialization.from_dict(copy.deepcopy(base)), field) == getattr(
+ DagSerialization.from_dict(copy.deepcopy(target)), field
+ )
+
+ result = build_serialized_dag_diff(base_data=base, target_data=target)
+
+ assert result["mode"] == "observed_state"
+ assert result["changes"] == []
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+@pytest.mark.parametrize("single_task", [False, True])
+@mock.patch.object(dag_version_diff, "_is_json_equal", wraps=dag_version_diff._is_json_equal, spec=True)
+def test_build_diff_stops_comparing_after_truncation(compare, include_values, single_task):
+ if single_task:
+ base_tasks = [{"task_id": "task", "doc_md": "old", "owner": "old", "queue": "old"}]
+ target_tasks = [{"task_id": "task", "doc_md": "first", "owner": "second", "queue": "unvisited"}]
+ else:
+ base_tasks = [{"task_id": f"task{index}", "doc_md": "old"} for index in range(3)]
+ target_tasks = [
+ {"task_id": f"task{index}", "doc_md": value}
+ for index, value in enumerate(("first", "second", "unvisited"))
+ ]
+
+ result = build_serialized_dag_diff(
+ base_data=_build_payload(tasks=base_tasks),
+ target_data=_build_payload(tasks=target_tasks),
+ max_changes=1,
+ include_values=include_values,
+ )
+
+ assert len(result["changes"]) == 1
+ compare.assert_any_call("old", "second")
+ # A redacted diff repeating one already disclosed path drops nothing, so it keeps walking
+ # to count the remaining occurrences instead of reporting truncation.
+ repeats_one_public_path = not single_task and not include_values
+ assert result["truncated"] is not repeats_one_public_path
+ assert result["changes"][0]["occurrence_count"] == (3 if repeats_one_public_path else 1)
+ assert (mock.call("old", "unvisited") in compare.call_args_list) is repeats_one_public_path
+
+
+@pytest.mark.parametrize("include_values", [False, True])
+def test_build_diff_reports_a_real_deep_payload_as_canonicalization_failure(include_values):
+ def payload(value):
+ nested: Any = {"leaf": value}
+ # Use the live limit: BaseOperator.__deepcopy__ can raise it process-wide.
+ for _ in range(sys.getrecursionlimit()):
+ nested = {"__type": "dict", "__var": {"nested": nested}}
+ return _build_payload(tasks=[{"task_id": "extract", "executor_config": nested}])
+
+ result = build_serialized_dag_diff(
+ base_data=payload(1), target_data=payload(2), include_values=include_values
+ )
+
+ assert result["mode"] == "unavailable"
+ assert result["unavailable_reason"] == "serialized_dag_canonicalization_failed"
+
+
+def test_build_diff_canonicalization_failure_logs_why(caplog):
+ payload = _build_payload(tasks=[])
+ payload["client_defaults"] = {"dags": {"catchup": True}}
+
+ result = build_serialized_dag_diff(base_data=payload, target_data=_build_payload(tasks=[]))
+
+ assert result["unavailable_reason"] == "serialized_dag_canonicalization_failed"
+ assert {
+ "event": "Serialized Dag diff canonicalization failed",
+ "error_type": "ValueError",
+ "reason": "unsupported client_defaults sections: ['dags']",
+ } in caplog
+
+
+@pytest.mark.parametrize(
+ ("max_changes", "expected_counts", "truncated"),
+ [
+ # Counting occurrences would drop timezone at the default bound.
+ (2, {"/dag/tasks/*/retries": 600, "/dag/timezone": 1}, False),
+ (500, {"/dag/tasks/*/retries": 600, "/dag/timezone": 1}, False),
+ (1, {"/dag/tasks/*/retries": 600}, True),
+ ],
+)
+def test_build_diff_bound_caps_records_not_repeats(max_changes, expected_counts, truncated):
+ def payload(retries, timezone):
+ built = _build_payload(tasks=[{"task_id": f"t{index}", "retries": retries} for index in range(600)])
+ built["dag"]["timezone"] = timezone
+ return built
+
+ result = build_serialized_dag_diff(
+ base_data=payload(1, "UTC"), target_data=payload(2, "Europe/Berlin"), max_changes=max_changes
+ )
+
+ counts = {change["path"]: change["occurrence_count"] for change in result["changes"]}
+ assert counts == expected_counts
+ assert result["truncated"] is truncated