Add response migration for consumed_asset_events events field (#48940)

And, like I should have done in #48888, add tests this time.

The issue was caused by me misunderstanding how Cadwyn works, the
`instructions_to_migrate_to_previous_version` only affect the OpenAPI schema,
but since the change in this case results in an "invalid" model being
returned, we have to write our own migration for the data.

For those curious, the error without this migration function is:

> fastapi.exceptions.ResponseValidationError: 1 validation errors:
>  {'type': 'extra_forbidden', 'loc': ('response', 'dag_run', 'consumed_asset_events'), 'msg': 'Extra inputs are not permitted', 'input': []}

Author: ashb

airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_04_10.py

@@ -17,9 +17,9 @@
 
 from __future__ import annotations
 
-from cadwyn import VersionChange, schema
+from cadwyn import ResponseInfo, VersionChange, convert_response_to_previous_version_for, schema
 
-from airflow.api_fastapi.execution_api.datamodels.taskinstance import DagRun
+from airflow.api_fastapi.execution_api.datamodels.taskinstance import DagRun, TIRunContext
 
 
 class AddConsumedAssetEventsField(VersionChange):
@@ -28,3 +28,7 @@ class AddConsumedAssetEventsField(VersionChange):
     description = __doc__
 
     instructions_to_migrate_to_previous_version = (schema(DagRun).field("consumed_asset_events").didnt_exist,)
+
+    @convert_response_to_previous_version_for(TIRunContext)  # type: ignore
+    def remove_consumed_asset_events(response: ResponseInfo):  # type: ignore
+        response.body["dag_run"].pop("consumed_asset_events")

airflow-core/tests/unit/api_fastapi/execution_api/versions/v2025_03_26/__init__.py

@@ -0,0 +1,16 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.

airflow-core/tests/unit/api_fastapi/execution_api/versions/v2025_03_26/test_task_instances.py

@@ -0,0 +1,99 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from __future__ import annotations
+
+import pytest
+
+from airflow.utils import timezone
+from airflow.utils.state import State
+
+from tests_common.test_utils.db import clear_db_assets, clear_db_runs
+
+pytestmark = pytest.mark.db_test
+
+
+DEFAULT_START_DATE = timezone.parse("2024-10-31T11:00:00Z")
+DEFAULT_END_DATE = timezone.parse("2024-10-31T12:00:00Z")
+
+
+@pytest.fixture
+def ver_client(client):
+    client.headers["Airflow-API-Version"] = "2025-03-26"
+    return client
+
+
+class TestTIUpdateState:
+    def setup_method(self):
+        clear_db_assets()
+        clear_db_runs()
+
+    def teardown_method(self):
+        clear_db_assets()
+        clear_db_runs()
+
+    def test_ti_run(self, ver_client, session, create_task_instance, time_machine):
+        """
+        Test that this version of the endpoint works.
+
+        Later versions add a consumed_asset_events field.
+        """
+        instant_str = "2024-09-30T12:00:00Z"
+        instant = timezone.parse(instant_str)
+        time_machine.move_to(instant, tick=False)
+
+        ti = create_task_instance(
+            task_id="test_ti_run_state_to_running",
+            state=State.QUEUED,
+            session=session,
+            start_date=instant,
+        )
+        session.commit()
+
+        response = ver_client.patch(
+            f"/execution/task-instances/{ti.id}/run",
+            json={
+                "state": "running",
+                "hostname": "random-hostname",
+                "unixname": "random-unixname",
+                "pid": 100,
+                "start_date": instant_str,
+            },
+        )
+
+        assert response.status_code == 200
+        assert response.json() == {
+            "dag_run": {
+                "dag_id": "dag",
+                "run_id": "test",
+                "clear_number": 0,
+                "logical_date": instant_str,
+                "data_interval_start": instant.subtract(days=1).to_iso8601_string(),
+                "data_interval_end": instant_str,
+                "run_after": instant_str,
+                "start_date": instant_str,
+                "end_date": None,
+                "run_type": "manual",
+                "conf": {},
+            },
+            "task_reschedule_count": 0,
+            "max_tries": 0,
+            "should_retry": False,
+            "variables": [],
+            "connections": [],
+            "xcom_keys_to_clear": [],
+        }