tardunge commented on code in PR #63928:
URL: https://github.com/apache/airflow/pull/63928#discussion_r3147725636
##########
airflow-core/tests/unit/dag_processing/bundles/test_dag_bundle_manager.py:
##########
@@ -467,3 +470,259 @@ def test_multiple_bundles_one_fails(clear_db, session):
def test_get_all_bundle_names():
assert DagBundlesManager().get_all_bundle_names() == ["dags-folder",
"example_dags"]
+
+
+class TestDagBundleConfigPath:
+ """Tests for dynamic DAG bundle configuration from path."""
+
+ def test_parse_config_from_path(self):
+ """Test parsing DAG bundle configs from JSON files in a directory."""
+ with tempfile.TemporaryDirectory() as tmpdir:
+ bundle1 = {
+ "name": "test-bundle-1",
+ "classpath":
"unit.dag_processing.bundles.test_dag_bundle_manager.BasicBundle",
+ "kwargs": {"refresh_interval": 1},
+ }
+ bundle2 = {
+ "name": "test-bundle-2",
+ "classpath":
"unit.dag_processing.bundles.test_dag_bundle_manager.BasicBundle",
+ "kwargs": {"refresh_interval": 2},
+ }
+
+
Path(tmpdir).joinpath("bundle1.json").write_text(json.dumps(bundle1))
+
Path(tmpdir).joinpath("bundle2.json").write_text(json.dumps(bundle2))
+
+ with conf_vars(
+ {
+ ("dag_processor", "dag_bundle_config_path"): tmpdir,
+ ("core", "LOAD_EXAMPLES"): "False",
+ }
+ ):
+ manager = DagBundlesManager()
+ bundle_names = {b.name for b in manager.get_all_dag_bundles()}
+ assert bundle_names == {"test-bundle-1", "test-bundle-2"}
+
+ def test_config_path_takes_precedence_over_list(self):
+ """Test that config path takes precedence over config list."""
+ with tempfile.TemporaryDirectory() as tmpdir:
+ bundle = {
+ "name": "path-bundle",
+ "classpath":
"unit.dag_processing.bundles.test_dag_bundle_manager.BasicBundle",
+ "kwargs": {"refresh_interval": 1},
+ }
+ Path(tmpdir).joinpath("bundle.json").write_text(json.dumps(bundle))
+
+ list_config = [
+ {
+ "name": "list-bundle",
+ "classpath":
"unit.dag_processing.bundles.test_dag_bundle_manager.BasicBundle",
+ "kwargs": {"refresh_interval": 1},
+ }
+ ]
+
+ with conf_vars(
+ {
+ ("dag_processor", "dag_bundle_config_list"):
json.dumps(list_config),
+ ("dag_processor", "dag_bundle_config_path"): tmpdir,
+ ("core", "LOAD_EXAMPLES"): "False",
+ }
+ ):
+ manager = DagBundlesManager()
+ bundle_names = {b.name for b in manager.get_all_dag_bundles()}
+ # Should only have bundle from path, not from list
+ assert bundle_names == {"path-bundle"}
+
+ def test_check_config_path_changes(self):
+ """Test detection of config file changes."""
+ with tempfile.TemporaryDirectory() as tmpdir:
+ bundle_file = Path(tmpdir).joinpath("bundle.json")
+ bundle_config = {
+ "name": "test-bundle",
+ "classpath":
"unit.dag_processing.bundles.test_dag_bundle_manager.BasicBundle",
+ "kwargs": {"refresh_interval": 1},
+ }
+ bundle_file.write_text(json.dumps(bundle_config))
+
+ with conf_vars(
+ {
+ ("dag_processor", "dag_bundle_config_path"): tmpdir,
+ ("core", "LOAD_EXAMPLES"): "False",
+ }
+ ):
+ manager = DagBundlesManager()
+
+ # No changes initially
+ assert not manager.check_config_path_changes()
+
+ # Modify the file
+ time.sleep(0.01) # Ensure mtime changes
+ bundle_config["kwargs"]["refresh_interval"] = 99
+ bundle_file.write_text(json.dumps(bundle_config))
Review Comment:
Fixed in 368bfa50 — replaced time.sleep(0.01) with an explicit
os.utime(bundle_file, ns=(new+10s, new+10s)) bump. Test no longer depends on
filesystem timestamp granularity.
--
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]
