'Make sure a task in an orphaned state is not reset if it has no dagrun.'

def test_reset_orphaned_tasks_nonexistent_dagrun(self):

    dag_id = u'test_reset_orphaned_tasks_nonexistent_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()
    ti = models.TaskInstance(task=task, execution_date=DEFAULT_DATE)
    session.add(ti)
    session.commit()
    ti.refresh_from_db()
    ti.state = State.SCHEDULED
    session.merge(ti)
    session.commit()
    self.assertEquals(0, len(scheduler.reset_state_for_orphaned_tasks(session=session)))
