diff --git a/airflow-core/src/airflow/exceptions.py b/airflow-core/src/airflow/exceptions.py index cb45b050fce34..fdc0408137ee0 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 8aa7efcab01e2..59734eb7bcb83 100644 --- a/airflow-core/src/airflow/models/dag_version.py +++ b/airflow-core/src/airflow/models/dag_version.py @@ -17,9 +17,10 @@ from __future__ import annotations +import hashlib import logging 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 +30,7 @@ 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 @@ -40,6 +42,11 @@ log = logging.getLogger(__name__) +SourceStatus = Literal["current_stored_code", "redacted", "unavailable"] +ValuesStatus = Literal["available", "unavailable"] +_VALID_SOURCE_STATUSES = get_args(SourceStatus) +_VALID_VALUES_STATUSES = get_args(ValuesStatus) + class DagVersion(Base): """Model to track the versions of DAGs in the database.""" @@ -243,6 +250,135 @@ def version(self) -> str: """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, + *, + include_values: bool = False, + include_source: bool = False, + max_changes: int | None = None, + source_status: SourceStatus | None = None, + values_status: ValuesStatus | None = None, + session: Session = NEW_SESSION, + ) -> dict[str, Any]: + """ + Compare two versions of a Dag using their currently stored state. + + ``source_status`` is supplied by callers that have an authorization context. The CLI has + operator-level authority and leaves it unset, while API callers calculate it before this + method returns any source content. API callers use ``values_status`` to suppress raw values + when the user cannot access Dag code; the structural diff remains available in redacted form. + """ + # Keep this local to avoid the dag_version -> dag_version_diff -> serialized_objects cycle. + from airflow.serialization.dag_version_diff import ( + DEFAULT_MAX_CHANGES, + build_serialized_dag_diff, + ) + + if source_status is not None and source_status not in _VALID_SOURCE_STATUSES: + raise ValueError(f"source_status must be one of {_VALID_SOURCE_STATUSES}, not {source_status!r}") + if values_status is not None and 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 + if base_version_number < 1 or target_version_number < 1: + raise ValueError("Dag version numbers must be positive integers") + + if source_status is None: + source_status = "current_stored_code" if include_source else "unavailable" + + query = ( + select(cls) + .where( + cls.dag_id == dag_id, + cls.version_number.in_((base_version_number, target_version_number)), + ) + .options(joinedload(cls.serialized_dag)) + ) + if include_source and source_status == "current_stored_code": + query = query.options(joinedload(cls.dag_code)) + + 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] + effective_include_values = include_values and values_status in {None, "available"} + result = build_serialized_dag_diff( + base_data=base_version.serialized_dag.data if base_version.serialized_dag else None, + target_data=target_version.serialized_dag.data if target_version.serialized_dag else None, + base_provenance=_get_provenance(base_version), + target_provenance=_get_provenance(target_version), + include_values=effective_include_values, + max_changes=max_changes, + ) + if include_values: + values_available = effective_include_values and result["mode"] == "observed_state" + result["values"] = {"status": "available" if values_available else "unavailable"} + result["source"] = _get_source_diff( + base_version, + target_version, + include_source=include_source, + source_status=source_status, + ) + return result + + +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 _get_source_diff( + base_version: DagVersion, + target_version: DagVersion, + *, + include_source: bool, + source_status: SourceStatus, +) -> dict[str, Any]: + if not include_source: + return {"status": "unavailable", "fidelity": "unavailable"} + if source_status != "current_stored_code": + return {"status": source_status, "fidelity": source_status} + + base_code = base_version.dag_code + target_code = target_version.dag_code + if base_code is None or target_code is None: + return {"status": "unavailable", "fidelity": "unavailable"} + + base_source = base_code.source_code + target_source = target_code.source_code + return { + "status": "current_stored_code", + "fidelity": "current_stored_code", + "changed": base_source != target_source, + "base": {"digest": _get_source_digest(base_source), "content": base_source}, + "target": {"digest": _get_source_digest(target_source), "content": target_source}, + } + + +def _get_source_digest(source: str) -> str: + return f"sha256:{hashlib.sha256(source.encode()).hexdigest()}" + 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 0000000000000..e011644a89b5d --- /dev/null +++ b/airflow-core/src/airflow/serialization/dag_version_diff.py @@ -0,0 +1,527 @@ +# 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.""" + +from __future__ import annotations + +import copy +import hashlib +import json +from collections.abc import Callable, Mapping +from enum import Enum +from typing import Any, Literal + +from airflow.serialization.serialized_objects import DagSerialization + +DIFF_SCHEMA_VERSION = 1 +DEFAULT_MAX_CHANGES = 500 +MAX_ALLOWED_CHANGES = 5000 +SUPPORTED_SERIALIZED_DAG_SCHEMA_VERSIONS = frozenset((1, 2, 3)) + +DiffMode = Literal["observed_state", "unavailable"] + +_ORDER_INSENSITIVE_LIST_PATHS = { + ("dag", "tags"), + ("dag", "allowed_run_types"), +} +_KEYED_COLLECTION_PATHS = { + ("dag", "tasks"), + ("dag", "dag_dependencies"), + *_ORDER_INSENSITIVE_LIST_PATHS, +} +_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", + "_disallow_kwargs_override", + "_expand_input_attr", + "_is_mapped", + "_is_sensor", + "_logger_name", + "_needs_expansion", + "_operator_extra_links", + "_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_on_failure", + "email_on_retry", + "end_date", + "execution_timeout", + "executor", + "executor_config", + "has_on_execute_callback", + "has_on_failure_callback", + "has_on_retry_callback", + "has_on_skipped_callback", + "has_on_success_callback", + "ignore_first_depends_on_past", + "inlets", + "is_setup", + "is_teardown", + "map_index_template", + "max_active_tis_per_dag", + "max_active_tis_per_dagrun", + "max_retry_delay", + "multiple_outputs", + "on_failure_fail_dagrun", + "outlets", + "owner", + "params", + "partial_kwargs", + "pool", + "pool_slots", + "priority_weight", + "queue", + "render_template_as_native_obj", + "retries", + "retry_delay", + "retry_exponential_backoff", + "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[str] = frozenset() +_REDACTED_RECURSIVE_MAPPING_PATHS = { + (), + ("dag",), + ("dag", "task_group"), + ("provenance",), + *_KEYED_COLLECTION_PATHS, +} + + +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]: + """ + Build a bounded, deterministic diff from two stored serialized Dag payloads. + + Raw values, digests, and value-derived path components are returned only when + ``include_values`` is true. + """ + _validate_max_changes(max_changes) + + base_schema_version = _get_schema_version(base_data) + target_schema_version = _get_schema_version(target_data) + result: dict[str, Any] = { + "diff_schema_version": DIFF_SCHEMA_VERSION, + "serialized_dag_schema_versions": { + "base": base_schema_version, + "target": target_schema_version, + }, + "mode": "observed_state", + "changes": [], + "truncated": False, + } + + if base_data is None or target_data is None: + return _unavailable(result, "serialized_dag_missing") + + if base_schema_version is None or target_schema_version is None: + return _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 _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) + except (AttributeError, KeyError, OverflowError, TypeError, ValueError): + return _unavailable(result, "serialized_dag_canonicalization_failed") + + base_document["provenance"] = _canonicalize_value(dict(base_provenance or {}), path=("provenance",)) + target_document["provenance"] = _canonicalize_value(dict(target_provenance or {}), path=("provenance",)) + + collector = _ChangeCollector(max_changes=max_changes, include_values=include_values) + _collect_changes(base_document, target_document, path=(), collector=collector) + result["changes"] = collector.changes + result["truncated"] = collector.count > max_changes + return result + + +class _ChangeCollector: + def __init__(self, *, max_changes: int, include_values: bool) -> None: + self.changes: list[dict[str, Any]] = [] + self.count = 0 + self.max_changes = max_changes + self.include_values = include_values + + def add( + self, + *, + path: tuple[str, ...], + operation: Literal["added", "removed", "changed"], + before: Any, + after: Any, + ) -> None: + self.count += 1 + if len(self.changes) >= self.max_changes: + return + + public_path = _get_public_path(path) + 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), + } + 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 + self.changes.append(change) + + +_MISSING = object() + + +def _validate_max_changes(max_changes: int) -> None: + if 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 _unavailable(result: dict[str, Any], reason: str) -> dict[str, Any]: + result["mode"] = "unavailable" + result["unavailable_reason"] = reason + return result + + +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") + _apply_client_defaults(payload) + payload.pop("__version", None) + return _canonicalize_value(payload, path=()) + + +def _apply_client_defaults(payload: dict[str, Any]) -> None: + client_defaults = payload.pop("client_defaults", None) + if client_defaults is None: + return + if not isinstance(client_defaults, Mapping): + raise ValueError("client_defaults is not an object") + + task_defaults = client_defaults.get("tasks", {}) + if not isinstance(task_defaults, Mapping): + raise ValueError("client_defaults.tasks is not an object") + + tasks = payload["dag"].get("tasks", []) + if not isinstance(tasks, list): + raise ValueError("dag.tasks is not a list") + for task in tasks: + if not isinstance(task, dict) or not isinstance(task.get("__var"), Mapping): + raise ValueError("task entry is not an object") + task_data = dict(task_defaults) + task_data.update(task["__var"]) + if isinstance(partial_kwargs := task_data.get("partial_kwargs"), Mapping): + effective_partial_kwargs = dict(task_defaults) + effective_partial_kwargs.update(partial_kwargs) + task_data["partial_kwargs"] = effective_partial_kwargs + task["__var"] = task_data + + +def _canonicalize_value(value: Any, *, path: tuple[str, ...]) -> Any: + if isinstance(value, Mapping): + 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): + 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) + 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 == ("dag", "dag_dependencies") and keyed_values[key] == value: + continue + raise ValueError(f"duplicate key {key!r} in /{'/'.join(path)}") + canonical_value = value + if path == ("dag", "tasks") and isinstance(value, Mapping) and "__var" in value: + task_value = dict(value["__var"]) + if "__type" in value: + task_value["__type"] = value["__type"] + canonical_value = task_value + keyed_values[key] = canonical_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, Mapping): + raise ValueError("dependency entry is not an object") + components = ( + dependency.get("dependency_type"), + dependency.get("dependency_id"), + dependency.get("source"), + dependency.get("target"), + dependency.get("label"), + ) + return json.dumps(components, ensure_ascii=False, separators=(",", ":")) + + +def _collect_changes( + before: Any, + after: Any, + *, + path: tuple[str, ...], + collector: _ChangeCollector, +) -> None: + if isinstance(before, Mapping) and isinstance(after, Mapping): + if not collector.include_values and _is_task_mapping_path(path): + _collect_redacted_task_changes(before, after, path=path, collector=collector) + return + if not collector.include_values and not _should_recurse_redacted_mapping(path): + if not _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) + return + + if isinstance(before, list) and isinstance(after, list): + if not _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 _json_equal(before, after): + collector.add(path=path, operation="changed", before=before, after=after) + + +def _collect_redacted_task_changes( + before: Mapping[str, Any], + after: Mapping[str, Any], + *, + path: tuple[str, ...], + collector: _ChangeCollector, +) -> None: + public_fields = _DIFF_V1_PUBLIC_TASK_FIELDS + 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, + ) + + 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 _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 len(path) == 3 and path[:2] == ("dag", "tasks") + + +def _should_recurse_redacted_mapping(path: tuple[str, ...]) -> bool: + return path in _REDACTED_RECURSIVE_MAPPING_PATHS + + +def _get_public_path(path: tuple[str, ...]) -> tuple[str, ...]: + 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") and path[3] not in _DIFF_V1_PUBLIC_TASK_FIELDS: + return (*path[:3], _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) + + +def _canonical_json(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def _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 _canonical_json(before) == _canonical_json(after) + + +def _get_digest(value: Any) -> str: + return f"sha256:{hashlib.sha256(_canonical_json(value).encode()).hexdigest()}" + + +def _get_category(path: tuple[str, ...]) -> str: + lowered_path = tuple(component.lower() for component in path) + if lowered_path and lowered_path[0] == "provenance": + return "provenance" + if lowered_path[:2] == ("dag", "dag_dependencies"): + return "dependency" + if lowered_path[:2] == ("dag", "tags"): + return "metadata" + if lowered_path[:2] == ("dag", "allowed_run_types"): + return "schedule" + + if lowered_path[:2] == ("dag", "tasks"): + field = lowered_path[3] if len(lowered_path) >= 4 else None + default_category = "task" + elif lowered_path and lowered_path[0] == "dag": + field = lowered_path[1] if len(lowered_path) >= 2 else None + default_category = "unknown" + else: + return "unknown" + + if field == "deadline": + return "deadline" + if field is not None and "callback" in field: + return "callback" + if field in {"asset", "assets", "inlets", "outlets"}: + return "asset" + if field in {"param", "params", "default_args"}: + return "param" + if field in { + "dag_dependencies", + "dependencies", + "edge_info", + "upstream_task_ids", + "downstream_task_ids", + }: + return "dependency" + if field in { + "schedule", + "schedule_interval", + "timetable", + "catchup", + "max_active_runs", + "max_active_tasks", + "dagrun_timeout", + "fail_fast", + "fail_stop", + "allowed_run_types", + }: + return "schedule" + if field in {"tags", "description", "doc_md", "owner", "owners", "email", "fileloc"}: + return "metadata" + return default_category + + +def _get_impact(category: str) -> str: + if category == "provenance": + return "provenance" + if category in {"task", "dependency", "schedule", "param", "asset", "deadline", "callback"}: + return "execution" + if category == "metadata": + return "metadata" + return "unknown" diff --git a/airflow-core/tests/unit/models/test_dag_version.py b/airflow-core/tests/unit/models/test_dag_version.py index af78736826ee1..231c230a68687 100644 --- a/airflow-core/tests/unit/models/test_dag_version.py +++ b/airflow-core/tests/unit/models/test_dag_version.py @@ -17,12 +17,14 @@ from __future__ import annotations from datetime import timedelta +from typing import Any from unittest import mock import pytest -from sqlalchemy import func, select +from sqlalchemy import delete, func, select 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 @@ -209,3 +211,124 @@ def test_resolve_version_data(self, dag_version, bundle_version, expected): 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): + import datetime + + from tests_common.test_utils.db import clear_db_serialized_dags + + 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 + + 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 result["source"] == {"status": "unavailable", "fidelity": "unavailable"} + assert "values" not in result + + 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("values_status", [None, "available"]) + def test_marks_values_available_when_allowed(self, dag_id, session, values_status): + result = DagVersion.get_diff( + dag_id, 1, 2, include_values=True, values_status=values_status, session=session + ) + + assert result["values"] == {"status": "available"} + 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, include_values=True, 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("source_status", [None, "current_stored_code"]) + def test_includes_current_stored_source(self, dag_id, session, source_status): + result = DagVersion.get_diff( + dag_id, 1, 2, include_source=True, source_status=source_status, session=session + ) + + source = result["source"] + assert source["status"] == "current_stored_code" + assert isinstance(source["changed"], bool) + assert source["base"]["digest"].startswith("sha256:") + assert source["target"]["content"] is not None + + @pytest.mark.parametrize("source_status", ["redacted", "unavailable"]) + def test_hides_source_when_status_denied(self, dag_id, session, source_status): + result = DagVersion.get_diff( + dag_id, 1, 2, include_source=True, source_status=source_status, session=session + ) + + assert result["source"] == {"status": source_status, "fidelity": source_status} + + @pytest.mark.parametrize( + ("status_name", "invalid_status"), + [("source_status", ""), ("values_status", "redacted")], + ) + def test_rejects_invalid_authorization_status(self, dag_id, session, status_name, invalid_status): + status: dict[str, Any] = {status_name: invalid_status} + + with pytest.raises(ValueError, match=rf"{status_name} must be one of"): + DagVersion.get_diff( + dag_id, + 1, + 2, + include_source=True, + include_values=True, + session=session, + **status, + ) + + def test_marks_source_unavailable_when_code_missing(self, dag_id, session): + from airflow.models.dagcode import DagCode + + session.execute(delete(DagCode)) + session.commit() + session.expunge_all() + + result = DagVersion.get_diff(dag_id, 1, 2, include_source=True, session=session) + + assert result["source"] == {"status": "unavailable", "fidelity": "unavailable"} 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 0000000000000..896e5ec892668 --- /dev/null +++ b/airflow-core/tests/unit/serialization/test_dag_version_diff.py @@ -0,0 +1,672 @@ +# 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 json + +import pytest + +from airflow.serialization.dag_version_diff import ( + _DIFF_V1_PUBLIC_TASK_FIELDS, + _DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS, + build_serialized_dag_diff, +) +from airflow.serialization.json_schema import load_dag_schema_dict + + +def _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 test_diff_v1_task_field_policy_tracks_serializer_schema() -> None: + schema_fields = frozenset(load_dag_schema_dict()["definitions"]["operator"]["properties"]) + + assert schema_fields == ( + (_DIFF_V1_PUBLIC_TASK_FIELDS - {"__type"}) | _DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS + ) + assert not (_DIFF_V1_PUBLIC_TASK_FIELDS & _DIFF_V1_REDACTED_SCHEMA_TASK_FIELDS) + + +def test_build_diff_is_deterministic_and_normalizes_order() -> None: + base = _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 = _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 = _payload(tasks=[{"task_id": "extract", "retries": 1}], tags=["old"]) + target = _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_bounds_changes_and_reports_truncation() -> None: + result = build_serialized_dag_diff( + base_data=_payload(tasks=[{"task_id": "extract", "retries": 1}], tags=["old"]), + target_data=_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=_payload(tasks=[], dependencies=[dependency]), + target_data=_payload(tasks=[]), + ) + + assert len(result["changes"]) == 1 + assert result["changes"][0]["category"] == "dependency" + assert result["changes"][0]["impact"] == "execution" + + +def test_build_diff_classifies_fail_fast_as_schedule() -> None: + base = _payload(tasks=[]) + target = _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, _payload(tasks=[]), "serialized_dag_missing"), + ({"__version": 99, "dag": {}}, _payload(tasks=[]), "unsupported_serialized_dag_schema_version:99"), + ({"__version": 3, "dag": []}, _payload(tasks=[]), "serialized_dag_canonicalization_failed"), + ( + { + "__version": 1, + "dag": { + "dag_id": "example", + "tasks": [], + "task_group": {}, + "schedule_interval": 1, + }, + }, + _payload(tasks=[]), + "serialized_dag_canonicalization_failed", + ), + ( + { + "__version": 1, + "dag": { + "dag_id": "example", + "tasks": [], + "task_group": {}, + "schedule_interval": {"__type": "timedelta", "__var": 10**20}, + }, + }, + _payload(tasks=[]), + "serialized_dag_canonicalization_failed", + ), + ], +) +def test_build_diff_returns_unavailable_for_unsafe_inputs(base_data, target_data, reason) -> 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 + + +def test_build_diff_rejects_non_positive_change_bound() -> None: + with pytest.raises(ValueError, match="max_changes must be a positive integer"): + build_serialized_dag_diff(base_data=_payload(tasks=[]), target_data=_payload(tasks=[]), max_changes=0) + + +def test_build_diff_rejects_unbounded_change_bound() -> None: + with pytest.raises(ValueError, match="max_changes must not exceed 5000"): + build_serialized_dag_diff( + base_data=_payload(tasks=[]), target_data=_payload(tasks=[]), max_changes=5001 + ) + + +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=_payload( + tasks=[{"task_id": "secret-task", "retries": 1}], + tags=["secret-old-tag"], + dependencies=[dependency], + ), + target_data=_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=_payload(tasks=[base_task]), + target_data=_payload(tasks=[target_task]), + ) + + assert result["changes"] == [ + { + "path": "/dag/tasks/*/custom_fields", + "operation": "changed", + "category": "task", + "impact": "execution", + } + ] + 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=_payload(tasks=[base_task]), + target_data=_payload(tasks=[target_task]), + include_values=True, + ) + + change = next(change for change in result["changes"] if change["path"].endswith(custom_field)) + assert change["path"] == f"/dag/tasks/extract/{custom_field}" + assert change["category"] == "task" + assert change["impact"] == "execution" + assert change["before_value"] == "old-secret" + assert change["after_value"] == "new-secret" + + +@pytest.mark.parametrize( + ( + "base_data", + "target_data", + "hidden_identifier", + "expected_path", + "expected_operation", + "expected_category", + "expected_impact", + ), + [ + pytest.param( + _payload(tasks=[{"task_id": "deadline", "retries": 1}]), + _payload(tasks=[{"task_id": "deadline", "retries": 2}]), + "deadline", + "/dag/tasks/*/retries", + "changed", + "task", + "execution", + id="task-id", + ), + pytest.param( + _payload(tasks=[]), + _payload(tasks=[], tags=["deadline"]), + "deadline", + "/dag/tags/*", + "added", + "metadata", + "metadata", + id="tag", + ), + pytest.param( + _payload(tasks=[]), + _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, + } + ] + 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=_payload(tasks=[{"task_id": "extract", "outlets": []}]), + target_data=_payload(tasks=[{"task_id": "extract", "outlets": ["asset"]}]), + ) + + assert result["changes"] == [ + { + "path": "/dag/tasks/*/outlets", + "operation": "changed", + "category": "asset", + "impact": "execution", + } + ] + + +def test_build_diff_collapses_arbitrary_mappings_without_values() -> None: + base = _payload(tasks=[]) + target = _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": "param", + "impact": "execution", + } + ] + + +def test_build_diff_reports_unkeyed_lists_as_one_stable_change() -> None: + base = _payload(tasks=[]) + target = _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 = _payload(tasks=[]) + target = _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 = _payload(tasks=[]) + target = _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", "retries": 2, "partial_kwargs": {"retries": 2}}, + {"task_id": "extract", "partial_kwargs": {}}, + ), + ], +) +def test_build_diff_normalizes_v3_client_defaults(base_task, target_task) -> None: + base = _payload(tasks=[base_task]) + base["__version"] = 2 + target = _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( + "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": {}}}, + ], +) +def test_build_diff_rejects_malformed_client_defaults(base_data) -> None: + result = build_serialized_dag_diff(base_data=base_data, target_data=_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=_payload(tasks=[], dependencies=[dependency, dependency]), + target_data=_payload(tasks=[], dependencies=[dependency]), + ) + + 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", + "target": "b", + "label": "c|d", + } + + result = build_serialized_dag_diff( + base_data=_payload(tasks=[], dependencies=[first_dependency, second_dependency]), + target_data=_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 = _payload(tasks=[]) + target = _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" + + +def test_build_diff_classifies_callback_as_execution() -> None: + base = _payload(tasks=[]) + target = _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 _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 _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=_v1_payload(), target_data=_equivalent_v3_payload()) + + assert result["mode"] == "observed_state" + assert result["serialized_dag_schema_versions"] == {"base": 1, "target": 3} + assert result["changes"] == []