'Tests that tasks without dagrun still get executed.'

def test_execute_task_instances_no_dagrun_task_will_execute(self):

    dag_id = u'SchedulerJobTest.test_execute_task_instances_no_dagrun_task_will_execute'
    task_id_1 = u'dummy_task'
    dag = DAG(dag_id=dag_id, start_date=DEFAULT_DATE)
    task1 = DummyOperator(dag=dag, task_id=task_id_1)
    dagbag = SimpleDagBag([dag])
    scheduler = SchedulerJob(**self.default_scheduler_args)
    session = settings.Session()
    dr1 = scheduler.create_dag_run(dag)
    ti1 = TI(task1, DEFAULT_DATE)
    ti1.state = State.SCHEDULED
    ti1.execution_date = (ti1.execution_date + datetime.timedelta(days=1))
    session.merge(ti1)
    session.commit()
    scheduler._execute_task_instances(dagbag, [State.SCHEDULED])
    ti1.refresh_from_db()
    self.assertEquals(State.QUEUED, ti1.state)
