Properly fix dagrun update state endpoint
diff --git a/airflow/api/common/experimental/mark_tasks.py b/airflow/api/common/experimental/mark_tasks.py
index 945a9cc4a3596..28e733dd96a89 100644
--- a/airflow/api/common/experimental/mark_tasks.py
+++ b/airflow/api/common/experimental/mark_tasks.py
@@ -131,7 +131,6 @@ def set_state(
         if sub_dag_run_ids:
             qry_sub_dag = all_subdag_tasks_query(sub_dag_run_ids, session, state, confirmed_dates)
             tis_altered += qry_sub_dag.all()
-
     return tis_altered
 
 
diff --git a/airflow/api_connexion/endpoints/dag_run_endpoint.py b/airflow/api_connexion/endpoints/dag_run_endpoint.py
index 93278049ba342..edd4f787fe9bf 100644
--- a/airflow/api_connexion/endpoints/dag_run_endpoint.py
+++ b/airflow/api_connexion/endpoints/dag_run_endpoint.py
@@ -21,6 +21,10 @@
 from sqlalchemy import or_
 
 from airflow._vendor.connexion import NoContent
+from airflow.api.common.experimental.mark_tasks import (
+    set_dag_run_state_to_failed,
+    set_dag_run_state_to_success,
+)
 from airflow.api_connexion import security
 from airflow.api_connexion.exceptions import AlreadyExists, BadRequest, NotFound
 from airflow.api_connexion.parameters import apply_sorting, check_limit, format_datetime, format_parameters
@@ -34,7 +38,7 @@
 from airflow.models import DagModel, DagRun
 from airflow.security import permissions
 from airflow.utils.session import provide_session
-from airflow.utils.state import DagRunState, State
+from airflow.utils.state import State
 from airflow.utils.types import DagRunType
 
 
@@ -302,6 +306,10 @@ def update_dag_run_state(dag_id: str, dag_run_id: str, session) -> dict:
         raise BadRequest(detail=str(err))
 
     state = post_body['state']
-    dag_run.set_state(state=DagRunState(state))
-    session.merge(dag_run)
+    dag = current_app.dag_bag.get_dag(dag_id)
+    if state == State.SUCCESS:
+        set_dag_run_state_to_success(dag, dag_run.execution_date, commit=True)
+    else:
+        set_dag_run_state_to_failed(dag, dag_run.execution_date, commit=True)
+    dag_run = session.query(DagRun).get(dag_run.id)
     return dagrun_schema.dump(dag_run)
diff --git a/tests/api_connexion/endpoints/test_dag_run_endpoint.py b/tests/api_connexion/endpoints/test_dag_run_endpoint.py
index 69c06b79dc80d..54410b9d5a217 100644
--- a/tests/api_connexion/endpoints/test_dag_run_endpoint.py
+++ b/tests/api_connexion/endpoints/test_dag_run_endpoint.py
@@ -27,10 +27,11 @@
 from airflow.security import permissions
 from airflow.utils import timezone
 from airflow.utils.session import create_session, provide_session
+from airflow.utils.state import State
 from airflow.utils.types import DagRunType
 from tests.test_utils.api_connexion_utils import assert_401, create_user, delete_roles, delete_user
 from tests.test_utils.config import conf_vars
-from tests.test_utils.db import clear_db_dags, clear_db_runs
+from tests.test_utils.db import clear_db_dags, clear_db_runs, clear_db_serialized_dags
 
 
 @pytest.fixture(scope="module")
@@ -103,11 +104,13 @@ def setup_attrs(self, configured_app) -> None:
         self.app = configured_app
         self.client = self.app.test_client()  # type:ignore
         clear_db_runs()
+        clear_db_serialized_dags()
         clear_db_dags()
 
     def teardown_method(self) -> None:
         clear_db_runs()
         clear_db_dags()
+        clear_db_serialized_dags()
 
     def _create_dag(self, dag_id):
         dag_instance = DagModel(dag_id=dag_id)
@@ -1175,15 +1178,20 @@ def test_should_raises_403_unauthorized(self, username):
         assert response.status_code == 403
 
 
-class TestPostSetDagRunState(TestDagRunEndpoint):
+class TestPatchDagRunState(TestDagRunEndpoint):
     @pytest.mark.parametrize("state", ["failed", "success"])
-    @freeze_time(TestDagRunEndpoint.default_time)
-    def test_should_respond_200(self, state, dag_maker):
+    def test_should_respond_200(self, state, dag_maker, session):
         dag_id = "TEST_DAG_ID"
         dag_run_id = 'TEST_DAG_RUN_ID'
-        with dag_maker(dag_id):
-            DummyOperator(task_id='task_id')
-        dag_maker.create_dagrun(run_id=dag_run_id)
+        with dag_maker(dag_id) as dag:
+            task = DummyOperator(task_id='task_id', dag=dag)
+        self.app.dag_bag.bag_dag(dag, root_dag=dag)
+        dr = dag_maker.create_dagrun(run_id=dag_run_id)
+        ti = dr.get_task_instance(task_id='task_id')
+        ti.task = task
+        ti.state = State.RUNNING
+        session.merge(ti)
+        session.commit()
 
         request_json = {"state": state}
 
@@ -1193,16 +1201,19 @@ def test_should_respond_200(self, state, dag_maker):
             environ_overrides={"REMOTE_USER": "test"},
         )
 
+        ti.refresh_from_db()
+        assert ti.state == state
+        dr = session.query(DagRun).filter(DagRun.run_id == dr.run_id).first()
         assert response.status_code == 200
         assert response.json == {
             'conf': {},
             'dag_id': dag_id,
             'dag_run_id': dag_run_id,
-            'end_date': self.default_time,
-            'execution_date': dag_maker.start_date.isoformat(),
+            'end_date': dr.end_date.isoformat(),
+            'execution_date': dr.execution_date.isoformat(),
             'external_trigger': False,
-            'logical_date': dag_maker.start_date.isoformat(),
-            'start_date': dag_maker.start_date.isoformat(),
+            'logical_date': dr.execution_date.isoformat(),
+            'start_date': dr.start_date.isoformat(),
             'state': state,
         }
 

The dagrun update state endpoint was recently added but is not working
as expected. This PR fixes it to work exactly like the UI mark dagrun state API

Closes: https://github.com/apache/airflow/issues/18363

---
**^ Add meaningful description above**

Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/main/UPDATING.md).
