dag = DAG("test_dagrun_no_deadlock_with_shutdown", start_date=DEFAULT_DATE)
        with dag:
            op1 = EmptyOperator(task_id="upstream_task")
            op2 = EmptyOperator(task_id="downstream_task")
            op2.set_upstream(op1)
        dr = dag.create_dagrun(
            run_id="test_dagrun_no_deadlock_with_shutdown",
            state=DagRunState.RUNNING,
            execution_date=DEFAULT_DATE,
            data_interval=dag.timetable.infer_manual_data_interval(run_after=DEFAULT_DATE),
            start_date=DEFAULT_DATE,
        )
        upstream_ti = dr.get_task_instance(task_id="upstream_task")
        upstream_ti.set_state(TaskInstanceState.SHUTDOWN, session=session)
        dr.update_state()
        assert dr.state == DagRunState.RUNNING
    def test_dagrun_no_deadlock_with_depends_on_past(self, session):