'This checks the defensive against non existent tasks in a dag run'

def test_without_dag_run(self):

    value = False
    dag = DAG(u'shortcircuit_operator_test_without_dag_run', default_args={u'owner': u'airflow', u'start_date': DEFAULT_DATE}, schedule_interval=INTERVAL)
    short_op = ShortCircuitOperator(task_id=u'make_choice', dag=dag, python_callable=(lambda : value))
    branch_1 = DummyOperator(task_id=u'branch_1', dag=dag)
    branch_1.set_upstream(short_op)
    branch_2 = DummyOperator(task_id=u'branch_2', dag=dag)
    branch_2.set_upstream(branch_1)
    upstream = DummyOperator(task_id=u'upstream', dag=dag)
    upstream.set_downstream(short_op)
    dag.clear()
    short_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
    session = Session()
    tis = session.query(TI).filter((TI.dag_id == dag.dag_id), (TI.execution_date == DEFAULT_DATE))
    for ti in tis:
        if (ti.task_id == u'make_choice'):
            self.assertEquals(ti.state, State.SUCCESS)
        elif (ti.task_id == u'upstream'):
            raise
        elif ((ti.task_id == u'branch_1') or (ti.task_id == u'branch_2')):
            self.assertEquals(ti.state, State.SKIPPED)
        else:
            raise
    value = True
    dag.clear()
    short_op.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE)
    for ti in tis:
        if (ti.task_id == u'make_choice'):
            self.assertEquals(ti.state, State.SUCCESS)
        elif (ti.task_id == u'upstream'):
            raise
        elif ((ti.task_id == u'branch_1') or (ti.task_id == u'branch_2')):
            self.assertEquals(ti.state, State.NONE)
        else:
            raise
    session.close()
