'Test if _process_task_instances only schedules ti\'s up to max_active_runs
(related to issue AIRFLOW-137)'

def test_scheduler_max_active_runs_respected_after_clear(self):

    dag = DAG(dag_id=u'test_scheduler_max_active_runs_respected_after_clear', start_date=DEFAULT_DATE)
    dag.max_active_runs = 3
    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()
    session.close()
    scheduler = SchedulerJob()
    dag.clear()
    scheduler.create_dag_run(dag)
    dag.max_active_runs = 1
    queue = mock.Mock()
    scheduler._process_task_instances(dag, queue=queue)
    queue.append.assert_called_with((dag.dag_id, dag_task1.task_id, DEFAULT_DATE))
