'Integration test of the scheduler not accidentally resetting
the try_numbers for a task'

@unittest.skipUnless((u'INTEGRATION' in os.environ), u'Can    only    run    end    to    end')
def test_retry_handling_job(self):

    dag = self.dagbag.get_dag(u'test_retry_handling_job')
    dag_task1 = dag.get_task(u'test_retry_handling_op')
    dag.clear()
    scheduler = SchedulerJob(dag_id=dag.dag_id, num_runs=1)
    scheduler.heartrate = 0
    scheduler.run()
    session = settings.Session()
    ti = session.query(TI).filter((TI.dag_id == dag.dag_id), (TI.task_id == dag_task1.task_id)).first()
    self.assertEqual(ti.try_number, 2)
    self.assertEqual(ti.state, State.UP_FOR_RETRY)
