bingqin2 commented on code in PR #72906:
URL: https://github.com/apache/airflow/pull/72906#discussion_r4371224897
##########
task-sdk/src/airflow/sdk/definitions/dag.py:
##########
@@ -1378,7 +1384,14 @@ def test(
dag_folder=bundle.path,
bundle_path=bundle.path,
bundle_name=bundle.name,
+ collect_dags=False,
)
+ files_to_parse = _dag_test_files_to_sync(self, bundle,
session=session)
+ if files_to_parse is not None:
+ for file_path in files_to_parse:
+ dagbag.collect_dags(dag_folder=file_path)
+ if files_to_parse is None or self.dag_id not in dagbag.dags:
Review Comment:
Good catch, thank you. You're right: after parsing the selected files, only
the tested Dag was checked, so a target that moved to a new file while its old
file stayed for another Dag was never re-synced, and the trigger used its stale
serialized version.
`_dag_test_files_to_sync` now also returns the ids of the Dags those files
have to define (the tested Dag and its targets in the bundle), and `DAG.test()`
walks the whole bundle when any of them is missing after parsing.
`test_dag_test_walks_bundle_when_trigger_target_moved_to_another_file`
reproduces your scenario. Without the fallback it fails, because the target
stays recorded at `old.py`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
