'Test task instances not queued when pool is full'

@patch.object(TI, u'pool_full')
def test_scheduler_verify_pool_full(self, mock_pool_full):

    mock_pool_full.return_value = False
    dag = DAG(dag_id=u'test_scheduler_verify_pool_full', start_date=DEFAULT_DATE)
    DummyOperator(task_id=u'dummy', dag=dag, owner=u'airflow', pool=u'test_scheduler_verify_pool_full')
    session = settings.Session()
    pool = Pool(pool=u'test_scheduler_verify_pool_full', slots=1)
    session.add(pool)
    orm_dag = DagModel(dag_id=dag.dag_id)
    orm_dag.is_paused = False
    session.merge(orm_dag)
    session.commit()
    scheduler = SchedulerJob()
    dag.clear()
    dr = scheduler.create_dag_run(dag)
    self.assertIsNotNone(dr)
    self.assertEquals(dr.execution_date, DEFAULT_DATE)
    dr = scheduler.create_dag_run(dag)
    self.assertIsNotNone(dr)
    queue = []
    scheduler._process_task_instances(dag, queue=queue)
    self.assertEquals(len(queue), 2)
    dagbag = SimpleDagBag([dag])
    for ti_key in queue:
        task = dag.get_task(ti_key[1])
        ti = TI(task, ti_key[2])
        ti.state = State.SCHEDULED
        session.merge(ti)
        session.commit()
    scheduler._execute_task_instances(dagbag, (State.SCHEDULED, State.UP_FOR_RETRY))
    self.assertEquals(len(scheduler.executor.queued_tasks), 1)
