Add test

Author: ephraimbuddy

airflow/dag_processing/manager.py

@@ -403,6 +403,7 @@ def _add_callback_to_queue(self, request: CallbackRequest):
             bundle = DagBundlesManager().get_bundle(name=request.bundle_name, version=request.bundle_version)
         except ValueError:
             # Bundle no longer configured
+            self.log.error("Bundle %s no longer configured, skipping callback", request.bundle_name)
             return None
 
         dag_absolute_path = os.fspath(Path(bundle.path, request.filepath))

tests/dag_processing/test_manager.py

@@ -858,7 +858,7 @@ def test_non_versioned_bundle_get_version_not_called(self):
         mybundle.refresh_interval = 0
         mybundle.supports_versioning = False
 
-        with conf_vars({("dag_bundles", "config_list"): json.dumps(config)}):
+        with conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(config)}):
             DagBundlesManager().sync_bundles_to_db()
             with mock.patch(
                 "airflow.dag_processing.bundles.manager.DagBundlesManager"
@@ -887,7 +887,7 @@ def test_versioned_bundle_get_version_called_once(self):
         mybundle.supports_versioning = True
         mybundle.get_current_version.return_value = "123"
 
-        with conf_vars({("dag_bundles", "config_list"): json.dumps(config)}):
+        with conf_vars({("dag_processor", "dag_bundle_config_list"): json.dumps(config)}):
             DagBundlesManager().sync_bundles_to_db()
             with mock.patch(
                 "airflow.dag_processing.bundles.manager.DagBundlesManager"
@@ -899,5 +899,56 @@ def test_versioned_bundle_get_version_called_once(self):
 
                 # now run it again so we can check we only call get_current_version once
                 mybundle.get_current_version.reset_mock()
+                mybundle.supports_versioning = True
                 manager.run()
                 mybundle.get_current_version.assert_called_once()
+
+    def test_no_longer_existing_bundle_doesnt_add_callack_to_queue(self, configure_testing_dag_bundle):
+        """Make sure that if a bundle is no longer available, we don't add a callback for the bundle"""
+
+        manager = DagFileProcessorManager(max_runs=1)
+        callback = DagCallbackRequest(
+            dag_id="test_start_date_scheduling",
+            bundle_name="testing",
+            filepath="test_on_failure_callback_dag.py",
+            is_failure_callback=True,
+            run_id="123",
+        )
+        manager._add_callback_to_queue(callback)
+        # No callback added since the bundle is not configured
+        assert manager._file_path_queue == deque()
+        # let's configure a bundle
+        with configure_testing_dag_bundle("/tmp"):
+            manager = DagFileProcessorManager(max_runs=1)
+            manager._add_callback_to_queue(callback)
+            # Now the callback should be added
+            assert manager._file_path_queue == deque(
+                [
+                    DagFileInfo(
+                        bundle_name="testing",
+                        path="/tmp/test_on_failure_callback_dag.py",
+                        bundle_version=None,
+                    )
+                ]
+            )
+
+    def test_callbacks_are_added_with_the_version_of_the_bundle(self, configure_testing_dag_bundle):
+        """Make sure that when we add a callback, we add the version of the bundle to the callback"""
+        with configure_testing_dag_bundle("/tmp"):
+            manager = DagFileProcessorManager(max_runs=1)
+            callback = DagCallbackRequest(
+                dag_id="test_start_date_scheduling",
+                bundle_name="testing",
+                filepath="test_on_failure_callback_dag.py",
+                is_failure_callback=True,
+                run_id="123",
+                bundle_version="1",
+            )
+            manager._add_callback_to_queue(callback)
+            assert manager._file_path_queue == deque(
+                [
+                    DagFileInfo(
+                        bundle_name="testing", path="/tmp/test_on_failure_callback_dag.py", bundle_version="1"
+                    )
+                ]
+            )