'Tests that a Dag created without an end_date can only be scheduled up
to and including the current datetime.
For example, if today is 2016-01-01 and we are scheduling from a
start_date of 2015-01-01, only jobs up to, but not including
2016-01-01 should be scheduled.'

@mock.patch('airflow.jobs.datetime', FakeDatetime)
def test_schedule_dag_no_end_date_up_to_today_only(self):

    from datetime import datetime
    FakeDatetime.now = classmethod((lambda cls: datetime(2016, 1, 1)))
    session = settings.Session()
    delta = timedelta(days=1)
    start_date = DEFAULT_DATE
    runs = 365
    dag = DAG((TEST_DAG_ID + 'test_schedule_dag_no_end_date_up_to_today_only'), start_date=start_date, schedule_interval=delta)
    dag.add_task(models.BaseOperator(task_id='faketastic', owner='Also    fake'))
    dag_runs = []
    scheduler = jobs.SchedulerJob(**self.default_scheduler_args)
    for i in range(runs):
        dag_run = scheduler.create_dag_run(dag)
        dag_runs.append(dag_run)
        dag_run.state = State.SUCCESS
        session.merge(dag_run)
        session.commit()
    additional_dag_run = scheduler.create_dag_run(dag)
    for dag_run in dag_runs:
        self.assertIsNotNone(dag_run)
    self.assertIsNone(additional_dag_run)
