'Test that queued tasks are executed by BackfillJob
Test for https://github.com/airbnb/airflow/pull/1225'

def test_backfill_pooled_tasks(self):

    session = settings.Session()
    pool = Pool(pool=u'test_backfill_pooled_task_pool', slots=1)
    session.add(pool)
    session.commit()
    dag = self.dagbag.get_dag(u'test_backfill_pooled_task_dag')
    dag.clear()
    job = BackfillJob(dag=dag, start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
    with timeout(seconds=30):
        job.run()
    ti = TI(task=dag.get_task(u'test_backfill_pooled_task'), execution_date=DEFAULT_DATE)
    ti.refresh_from_db()
    self.assertEqual(ti.state, State.SUCCESS)
