From 1fe4bb6d2d160271596bf6ede5cdfd10f55d5ad9 Mon Sep 17 00:00:00 2001 From: kjh0623 <8412070+kjh0623@users.noreply.github.com> Date: Wed, 15 Jul 2026 09:38:36 +0900 Subject: [PATCH 1/4] Fix import errors silently dropped by full_file_path() in the parsing transaction When recording a DAG import error, _update_import_errors() called ParseImportError.full_file_path() to build the argument for the on_new/on_existing_dag_import_error listeners. full_file_path() re-instantiates the DAG bundle via DagBundlesManager().get_bundle(); for some bundle types (e.g. the Git bundle) this resolves an Airflow Connection. Doing that inside the DAG-parsing transaction, before the pending ParseImportError row is flushed, could disturb the transaction and roll the row back -- so the DAG was flagged as broken (has_import_errors) while no import error was ever shown in the UI. Thread the already-known bundle path (DagFileInfo.bundle_path) down through persist_parsing_result -> update_dag_parsing_results_in_db -> _update_import_errors and build the listener filename with str(bundle_path / relative_fileloc), so the bundle is not re-instantiated inside the transaction. full_file_path() is kept only as a fallback for callers that do not thread the bundle path. This also drops the extra select(ParseImportError) that previously existed only to feed the listener argument on the existing-error path. Approach suggested by @dhkim1920 in the PR review. Signed-off-by: kjh0623 <8412070+kjh0623@users.noreply.github.com> --- airflow-core/newsfragments/69845.bugfix.rst | 1 + .../src/airflow/dag_processing/collection.py | 53 +++++++++++++++---- .../src/airflow/dag_processing/manager.py | 3 ++ .../unit/dag_processing/test_collection.py | 41 ++++++++++++++ 4 files changed, 87 insertions(+), 11 deletions(-) create mode 100644 airflow-core/newsfragments/69845.bugfix.rst diff --git a/airflow-core/newsfragments/69845.bugfix.rst b/airflow-core/newsfragments/69845.bugfix.rst new file mode 100644 index 0000000000000..4b5918bdc404a --- /dev/null +++ b/airflow-core/newsfragments/69845.bugfix.rst @@ -0,0 +1 @@ +Fix DAG import errors being silently dropped (never shown in the *Import Errors* view). While recording an import error, ``_update_import_errors`` called ``ParseImportError.full_file_path()`` to build the import-error listener argument. ``full_file_path()`` re-instantiates the DAG bundle and, for some bundle types (e.g. the Git bundle), resolves an Airflow Connection; doing so inside the DAG-parsing transaction could roll back the not-yet-flushed import error row, leaving the DAG marked as broken with nothing shown in the UI. The already-known bundle path is now threaded down to ``_update_import_errors`` and used to build the listener filename directly, so the bundle is no longer re-instantiated inside the transaction. diff --git a/airflow-core/src/airflow/dag_processing/collection.py b/airflow-core/src/airflow/dag_processing/collection.py index 8fa5209ff15fb..13a08f5fd30d1 100644 --- a/airflow-core/src/airflow/dag_processing/collection.py +++ b/airflow-core/src/airflow/dag_processing/collection.py @@ -71,6 +71,7 @@ if TYPE_CHECKING: from collections.abc import Collection, Iterable, Iterator + from pathlib import Path from sqlalchemy.orm import Session from sqlalchemy.sql import Select @@ -397,9 +398,36 @@ def _update_import_errors( bundle_name: str, import_errors: dict[tuple[str, str], str], session: Session, + *, + bundle_path: Path | None = None, ): from airflow.listeners.listener import get_listener_manager + def _listener_filename( + relative_fileloc: str, bundle_name_: str, import_error: ParseImportError | None = None + ) -> str | None: + """ + Build the file path handed to import-error listeners. + + When the bundle path is known (the DAG processor threads it down from the + already-initialised bundle) we build the path directly. This avoids calling + ``ParseImportError.full_file_path()``, which re-instantiates the DAG bundle -- + and, for some bundle types (e.g. the Git bundle), resolves a Connection -- + inside this parsing transaction, where it can roll back the pending import + error row. Falls back to ``full_file_path()`` only when the bundle path was + not provided. + """ + if bundle_path is not None: + return str(bundle_path / relative_fileloc) + if import_error is None: + import_error = session.scalar( + select(ParseImportError).where( + ParseImportError.bundle_name == bundle_name_, + ParseImportError.filename == relative_fileloc, + ) + ) + return import_error.full_file_path() if import_error is not None else None + # Check existing import errors BEFORE deleting, so we can determine if we should update or create existing_import_error_files = set( session.execute(select(ParseImportError.bundle_name, ParseImportError.filename)) @@ -436,15 +464,10 @@ def _update_import_errors( # sending notification when an existing dag import error occurs try: # todo: make listener accept bundle_name and relative_filename - import_error = session.scalar( - select(ParseImportError).where( - ParseImportError.bundle_name == bundle_name_, - ParseImportError.filename == relative_fileloc, - ) - ) - if import_error is not None: + filename = _listener_filename(relative_fileloc, bundle_name_) + if filename is not None: get_listener_manager().hook.on_existing_dag_import_error( - filename=import_error.full_file_path(), stacktrace=stacktrace + filename=filename, stacktrace=stacktrace ) except Exception: log.exception("error calling listener") @@ -458,9 +481,11 @@ def _update_import_errors( session.add(import_error) # sending notification when a new dag import error occurs try: - get_listener_manager().hook.on_new_dag_import_error( - filename=import_error.full_file_path(), stacktrace=stacktrace - ) + filename = _listener_filename(relative_fileloc, bundle_name, import_error) + if filename is not None: + get_listener_manager().hook.on_new_dag_import_error( + filename=filename, stacktrace=stacktrace + ) except Exception: log.exception("error calling listener") session.execute( @@ -494,6 +519,7 @@ def update_dag_parsing_results_in_db( DagWarningType.RUNTIME_VARYING_VALUE, ), files_parsed: set[tuple[str, str]] | None = None, + bundle_path: Path | None = None, ): """ Update everything to do with DAG parsing in the DB. @@ -515,6 +541,10 @@ def update_dag_parsing_results_in_db( :param files_parsed: Set of (bundle_name, relative_fileloc) tuples for all files that were parsed. If None, will be inferred from dags and import_errors. Passing this explicitly ensures that import errors are cleared for files that were parsed but no longer contain DAGs. + :param bundle_path: Filesystem path of the already-initialised DAG bundle. When provided, it is + used to build the file path passed to import-error listeners without re-instantiating the + bundle inside this transaction. If None, ``ParseImportError.full_file_path()`` is used as a + fallback. """ # Retry 'DAG.bulk_write_to_db' & 'SerializedDagModel.bulk_sync_to_db' in case # of any Operational Errors @@ -571,6 +601,7 @@ def update_dag_parsing_results_in_db( bundle_name=bundle_name, import_errors=import_errors, session=session, + bundle_path=bundle_path, ) except Exception: log.exception("Error logging import errors!") diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index 3f60cb3f4c4de..7c6565038bab5 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/airflow-core/src/airflow/dag_processing/manager.py @@ -1243,6 +1243,7 @@ def handle_parsing_result( parsing_result=proc.parsing_result, run_duration=run_duration, relative_fileloc=str(file.rel_path), + bundle_path=file.bundle_path, session=session, ) except Exception: @@ -1276,6 +1277,7 @@ def persist_parsing_result( run_duration: float, relative_fileloc: str | None, session: Session, + bundle_path: Path | None = None, ) -> None: """Persist parsed DAG data to the metadata database.""" import_errors: dict[tuple[str, str], str] = {} @@ -1305,6 +1307,7 @@ def persist_parsing_result( warnings=set(warnings), session=session, files_parsed=files_parsed, + bundle_path=bundle_path, ) def _collect_results(self): diff --git a/airflow-core/tests/unit/dag_processing/test_collection.py b/airflow-core/tests/unit/dag_processing/test_collection.py index ad21350be6bcb..5c20611211d7d 100644 --- a/airflow-core/tests/unit/dag_processing/test_collection.py +++ b/airflow-core/tests/unit/dag_processing/test_collection.py @@ -22,6 +22,7 @@ import warnings from collections.abc import Generator from datetime import timedelta +from pathlib import Path from typing import TYPE_CHECKING from unittest import mock from unittest.mock import patch @@ -39,6 +40,7 @@ _get_latest_runs_stmt, _get_latest_runs_stmt_partitioned, _update_dag_tags, + _update_import_errors, update_dag_parsing_results_in_db, ) from airflow.exceptions import SerializationError @@ -905,6 +907,45 @@ def test_serialized_dag_errors_are_import_errors( assert len(dag_import_error_listener.existing) == 0 assert dag_import_error_listener.new["abc.py"] == import_error.stacktrace + @patch.object(ParseImportError, "full_file_path") + @pytest.mark.usefixtures("clean_db") + def test_import_error_listener_uses_bundle_path_without_full_file_path( + self, mock_full_path, session, dag_import_error_listener, testing_dag_bundle + ): + """ + Import errors are persisted and the listener filename is built from ``bundle_path``. + + ``ParseImportError.full_file_path()`` re-instantiates the DAG bundle (and, for some + bundle types such as the Git bundle, resolves an Airflow Connection); calling it inside + the parsing transaction could roll back the not-yet-flushed import error row, leaving the + DAG marked as broken while nothing showed up in the UI. When the caller threads the + already-known ``bundle_path`` down, the filename is built directly and + ``full_file_path()`` must not be called at all. + """ + bundle_path = Path("/opt/airflow/bundles/testing") + import_errors = {("testing", "abc.py"): "AnImportError"} + _update_import_errors( + files_parsed={("testing", "abc.py")}, + bundle_name="testing", + import_errors=import_errors, + session=session, + bundle_path=bundle_path, + ) + session.flush() + + # The bundle-instantiating full_file_path() must not be used when bundle_path is given. + mock_full_path.assert_not_called() + + # The import error is still persisted. + persisted = session.scalars(select(ParseImportError)).all() + assert len(persisted) == 1 + assert persisted[0].bundle_name == "testing" + assert persisted[0].filename == "abc.py" + assert persisted[0].stacktrace == "AnImportError" + + # The listener still received the correct filename, built from bundle_path. + assert dag_import_error_listener.new[str(bundle_path / "abc.py")] == "AnImportError" + @patch.object(ParseImportError, "full_file_path") @mark_fab_auth_manager_test @conf_vars({("core", "min_serialized_dag_update_interval"): "5"}) From 73e8faef9f4b631c3d69bca8d60b722900994e0b Mon Sep 17 00:00:00 2001 From: kjh0623 <8412070+kjh0623@users.noreply.github.com> Date: Fri, 17 Jul 2026 16:30:51 +0900 Subject: [PATCH 2/4] Thread bundle_path through sync_bag_to_db and skip listener args when no listener is registered Address review feedback: - sync_bag_to_db() now forwards dagbag.bundle_path to update_dag_parsing_results_in_db(), so the CLI/dag.test() path no longer falls back to the bundle-instantiating full_file_path() inside the parsing transaction. - Restore a cheap has_listeners early exit so the fallback (select + full_file_path) never runs when no listener plugin is registered, regardless of whether bundle_path was threaded down. - Use the per-file bundle_name_ consistently when building the listener filename in the new-error branch. - Add regression tests: through sync_bag_to_db() end-to-end (fails if a production caller stops forwarding bundle_path), for the existing-error listener branch, and for the no-listener fallback. --- airflow-core/newsfragments/69845.bugfix.rst | 2 +- .../src/airflow/dag_processing/collection.py | 29 +++-- .../src/airflow/dag_processing/dagbag.py | 1 + .../unit/dag_processing/test_collection.py | 110 ++++++++++++++++++ 4 files changed, 131 insertions(+), 11 deletions(-) diff --git a/airflow-core/newsfragments/69845.bugfix.rst b/airflow-core/newsfragments/69845.bugfix.rst index 4b5918bdc404a..c9cbc2bdb2bb4 100644 --- a/airflow-core/newsfragments/69845.bugfix.rst +++ b/airflow-core/newsfragments/69845.bugfix.rst @@ -1 +1 @@ -Fix DAG import errors being silently dropped (never shown in the *Import Errors* view). While recording an import error, ``_update_import_errors`` called ``ParseImportError.full_file_path()`` to build the import-error listener argument. ``full_file_path()`` re-instantiates the DAG bundle and, for some bundle types (e.g. the Git bundle), resolves an Airflow Connection; doing so inside the DAG-parsing transaction could roll back the not-yet-flushed import error row, leaving the DAG marked as broken with nothing shown in the UI. The already-known bundle path is now threaded down to ``_update_import_errors`` and used to build the listener filename directly, so the bundle is no longer re-instantiated inside the transaction. +Fix DAG import errors being silently dropped (never shown in the *Import Errors* view). While recording an import error, ``_update_import_errors`` called ``ParseImportError.full_file_path()`` to build the import-error listener argument. ``full_file_path()`` re-instantiates the DAG bundle and, for some bundle types (e.g. the Git bundle), resolves an Airflow Connection; doing so inside the DAG-parsing transaction could roll back the not-yet-flushed import error row, leaving the DAG marked as broken with nothing shown in the UI. The already-known bundle path is now threaded down to ``_update_import_errors`` (from both the DAG processor and ``sync_bag_to_db``) and used to build the listener filename directly, so the bundle is no longer re-instantiated inside the transaction. Additionally, when no listener plugin is registered, the listener arguments are no longer computed at all. diff --git a/airflow-core/src/airflow/dag_processing/collection.py b/airflow-core/src/airflow/dag_processing/collection.py index 13a08f5fd30d1..d2301fd047dee 100644 --- a/airflow-core/src/airflow/dag_processing/collection.py +++ b/airflow-core/src/airflow/dag_processing/collection.py @@ -403,6 +403,13 @@ def _update_import_errors( ): from airflow.listeners.listener import get_listener_manager + # When no listener plugin is registered at all, skip building the listener + # arguments entirely: for callers that do not thread ``bundle_path`` down, the + # fallback below would otherwise query ``ParseImportError`` and call + # ``full_file_path()`` -- re-instantiating the DAG bundle inside this parsing + # transaction -- just to feed a hook nobody listens to. + notify_listeners = get_listener_manager().has_listeners + def _listener_filename( relative_fileloc: str, bundle_name_: str, import_error: ParseImportError | None = None ) -> str | None: @@ -464,11 +471,12 @@ def _listener_filename( # sending notification when an existing dag import error occurs try: # todo: make listener accept bundle_name and relative_filename - filename = _listener_filename(relative_fileloc, bundle_name_) - if filename is not None: - get_listener_manager().hook.on_existing_dag_import_error( - filename=filename, stacktrace=stacktrace - ) + if notify_listeners: + filename = _listener_filename(relative_fileloc, bundle_name_) + if filename is not None: + get_listener_manager().hook.on_existing_dag_import_error( + filename=filename, stacktrace=stacktrace + ) except Exception: log.exception("error calling listener") else: @@ -481,11 +489,12 @@ def _listener_filename( session.add(import_error) # sending notification when a new dag import error occurs try: - filename = _listener_filename(relative_fileloc, bundle_name, import_error) - if filename is not None: - get_listener_manager().hook.on_new_dag_import_error( - filename=filename, stacktrace=stacktrace - ) + if notify_listeners: + filename = _listener_filename(relative_fileloc, bundle_name_, import_error) + if filename is not None: + get_listener_manager().hook.on_new_dag_import_error( + filename=filename, stacktrace=stacktrace + ) except Exception: log.exception("error calling listener") session.execute( diff --git a/airflow-core/src/airflow/dag_processing/dagbag.py b/airflow-core/src/airflow/dag_processing/dagbag.py index 3bf82804bffe1..8d4879864765e 100644 --- a/airflow-core/src/airflow/dag_processing/dagbag.py +++ b/airflow-core/src/airflow/dag_processing/dagbag.py @@ -579,4 +579,5 @@ def sync_bag_to_db( session=session, version_data=version_data, files_parsed=files_parsed, + bundle_path=dagbag.bundle_path, ) diff --git a/airflow-core/tests/unit/dag_processing/test_collection.py b/airflow-core/tests/unit/dag_processing/test_collection.py index 5c20611211d7d..562fe6102d92c 100644 --- a/airflow-core/tests/unit/dag_processing/test_collection.py +++ b/airflow-core/tests/unit/dag_processing/test_collection.py @@ -946,6 +946,116 @@ def test_import_error_listener_uses_bundle_path_without_full_file_path( # The listener still received the correct filename, built from bundle_path. assert dag_import_error_listener.new[str(bundle_path / "abc.py")] == "AnImportError" + @patch.object(ParseImportError, "full_file_path") + @pytest.mark.usefixtures("clean_db") + def test_existing_import_error_listener_uses_bundle_path_without_full_file_path( + self, mock_full_path, session, dag_import_error_listener, testing_dag_bundle + ): + """ + The existing-error branch also builds the listener filename from ``bundle_path``. + + When an import error row already exists for the file, ``on_existing_dag_import_error`` + is notified instead of ``on_new_dag_import_error``; that branch previously ran an extra + ``select(ParseImportError)`` and called ``full_file_path()`` just to build the listener + argument. With ``bundle_path`` threaded down, neither must happen. + """ + mock_full_path.side_effect = AssertionError( + "full_file_path() must not be called when bundle_path is provided" + ) + bundle_path = Path("/opt/airflow/bundles/testing") + session.add( + ParseImportError( + filename="abc.py", bundle_name="testing", timestamp=tz.utcnow(), stacktrace="OldError" + ) + ) + session.flush() + + _update_import_errors( + files_parsed={("testing", "abc.py")}, + bundle_name="testing", + import_errors={("testing", "abc.py"): "NewImportError"}, + session=session, + bundle_path=bundle_path, + ) + session.flush() + + # The existing row is updated in place. + persisted = session.scalars(select(ParseImportError)).all() + assert len(persisted) == 1 + assert persisted[0].stacktrace == "NewImportError" + + mock_full_path.assert_not_called() + + # The existing-error listener received the bundle_path-based filename. + assert dag_import_error_listener.new == {} + assert dag_import_error_listener.existing[str(bundle_path / "abc.py")] == "NewImportError" + + @patch.object(ParseImportError, "full_file_path") + @pytest.mark.usefixtures("clean_db") + def test_sync_bag_to_db_does_not_call_full_file_path( + self, mock_full_path, tmp_path, session, dag_import_error_listener, testing_dag_bundle + ): + """ + ``sync_bag_to_db`` threads ``dagbag.bundle_path`` down to ``_update_import_errors``. + + This exercises a production caller end-to-end (instead of the private helper), so it + fails if any caller in the chain stops forwarding ``bundle_path`` and the code falls + back to the bundle-instantiating ``full_file_path()`` inside the parsing transaction. + """ + from airflow.dag_processing.dagbag import DagBag, sync_bag_to_db + + mock_full_path.side_effect = AssertionError( + "full_file_path() must not be called when bundle_path is provided" + ) + bundle_path = tmp_path / "bundle" + bundle_path.mkdir() + dag_file = bundle_path / "broken.py" + dag_file.write_text("raise ImportError('broken dag')") + + dagbag = DagBag(dag_folder=str(dag_file), bundle_path=bundle_path, bundle_name="testing") + assert dagbag.import_errors + + sync_bag_to_db(dagbag, "testing", None, session=session) + session.flush() + + persisted = session.scalars(select(ParseImportError)).all() + assert len(persisted) == 1 + assert persisted[0].bundle_name == "testing" + assert persisted[0].filename == "broken.py" + + mock_full_path.assert_not_called() + assert dag_import_error_listener.new[str(bundle_path / "broken.py")] == persisted[0].stacktrace + + @patch.object(ParseImportError, "full_file_path") + @pytest.mark.usefixtures("clean_db") + def test_import_error_persisted_without_listener_or_bundle_path( + self, mock_full_path, session, testing_dag_bundle + ): + """ + With no listener registered, the ``full_file_path()`` fallback must not run at all. + + Callers that do not thread ``bundle_path`` down would otherwise still hit the + bundle-instantiating fallback inside the parsing transaction just to build an argument + for a hook nobody listens to -- the original silent-drop bug. + """ + mock_full_path.side_effect = AssertionError( + "full_file_path() must not be called when no listener is registered" + ) + _update_import_errors( + files_parsed={("testing", "abc.py")}, + bundle_name="testing", + import_errors={("testing", "abc.py"): "AnImportError"}, + session=session, + ) + session.flush() + + persisted = session.scalars(select(ParseImportError)).all() + assert len(persisted) == 1 + assert persisted[0].filename == "abc.py" + assert persisted[0].stacktrace == "AnImportError" + + mock_full_path.assert_not_called() + @patch.object(ParseImportError, "full_file_path") @mark_fab_auth_manager_test @conf_vars({("core", "min_serialized_dag_update_interval"): "5"}) From 57d1faf854fec9130e565c3e8fb99bcaf27c1ed9 Mon Sep 17 00:00:00 2001 From: kjh0623 <8412070+kjh0623@users.noreply.github.com> Date: Fri, 17 Jul 2026 16:51:17 +0900 Subject: [PATCH 3/4] Check has_listeners inside the listener try blocks and fix safe-mode skip in test - get_listener_manager() loads listener plugins on first call, so evaluating has_listeners at the top of _update_import_errors could let a broken listener plugin abort the whole function and drop the import-error rows -- the very symptom this PR fixes. Move the check back inside the per-file try blocks so a listener-manager failure is logged, as before, without affecting error recording. - The sync_bag_to_db regression test's DAG file did not contain the string 'airflow', so DAG_DISCOVERY_SAFE_MODE (default True) would skip the file entirely and the test would fail in CI. --- .../src/airflow/dag_processing/collection.py | 18 +++++++++--------- .../unit/dag_processing/test_collection.py | 3 ++- 2 files changed, 11 insertions(+), 10 deletions(-) diff --git a/airflow-core/src/airflow/dag_processing/collection.py b/airflow-core/src/airflow/dag_processing/collection.py index d2301fd047dee..864e448930596 100644 --- a/airflow-core/src/airflow/dag_processing/collection.py +++ b/airflow-core/src/airflow/dag_processing/collection.py @@ -403,13 +403,6 @@ def _update_import_errors( ): from airflow.listeners.listener import get_listener_manager - # When no listener plugin is registered at all, skip building the listener - # arguments entirely: for callers that do not thread ``bundle_path`` down, the - # fallback below would otherwise query ``ParseImportError`` and call - # ``full_file_path()`` -- re-instantiating the DAG bundle inside this parsing - # transaction -- just to feed a hook nobody listens to. - notify_listeners = get_listener_manager().has_listeners - def _listener_filename( relative_fileloc: str, bundle_name_: str, import_error: ParseImportError | None = None ) -> str | None: @@ -471,7 +464,14 @@ def _listener_filename( # sending notification when an existing dag import error occurs try: # todo: make listener accept bundle_name and relative_filename - if notify_listeners: + # When no listener plugin is registered, skip building the listener + # arguments entirely: for callers that do not thread ``bundle_path`` + # down, the fallback would otherwise query ``ParseImportError`` and + # call the bundle-instantiating ``full_file_path()`` inside this + # transaction just to feed a hook nobody listens to. The check stays + # inside the try block so a listener-manager failure is logged rather + # than aborting the recording of import errors. + if get_listener_manager().has_listeners: filename = _listener_filename(relative_fileloc, bundle_name_) if filename is not None: get_listener_manager().hook.on_existing_dag_import_error( @@ -489,7 +489,7 @@ def _listener_filename( session.add(import_error) # sending notification when a new dag import error occurs try: - if notify_listeners: + if get_listener_manager().has_listeners: filename = _listener_filename(relative_fileloc, bundle_name_, import_error) if filename is not None: get_listener_manager().hook.on_new_dag_import_error( diff --git a/airflow-core/tests/unit/dag_processing/test_collection.py b/airflow-core/tests/unit/dag_processing/test_collection.py index 562fe6102d92c..8f15220a48af7 100644 --- a/airflow-core/tests/unit/dag_processing/test_collection.py +++ b/airflow-core/tests/unit/dag_processing/test_collection.py @@ -1010,7 +1010,8 @@ def test_sync_bag_to_db_does_not_call_full_file_path( bundle_path = tmp_path / "bundle" bundle_path.mkdir() dag_file = bundle_path / "broken.py" - dag_file.write_text("raise ImportError('broken dag')") + # Must mention both "airflow" and "dag" or DAG_DISCOVERY_SAFE_MODE skips the file. + dag_file.write_text("from airflow.sdk import DAG\nraise ImportError('broken dag')") dagbag = DagBag(dag_folder=str(dag_file), bundle_path=bundle_path, bundle_name="testing") assert dagbag.import_errors From 52ed02e067ad6b666eed4c98f3327023343b6667 Mon Sep 17 00:00:00 2001 From: kjh0623 <8412070+kjh0623@users.noreply.github.com> Date: Mon, 20 Jul 2026 10:56:58 +0900 Subject: [PATCH 4/4] Update test_manager assertion for the new bundle_path argument handle_parsing_result now forwards file.bundle_path to persist_parsing_result, so the exact-kwargs assertion in test_handle_parsing_result_updates_stats_after_successful_persist needs the new argument. Assert the actual TEST_DAGS_FOLDER value (rather than mock.ANY) so the test also verifies the bundle path is threaded through correctly. Signed-off-by: kjh0623 <8412070+kjh0623@users.noreply.github.com> --- airflow-core/tests/unit/dag_processing/test_manager.py | 1 + 1 file changed, 1 insertion(+) diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index 74da895bca30a..83c2c59614ba2 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/airflow-core/tests/unit/dag_processing/test_manager.py @@ -1338,6 +1338,7 @@ def test_handle_parsing_result_updates_stats_after_successful_persist(self, sess parsing_result=processor.parsing_result, run_duration=mock.ANY, relative_fileloc="abc.txt", + bundle_path=TEST_DAGS_FOLDER, session=session, ) assert manager._file_stats[file] is not original_stat