'Test to check that a DAG with catchup = False only schedules beginning now, not back to the start date'

def test_dag_catchup_option(self):

    now = datetime.datetime.now()
    six_hours_ago_to_the_hour = (now - datetime.timedelta(hours=6)).replace(minute=0, second=0, microsecond=0)
    three_minutes_ago = (now - datetime.timedelta(minutes=3))
    two_hours_and_three_minutes_ago = (three_minutes_ago - datetime.timedelta(hours=2))
    START_DATE = six_hours_ago_to_the_hour
    DAG_NAME1 = u'no_catchup_test1'
    DAG_NAME2 = u'no_catchup_test2'
    DAG_NAME3 = u'no_catchup_test3'
    default_args = {u'owner': u'airflow', u'depends_on_past': False, u'start_date': START_DATE}
    dag1 = DAG(DAG_NAME1, schedule_interval=u'*    *    *    *    *', max_active_runs=1, default_args=default_args)
    default_catchup = configuration.getboolean(u'scheduler', u'catchup_by_default')
    self.assertEqual(default_catchup, True)
    self.assertEqual(dag1.catchup, True)
    dag2 = DAG(DAG_NAME2, schedule_interval=u'*    *    *    *    *', max_active_runs=1, catchup=False, default_args=default_args)
    run_this_1 = DummyOperator(task_id=u'run_this_1', dag=dag2)
    run_this_2 = DummyOperator(task_id=u'run_this_2', dag=dag2)
    run_this_2.set_upstream(run_this_1)
    run_this_3 = DummyOperator(task_id=u'run_this_3', dag=dag2)
    run_this_3.set_upstream(run_this_2)
    session = settings.Session()
    orm_dag = DagModel(dag_id=dag2.dag_id)
    session.merge(orm_dag)
    session.commit()
    session.close()
    scheduler = SchedulerJob()
    dag2.clear()
    dr = scheduler.create_dag_run(dag2)
    self.assertIsNotNone(dr)
    self.assertGreater(dr.execution_date, three_minutes_ago)
    self.assertLess(dr.execution_date, datetime.datetime.now())
    dag3 = DAG(DAG_NAME3, schedule_interval=u'@hourly', max_active_runs=1, catchup=False, default_args=default_args)
    run_this_1 = DummyOperator(task_id=u'run_this_1', dag=dag3)
    run_this_2 = DummyOperator(task_id=u'run_this_2', dag=dag3)
    run_this_2.set_upstream(run_this_1)
    run_this_3 = DummyOperator(task_id=u'run_this_3', dag=dag3)
    run_this_3.set_upstream(run_this_2)
    session = settings.Session()
    orm_dag = DagModel(dag_id=dag3.dag_id)
    session.merge(orm_dag)
    session.commit()
    session.close()
    scheduler = SchedulerJob()
    dag3.clear()
    dr = None
    dr = scheduler.create_dag_run(dag3)
    self.assertIsNotNone(dr)
    self.assertGreater(dr.execution_date, two_hours_and_three_minutes_ago)
    self.assertLess(dr.execution_date, datetime.datetime.now())
