This is an automated email from the ASF dual-hosted git repository.
kaxil 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 254758d5093 Fix dag.test() to sync sibling DAGs across calls (#64884)
(#66205)
254758d5093 is described below
commit 254758d50930c171bcced3932dadd192dc122745
Author: Kaxil Naik <[email protected]>
AuthorDate: Sat May 2 11:39:17 2026 +0100
Fix dag.test() to sync sibling DAGs across calls (#64884) (#66205)
`DAG.test()` previously skipped its bundle-sync loop when the parent DAG
was already serialized in the metadata DB. Sibling DAGs that the parent
triggers via `TriggerDagRunOperator` were therefore never written, and
the in-process Execution API returned 404 for them.
The fix:
- Look up `DagModel.bundle_name` for `self.dag_id`. If found and the
bundle is still configured, re-walk only that one bundle so sibling
DAGs reach the DB without re-parsing every other configured bundle on
every call.
- If the recorded bundle is no longer configured (renamed / removed),
fall back to walking all configured bundles.
- Drop `_airflow_parsing_context_manager(dag_id=self.dag_id)` from the
re-walk: setting `_AIRFLOW_PARSING_CONTEXT_DAG_ID` to the parent's id
would cause dynamic-DAG files using `get_parsing_context().dag_id` to
early-return and omit the trigger target from the bag, defeating the
re-sync.
Tests cover:
- The original repro (parent already synced, target absent).
- Dynamic-DAG variant using `get_parsing_context()` early-return.
- Stale recorded bundle name no longer in current config.
- Only the owning bundle is parsed when the parent is already in DB.
closes: #64884
closes: https://github.com/apache/airflow/pull/64945
---
.../dags/test_dag_test_with_dynamic_trigger.py | 55 +++++++++
.../tests/unit/dags/test_dag_test_with_trigger.py | 51 ++++++++
airflow-core/tests/unit/models/test_dag.py | 132 +++++++++++++++++++++
task-sdk/src/airflow/sdk/definitions/dag.py | 70 +++++++----
4 files changed, 283 insertions(+), 25 deletions(-)
diff --git a/airflow-core/tests/unit/dags/test_dag_test_with_dynamic_trigger.py
b/airflow-core/tests/unit/dags/test_dag_test_with_dynamic_trigger.py
new file mode 100644
index 00000000000..79f074a4ed9
--- /dev/null
+++ b/airflow-core/tests/unit/dags/test_dag_test_with_dynamic_trigger.py
@@ -0,0 +1,55 @@
+# 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.
+"""Dynamic-DAG file using ``get_parsing_context()`` to early-return.
+
+Used by the regression test for the case where ``DAG.test()`` must not set
+``_AIRFLOW_PARSING_CONTEXT_DAG_ID`` while re-walking the owning bundle, or it
+would strip the trigger target out of the bag.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.providers.standard.operators.trigger_dagrun import
TriggerDagRunOperator
+from airflow.sdk import DAG, get_parsing_context
+
+DEFAULT_DATE = datetime(2024, 1, 1)
+
+PARENT_ID = "test_dag_test_dynamic_trigger_parent"
+TARGET_ID = "test_dag_test_dynamic_trigger_target"
+
+current = get_parsing_context().dag_id
+
+if current is None or current == PARENT_ID:
+ with DAG(
+ dag_id=PARENT_ID,
+ schedule=None,
+ start_date=DEFAULT_DATE,
+ is_paused_upon_creation=False,
+ ):
+ TriggerDagRunOperator(task_id="trigger_target",
trigger_dag_id=TARGET_ID)
+
+if current is None or current == TARGET_ID:
+ with DAG(
+ dag_id=TARGET_ID,
+ schedule=None,
+ start_date=DEFAULT_DATE,
+ is_paused_upon_creation=False,
+ ):
+ EmptyOperator(task_id="target_task")
diff --git a/airflow-core/tests/unit/dags/test_dag_test_with_trigger.py
b/airflow-core/tests/unit/dags/test_dag_test_with_trigger.py
new file mode 100644
index 00000000000..dc50324bf15
--- /dev/null
+++ b/airflow-core/tests/unit/dags/test_dag_test_with_trigger.py
@@ -0,0 +1,51 @@
+# 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.
+"""DAGs used by ``test_dag_test_triggers_sibling_dag``.
+
+A parent DAG that triggers a sibling DAG defined in the same file. Used to
+exercise the path where ``DAG.test()`` must ensure the trigger target is
+present in the metadata DB even when the parent was serialized previously.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.providers.standard.operators.trigger_dagrun import
TriggerDagRunOperator
+from airflow.sdk import DAG
+
+DEFAULT_DATE = datetime(2024, 1, 1)
+
+with DAG(
+ dag_id="test_dag_test_trigger_parent",
+ schedule=None,
+ start_date=DEFAULT_DATE,
+ is_paused_upon_creation=False,
+):
+ TriggerDagRunOperator(
+ task_id="trigger_target",
+ trigger_dag_id="test_dag_test_trigger_target",
+ )
+
+with DAG(
+ dag_id="test_dag_test_trigger_target",
+ schedule=None,
+ start_date=DEFAULT_DATE,
+ is_paused_upon_creation=False,
+):
+ EmptyOperator(task_id="target_task")
diff --git a/airflow-core/tests/unit/models/test_dag.py
b/airflow-core/tests/unit/models/test_dag.py
index 4f437b44f51..b34ab12dd4a 100644
--- a/airflow-core/tests/unit/models/test_dag.py
+++ b/airflow-core/tests/unit/models/test_dag.py
@@ -222,6 +222,138 @@ class TestDag:
assert ser is not None
assert session.scalar(select(DagRun).where(DagRun.dag_id == dag_id))
is not None
+ @conf_vars({("core", "load_examples"): "false"})
+ def test_dag_test_syncs_sibling_for_trigger_dagrun(self, test_dags_bundle,
session):
+ """
+ Regression test for #64884.
+
+ ``DAG.test()`` must sync sibling DAGs in the bundle even when the
parent
+ DAG is already serialized in the metadata DB. Otherwise a parent that
+ uses ``TriggerDagRunOperator`` to fire a sibling fails with a 404 from
+ the in-process Execution API because the sibling was never written.
+ """
+ parent_id = "test_dag_test_trigger_parent"
+ target_id = "test_dag_test_trigger_target"
+
+ dagbag = DagBag(dag_folder=os.fspath(TEST_DAGS_FOLDER),
include_examples=False)
+ parent = dagbag.dags.get(parent_id)
+ assert parent is not None
+
+ # Pre-sync ONLY the parent: simulates the state where the parent was
+ # written previously (e.g. by an earlier dag.test() invocation or by
+ # the DAG processor) but the trigger target has never been synced.
+ sync_dag_to_db(parent, bundle_name="testing")
+ session.commit()
+
+ assert DBDagBag().get_latest_version_of_dag(parent_id,
session=session) is not None
+ assert DBDagBag().get_latest_version_of_dag(target_id,
session=session) is None
+
+ dr = parent.test()
+
+ assert dr.state == DagRunState.SUCCESS
+ assert DBDagBag().get_latest_version_of_dag(target_id,
session=session) is not None
+ assert session.scalar(select(DagRun).where(DagRun.dag_id ==
target_id)) is not None
+
+ @conf_vars({("core", "load_examples"): "false"})
+ def test_dag_test_syncs_sibling_for_dynamic_trigger_dagrun(self,
test_dags_bundle, session):
+ """
+ Regression test for the dynamic-DAG variant of #64884.
+
+ The bundle re-walk in ``DAG.test()`` must not run under
+ ``_airflow_parsing_context_manager(dag_id=self.dag_id)``: DAG files
that
+ use ``get_parsing_context().dag_id`` to early-return would otherwise
+ emit only the parent and the trigger target would be omitted from the
+ bag, leaving it missing from the metadata DB.
+ """
+ parent_id = "test_dag_test_dynamic_trigger_parent"
+ target_id = "test_dag_test_dynamic_trigger_target"
+
+ dagbag = DagBag(dag_folder=os.fspath(TEST_DAGS_FOLDER),
include_examples=False)
+ parent = dagbag.dags.get(parent_id)
+ assert parent is not None
+
+ sync_dag_to_db(parent, bundle_name="testing")
+ session.commit()
+
+ assert DBDagBag().get_latest_version_of_dag(target_id,
session=session) is None
+
+ dr = parent.test()
+
+ assert dr.state == DagRunState.SUCCESS
+ assert DBDagBag().get_latest_version_of_dag(target_id,
session=session) is not None
+ assert session.scalar(select(DagRun).where(DagRun.dag_id ==
target_id)) is not None
+
+ @conf_vars({("core", "load_examples"): "false"})
+ def test_dag_test_falls_back_when_recorded_bundle_no_longer_configured(
+ self, configure_dag_bundles, session
+ ):
+ """
+ If a parent DAG's recorded ``DagModel.bundle_name`` no longer appears
in
+ the current bundle config (renamed/removed), ``DAG.test()`` must walk
+ all configured bundles instead of silently skipping the sync.
+ """
+ parent_id = "test_dag_test_trigger_parent"
+ target_id = "test_dag_test_trigger_target"
+
+ dagbag = DagBag(dag_folder=os.fspath(TEST_DAGS_FOLDER),
include_examples=False)
+ parent = dagbag.dags.get(parent_id)
+ assert parent is not None
+
+ # Record the parent under a bundle name that we will NOT include in the
+ # configured bundle list below.
+ sync_dag_to_db(parent, bundle_name="ghost-bundle")
+ session.merge(DagBundleModel(name="ghost-bundle"))
+ session.commit()
+
+ with configure_dag_bundles({"current": TEST_DAGS_FOLDER}):
+ dr = parent.test()
+
+ assert dr.state == DagRunState.SUCCESS
+ assert DBDagBag().get_latest_version_of_dag(target_id,
session=session) is not None
+ assert session.scalar(select(DagRun).where(DagRun.dag_id ==
target_id)) is not None
+
+ @conf_vars({("core", "load_examples"): "false"})
+ def test_dag_test_only_syncs_owning_bundle_when_parent_already_serialized(
+ self, configure_dag_bundles, session
+ ):
+ """
+ When the parent DAG is already in the DB, ``DAG.test()`` should only
+ re-walk the bundle that owns it, not every configured bundle. Confirms
+ that the perf-sensitive optimization in the fix is in effect.
+ """
+ parent_id = "test_dag_test_trigger_parent"
+
+ dagbag = DagBag(dag_folder=os.fspath(TEST_DAGS_FOLDER),
include_examples=False)
+ parent = dagbag.dags.get(parent_id)
+ assert parent is not None
+
+ sync_dag_to_db(parent, bundle_name="testing")
+ # The owning bundle row needs to exist before dag.test() commits its
+ # bundle-sync; ``configure_dag_bundles`` only writes the unrelated
bundle.
+ session.merge(DagBundleModel(name="testing"))
+ session.commit()
+
+ with configure_dag_bundles(
+ {
+ "testing": TEST_DAGS_FOLDER,
+ "unrelated": TEST_DAGS_FOLDER,
+ }
+ ):
+ instantiated: list[str] = []
+ real_init = BundleDagBag.__init__
+
+ def _spy(self, *args, **kwargs):
+ instantiated.append(kwargs.get("bundle_name", ""))
+ real_init(self, *args, **kwargs)
+
+ with mock.patch.object(BundleDagBag, "__init__", _spy):
+ dr = parent.test()
+
+ assert dr.state == DagRunState.SUCCESS
+ # Only the owning bundle should have been parsed.
+ assert "testing" in instantiated
+ assert "unrelated" not in instantiated
+
def teardown_method(self) -> None:
clear_db_runs()
clear_db_dags()
diff --git a/task-sdk/src/airflow/sdk/definitions/dag.py
b/task-sdk/src/airflow/sdk/definitions/dag.py
index ab85396335e..18200bd6714 100644
--- a/task-sdk/src/airflow/sdk/definitions/dag.py
+++ b/task-sdk/src/airflow/sdk/definitions/dag.py
@@ -1289,34 +1289,54 @@ class DAG:
else:
timetable = coerce_to_core_timetable(self.timetable)
data_interval =
timetable.infer_manual_data_interval(run_after=logical_date)
+ # These imports are intentionally lazy: this Task SDK module must
not
+ # pull in airflow-core at import time (worker isolation).
+ from airflow.dag_processing.bundles.manager import
DagBundlesManager
+ from airflow.dag_processing.dagbag import BundleDagBag,
sync_bag_to_db
+ from airflow.models.dag import DagModel
from airflow.models.dag_version import DagVersion
- version = DagVersion.get_version(self.dag_id)
- if not version:
- from airflow.dag_processing.bundles.manager import
DagBundlesManager
- from airflow.dag_processing.dagbag import BundleDagBag,
sync_bag_to_db
- from airflow.sdk.definitions._internal.dag_parsing_context
import (
- _airflow_parsing_context_manager,
+ manager = DagBundlesManager()
+ manager.sync_bundles_to_db(session=session)
+ session.commit()
+
+ # Re-sync the bundle that owns ``self`` so sibling DAGs (e.g.
targets of
+ # TriggerDagRunOperator) are written to the metadata DB on every
call,
+ # not just the first one (apache/airflow#64884). When we can
identify
+ # ``self``'s bundle from a prior sync we walk only that bundle;
otherwise
+ # we walk every configured bundle until we find it.
``sync_bag_to_db``
+ # is idempotent at the per-DAG hash level.
+ #
+ # Note: we deliberately do NOT use
``_airflow_parsing_context_manager``
+ # here. Setting ``_AIRFLOW_PARSING_CONTEXT_DAG_ID`` to
``self.dag_id``
+ # would cause DAG files that follow the documented dynamic-DAG
+ # optimization (early-return when the parsing context dag_id
doesn't
+ # match) to omit sibling DAGs from the bag, defeating the whole
point
+ # of this re-sync.
+ all_bundles = list(manager.get_all_dag_bundles())
+ existing_dm = DagModel.get_current(self.dag_id, session=session)
+ existing_bundle_name = existing_dm.bundle_name if existing_dm else
None
+ owning = (
+ [b for b in all_bundles if b.name == existing_bundle_name]
+ if existing_bundle_name is not None
+ else []
+ )
+ # If the recorded bundle is no longer in the configured bundle list
+ # (renamed / removed), fall back to walking all bundles so the sync
+ # doesn't silently skip.
+ bundles_to_sync = owning if owning else all_bundles
+
+ for bundle in bundles_to_sync:
+ if not bundle.is_initialized:
+ bundle.initialize()
+ dagbag = BundleDagBag(
+ dag_folder=bundle.path,
+ bundle_path=bundle.path,
+ bundle_name=bundle.name,
)
-
- manager = DagBundlesManager()
- manager.sync_bundles_to_db(session=session)
- session.commit()
- # sync all bundles? or use the dags-folder bundle?
- # What if the test dag is in a different bundle?
- for bundle in manager.get_all_dag_bundles():
- if not bundle.is_initialized:
- bundle.initialize()
- with _airflow_parsing_context_manager(dag_id=self.dag_id):
- dagbag = BundleDagBag(
- dag_folder=bundle.path,
- bundle_path=bundle.path,
- bundle_name=bundle.name,
- )
- sync_bag_to_db(dagbag, bundle.name, bundle.version)
- version = DagVersion.get_version(self.dag_id)
- if version:
- break
+ sync_bag_to_db(dagbag, bundle.name, bundle.version)
+ if DagVersion.get_version(self.dag_id):
+ break
# Preserve callback functions from original Dag since they're lost
during serialization
# and yes it is a hack for now! It is a tradeoff for code
simplicity.
