session.add(dr)
        session.commit()
        assert dr.end_date is None
        dr.set_state(DagRunState.SUCCESS)
        session.merge(dr)
        session.commit()
        dr_database = session.query(DagRun).filter(DagRun.run_id == "test_dagrun_set_state_end_date").one()
        assert dr_database.end_date is not None
        assert dr.end_date == dr_database.end_date
        dr.set_state(DagRunState.RUNNING)
        session.merge(dr)
        session.commit()
        dr_database = session.query(DagRun).filter(DagRun.run_id == "test_dagrun_set_state_end_date").one()
        assert dr_database.end_date is None
        dr.set_state(DagRunState.FAILED)
        session.merge(dr)
        session.commit()
        dr_database = session.query(DagRun).filter(DagRun.run_id == "test_dagrun_set_state_end_date").one()
        assert dr_database.end_date is not None
        assert dr.end_date == dr_database.end_date
    def test_dagrun_update_state_end_date(self, session):
        dag = DAG(
            "test_dagrun_update_state_end_date", start_date=DEFAULT_DATE, default_args={"owner": "owner1"}
        )
        
        with dag:
            op1 = EmptyOperator(task_id="A")
            op2 = EmptyOperator(task_id="B")