'Checks if tasks that are not taken up by the executor
get rescheduled'

def test_scheduler_reschedule(self):

    executor = TestExecutor()
    dagbag = DagBag(executor=executor)
    dagbag.dags.clear()
    dagbag.executor = executor
    dag = DAG(dag_id=u'test_scheduler_reschedule', start_date=DEFAULT_DATE)
    dag_task1 = DummyOperator(task_id=u'dummy', dag=dag, owner=u'airflow')
    dag.clear()
    dag.is_subdag = False
    session = settings.Session()
    orm_dag = DagModel(dag_id=dag.dag_id)
    orm_dag.is_paused = False
    session.merge(orm_dag)
    session.commit()
    dagbag.bag_dag(dag=dag, root_dag=dag, parent_dag=dag)
    @mock.patch(u'airflow.models.DagBag', return_value=dagbag)
    @mock.patch(u'airflow.models.DagBag.collect_dags')
    def do_schedule(function, function2):
        scheduler = SchedulerJob(num_runs=1, executor=executor, subdir=os.path.join(settings.DAGS_FOLDER, u'no_dags.py'))
        scheduler.heartrate = 0
        scheduler.run()
    do_schedule()
    self.assertEquals(1, len(executor.queued_tasks))
    executor.queued_tasks.clear()
    do_schedule()
    self.assertEquals(2, len(executor.queued_tasks))
