Fix DagBag bug when a dag has invalid schedule_interval (apache#11344)

rajatsri28 · Oct 9, 2020 · 5605d10 · 5605d10
1 parent 7f674c6
commit 5605d10
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 1 deletion.
diff --git a/airflow/models/dagbag.py b/airflow/models/dagbag.py
@@ -352,9 +352,9 @@ def _process_modules(self, filepath, mods, file_last_changed_on_disk):
                     dag.fileloc = filepath
             try:
                 dag.is_subdag = False
-                self.bag_dag(dag=dag, root_dag=dag)
                 if isinstance(dag.normalized_schedule_interval, str):
                     croniter(dag.normalized_schedule_interval)
+                self.bag_dag(dag=dag, root_dag=dag)
                 found_dags.append(dag)
                 found_dags += dag.subdags
             except (CroniterBadCronError,

diff --git a/tests/models/test_dagbag.py b/tests/models/test_dagbag.py
@@ -174,6 +174,7 @@ def test_process_file_cron_validity_check(self):
         for file in invalid_dag_files:
             dagbag.process_file(os.path.join(TEST_DAGS_FOLDER, file))
         self.assertEqual(len(dagbag.import_errors), len(invalid_dag_files))
+        self.assertEqual(len(dagbag.dags), 0)
 
     @patch.object(DagModel, 'get_current')
     def test_get_dag_without_refresh(self, mock_dagmodel):