op1.set_upstream(op2)
        dag.clear()
        now = pendulum.now("UTC")
        dr = dag.create_dagrun(
            run_id="test_dagrun_update_state_end_date",
            state=DagRunState.RUNNING,
            execution_date=now,
            data_interval=dag.timetable.infer_manual_data_interval(now),
            start_date=now,
        )
        
        
        
        session.merge(dr)
        session.commit()
        assert dr.end_date is None
        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_op2.set_state(state=TaskInstanceState.SUCCESS, session=session)
        dr.update_state()
        dr_database = session.query(DagRun).filter(DagRun.run_id == "test_dagrun_update_state_end_date").one()
        assert dr_database.end_date is not None
        assert dr.end_date == dr_database.end_date
        ti_op1.set_state(state=TaskInstanceState.RUNNING, session=session)
        ti_op2.set_state(state=TaskInstanceState.RUNNING, session=session)
        dr.update_state()
        dr_database = session.query(DagRun).filter(DagRun.run_id == "test_dagrun_update_state_end_date").one()
        assert dr._state == DagRunState.RUNNING
        assert dr.end_date is None