Return 400 for dag_run_id with include_past/include_future in clearTaskInstances API (apache#47036)

Author: sunank200

airflow/api_fastapi/core_api/routes/public/task_instances.py

@@ -630,8 +630,14 @@ def post_clear_task_instances(
         if dag_run is None:
             error_message = f"Dag Run id {dag_run_id} not found in dag {dag_id}"
             raise HTTPException(status.HTTP_404_NOT_FOUND, error_message)
-        body.start_date = dag_run.logical_date
-        body.end_date = dag_run.logical_date
+
+        if past or future:
+            raise HTTPException(
+                status.HTTP_400_BAD_REQUEST,
+                "Cannot use include_past or include_future when dag_run_id is provided because logical_date is not applicable.",
+            )
+        body.start_date = dag_run.logical_date if dag_run.logical_date is not None else None
+        body.end_date = dag_run.logical_date if dag_run.logical_date is not None else None
 
     if past:
         body.start_date = None

tests/api_fastapi/core_api/routes/public/test_task_instances.py

@@ -2062,6 +2062,31 @@ def test_should_respond_200(
         assert response.status_code == 200
         assert response.json()["total_entries"] == expected_ti
 
+    @pytest.mark.parametrize("flag", ["include_future", "include_past"])
+    def test_dag_run_with_future_or_past_flag_returns_400(self, test_client, session, flag):
+        dag_id = "example_python_operator"
+        payload = {
+            "dry_run": True,
+            "dag_run_id": "TEST_DAG_RUN_ID_0",
+            "only_failed": True,
+            flag: True,
+        }
+        task_instances = [{"logical_date": DEFAULT_DATETIME_1, "state": State.FAILED}]
+        self.create_task_instances(
+            session,
+            dag_id=dag_id,
+            task_instances=task_instances,
+            update_extras=False,
+            dag_run_state=State.FAILED,
+        )
+        self.dagbag.sync_to_db("dags-folder", None)
+        response = test_client.post(f"/public/dags/{dag_id}/clearTaskInstances", json=payload)
+        assert response.status_code == 400
+        assert (
+            "Cannot use include_past or include_future when dag_run_id is provided"
+            in response.json()["detail"]
+        )
+
     @pytest.mark.parametrize(
         "main_dag, task_instances, request_dag, payload, expected_ti",
         [