op2 = EmptyOperator(task_id="B")
            op3 = EmptyOperator(task_id="C")
            op4 = EmptyOperator(task_id="D")
            op1.set_upstream([op2, op3])
            op3.set_upstream(op4)
        dag.clear()
        now = pendulum.now("UTC")
        dr = dag.create_dagrun(
            run_id="test_dagrun_success_conditions",
            state=DagRunState.RUNNING,
            execution_date=now,
            data_interval=dag.timetable.infer_manual_data_interval(run_after=now),
            start_date=now,
        )
        
        ti_op1 = dr.get_task_instance(task_id=op1.task_id)
        ti_op1.set_state(state=TaskInstanceState.SUCCESS, session=session)
        ti_op2 = dr.get_task_instance(task_id=op2.task_id)
        ti_op3 = dr.get_task_instance(task_id=op3.task_id)
        ti_op4 = dr.get_task_instance(task_id=op4.task_id)
        
        dr.update_state()
        assert DagRunState.RUNNING == dr.state
        
        ti_op2.set_state(state=TaskInstanceState.FAILED, session=session)
        ti_op3.set_state(state=TaskInstanceState.SUCCESS, session=session)
        ti_op4.set_state(state=TaskInstanceState.SUCCESS, session=session)
        dr.update_state()
        assert DagRunState.SUCCESS == dr.state
    def test_dagrun_deadlock(self, session):