'Try to reset when we specify a dagrun and ensure nothing else is.'

def test_reset_orphaned_tasks_specified_dagrun(self):

    dag_id = u'test_reset_orphaned_tasks_specified_dagrun'
    dag = DAG(dag_id=dag_id, start_date=DEFAULT_DATE, schedule_interval=u'@daily')
    task_id = (dag_id + u'_task')
    task = DummyOperator(task_id=task_id, dag=dag)
    scheduler = SchedulerJob(**self.default_scheduler_args)
    session = settings.Session()
    dr1 = scheduler.create_dag_run(dag)
    dr2 = scheduler.create_dag_run(dag)
    dr1.state = State.SUCCESS
    dr2.state = State.RUNNING
    ti1 = dr1.get_task_instances(session=session)[0]
    ti2 = dr2.get_task_instances(session=session)[0]
    ti1.state = State.SCHEDULED
    ti2.state = State.SCHEDULED
    session.merge(ti1)
    session.merge(ti2)
    session.merge(dr1)
    session.merge(dr2)
    session.commit()
    reset_tis = scheduler.reset_state_for_orphaned_tasks(filter_by_dag_run=dr2, session=session)
    self.assertEquals(1, len(reset_tis))
    ti1.refresh_from_db(session=session)
    ti2.refresh_from_db(session=session)
    self.assertEquals(State.SCHEDULED, ti1.state)
    self.assertEquals(State.NONE, ti2.state)
