Handle no Dagrun in DagrunIdDep (apache#8389) (apache#11343)

Co-authored-by: Ben Nadler <[email protected]>
RaviTezu · Oct 25, 2020 · d4ecba0 · d4ecba0
1 parent 3cca334
commit d4ecba0
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 2 deletions.
diff --git a/airflow/ti_deps/deps/dagrun_id_dep.py b/airflow/ti_deps/deps/dagrun_id_dep.py
@@ -48,9 +48,9 @@ def _get_dep_statuses(self, ti, session, dep_context=None):
         from airflow.jobs import BackfillJob  # To avoid a circular dependency
         dagrun = ti.get_dagrun(session)
 
-        if not dagrun.run_id or not match(BackfillJob.ID_PREFIX + '.*', dagrun.run_id):
+        if not dagrun or not dagrun.run_id or not match(BackfillJob.ID_PREFIX + '.*', dagrun.run_id):
             yield self._passing_status(
-                reason="Task's DagRun run_id is either NULL "
+                reason="Task's DagRun doesn't exist or the run_id is either NULL "
                        "or doesn't start with {}".format(BackfillJob.ID_PREFIX))
         else:
             yield self._failing_status(

diff --git a/tests/ti_deps/deps/test_dagrun_id_dep.py b/tests/ti_deps/deps/test_dagrun_id_dep.py
@@ -49,3 +49,10 @@ def test_dagrun_id_is_not_backfill(self):
         dagrun.run_id = None
         ti = Mock(get_dagrun=Mock(return_value=dagrun))
         self.assertTrue(DagrunIdDep().is_met(ti=ti))
+
+    def test_dagrun_is_none(self):
+        """
+        Task instances which don't yet have an associated dagrun.
+        """
+        ti = Mock(get_dagrun=Mock(return_value=None))
+        self.assertTrue(DagrunIdDep().is_met(ti=ti))