'Checks if the scheduler does not put a task in limbo, when a task is retried
but is still present in the executor.'

def test_retry_still_in_executor(self):

    executor = TestExecutor()
    dagbag = DagBag(executor=executor)
    dagbag.dags.clear()
    dagbag.executor = executor
    dag = DAG(dag_id=u'test_retry_still_in_executor', start_date=DEFAULT_DATE, schedule_interval=u'@once')
    dag_task1 = BashOperator(task_id=u'test_retry_handling_op', bash_command=u'exit    1', retries=1, 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))
    def run_with_error(task):
        try:
            task.run()
        except AirflowException:
            pass
    ti_tuple = six.next(six.itervalues(executor.queued_tasks))
    (command, priority, queue, ti) = ti_tuple
    ti.task = dag_task1
    run_with_error(ti)
    self.assertEqual(ti.state, State.UP_FOR_RETRY)
    self.assertEqual(ti.try_number, 1)
    ti.refresh_from_db(lock_for_update=True, session=session)
    ti.state = State.SCHEDULED
    session.merge(ti)
    session.commit()
    do_schedule()
    self.assertTrue(executor.has_task(ti))
    ti.refresh_from_db()
    self.assertEqual(ti.state, State.SCHEDULED)
    executor.queued_tasks.clear()
    do_schedule()
    ti.refresh_from_db()
    self.assertEqual(ti.state, State.QUEUED)
