[AIRFLOW-1837] Respect task start_date when different from dag's (#4010)

Author: dima-asana

tests/core.py

@@ -2078,14 +2078,14 @@ def test_dag_view_task_with_python_operator_using_partial(self):
         response = self.app.get(
             '/admin/airflow/task?'
             'task_id=test_dagrun_functool_partial&dag_id=test_task_view_type_check&'
-            'execution_date={}'.format(DEFAULT_DATE_DS))
+            'execution_date={}'.format(EXAMPLE_DAG_DEFAULT_DATE))
         self.assertIn("A function with two args", response.data.decode('utf-8'))
 
     def test_dag_view_task_with_python_operator_using_instance(self):
         response = self.app.get(
             '/admin/airflow/task?'
             'task_id=test_dagrun_instance&dag_id=test_task_view_type_check&'
-            'execution_date={}'.format(DEFAULT_DATE_DS))
+            'execution_date={}'.format(EXAMPLE_DAG_DEFAULT_DATE))
         self.assertIn("A __call__ method", response.data.decode('utf-8'))
 
     def tearDown(self):

tests/jobs.py

@@ -2273,6 +2273,27 @@ def test_scheduler_task_start_date(self):
         for t in ti2s:
             self.assertEqual(t.state, State.SUCCESS)
 
+    def test_scheduler_task_start_date(self):
+        """
+        Test that the scheduler respects task start dates that are different
+        from DAG start dates
+        """
+        dag_id = 'test_task_start_date_scheduling'
+        dag = self.dagbag.get_dag(dag_id)
+        dag.clear()
+        scheduler = SchedulerJob(dag_id,
+                                 num_runs=2)
+        scheduler.run()
+
+        session = settings.Session()
+        tiq = session.query(TI).filter(TI.dag_id == dag_id)
+        ti1s = tiq.filter(TI.task_id == 'dummy1').all()
+        ti2s = tiq.filter(TI.task_id == 'dummy2').all()
+        self.assertEqual(len(ti1s), 0)
+        self.assertEqual(len(ti2s), 2)
+        for t in ti2s:
+            self.assertEqual(t.state, State.SUCCESS)
+
     def test_scheduler_multiprocessing(self):
         """
         Test that the scheduler can successfully queue multiple dags in parallel

tests/www_rbac/test_views.py

@@ -409,7 +409,7 @@ def test_paused(self):
     def test_failed(self):
         url = ('failed?task_id=run_this_last&dag_id=example_bash_operator&'
                'execution_date={}&upstream=false&downstream=false&future=false&past=false'
-               .format(self.percent_encode(self.default_date)))
+               .format(self.percent_encode(self.EXAMPLE_DAG_DEFAULT_DATE)))
         resp = self.client.get(url)
         self.check_content_in_response('Wait a minute', resp)