'Test scheduling a dag where there is a prior DagRun
which has the same run_id as the next run should have'

def test_schedule_dag_fake_scheduled_previous(self):

    delta = timedelta(hours=1)
    dag = DAG((TEST_DAG_ID + 'test_schedule_dag_fake_scheduled_previous'), schedule_interval=delta, start_date=DEFAULT_DATE)
    dag.add_task(models.BaseOperator(task_id='faketastic', owner='Also    fake', start_date=DEFAULT_DATE))
    scheduler = jobs.SchedulerJob(**self.default_scheduler_args)
    dag.create_dagrun(run_id=models.DagRun.id_for_date(DEFAULT_DATE), execution_date=DEFAULT_DATE, state=State.SUCCESS, external_trigger=True)
    dag_run = scheduler.create_dag_run(dag)
    self.assertIsNotNone(dag_run)
    self.assertEqual(dag.dag_id, dag_run.dag_id)
    self.assertIsNotNone(dag_run.run_id)
    self.assertNotEqual('', dag_run.run_id)
    self.assertEqual((DEFAULT_DATE + delta), dag_run.execution_date, msg='dag_run.execution_date    did    not    match    expectation:    {0}'.format(dag_run.execution_date))
    self.assertEqual(State.RUNNING, dag_run.state)
    self.assertFalse(dag_run.external_trigger)
