'Test if the schedule_interval will be auto aligned with the start_date
such that if the start_date coincides with the schedule the first
execution_date will be start_date, otherwise it will be start_date +
interval.'

def test_scheduler_auto_align(self):

    dag = DAG(dag_id=u'test_scheduler_auto_align_1', start_date=datetime.datetime(2016, 1, 1, 10, 10, 0), schedule_interval=u'4    5    *    *    *')
    dag_task1 = DummyOperator(task_id=u'dummy', dag=dag, owner=u'airflow')
    session = settings.Session()
    orm_dag = DagModel(dag_id=dag.dag_id)
    session.merge(orm_dag)
    session.commit()
    scheduler = SchedulerJob()
    dag.clear()
    dr = scheduler.create_dag_run(dag)
    self.assertIsNotNone(dr)
    self.assertEquals(dr.execution_date, datetime.datetime(2016, 1, 2, 5, 4))
    dag = DAG(dag_id=u'test_scheduler_auto_align_2', start_date=datetime.datetime(2016, 1, 1, 10, 10, 0), schedule_interval=u'10    10    *    *    *')
    dag_task1 = DummyOperator(task_id=u'dummy', dag=dag, owner=u'airflow')
    session = settings.Session()
    orm_dag = DagModel(dag_id=dag.dag_id)
    session.merge(orm_dag)
    session.commit()
    scheduler = SchedulerJob()
    dag.clear()
    dr = scheduler.create_dag_run(dag)
    self.assertIsNotNone(dr)
    self.assertEquals(dr.execution_date, datetime.datetime(2016, 1, 1, 10, 10))
