dag = DAG("text_dagrun_deadlock", start_date=DEFAULT_DATE, default_args={"owner": "owner1"})
        with dag:
            op1 = EmptyOperator(task_id="A")
            op2 = EmptyOperator(task_id="B")
            op2.trigger_rule = TriggerRule.ONE_FAILED
            op2.set_upstream(op1)
        dag.clear()
        now = pendulum.now("UTC")
        dr = dag.create_dagrun(
            run_id="test_dagrun_deadlock",
            state=DagRunState.RUNNING,
            execution_date=now,
            data_interval=dag.timetable.infer_manual_data_interval(run_after=now),
            start_date=now,
            session=session,
        )
        ti_op1: TI = dr.get_task_instance(task_id=op1.task_id, session=session)
        ti_op2: TI = dr.get_task_instance(task_id=op2.task_id, session=session)
        ti_op1.set_state(state=TaskInstanceState.SUCCESS, session=session)
        ti_op2.set_state(state=None, session=session)
        dr.update_state(session=session)
        assert dr.state == DagRunState.RUNNING
        ti_op2.set_state(state=None, session=session)
        op2.trigger_rule = "invalid"  
        dr.update_state(session=session)
        assert dr.state == DagRunState.FAILED
    def test_dagrun_no_deadlock_with_shutdown(self, session):