'Ensure orphaned tasks with non-running dagruns are not reset.'

def test_reset_orphaned_tasks_non_running_dagruns(self):

    dag_id = u'test_reset_orphaned_tasks_non_running_dagruns'
    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)
    dr1.state = State.SUCCESS
    tis = dr1.get_task_instances(session=session)
    self.assertEquals(1, len(tis))
    tis[0].state = State.SCHEDULED
    session.merge(dr1)
    session.merge(tis[0])
    session.commit()
    self.assertEquals(0, len(scheduler.reset_state_for_orphaned_tasks(session=session)))
