From f47011d4982249c86b32e0d6ee179688d576bf0f Mon Sep 17 00:00:00 2001 From: Henry Chen Date: Sun, 26 Jul 2026 15:05:38 +0800 Subject: [PATCH] Fix incorrect asset queued-event endpoint paths in airflowctl (#70461) (cherry picked from commit 15df3aaac07284aa4d4afecd242aa65cd9685761) --- airflow-ctl/src/airflowctl/api/operations.py | 6 ++-- .../tests/airflow_ctl/api/test_operations.py | 31 +++++++++++++++++++ 2 files changed, 34 insertions(+), 3 deletions(-) diff --git a/airflow-ctl/src/airflowctl/api/operations.py b/airflow-ctl/src/airflowctl/api/operations.py index 3b840845f41c4..6136600467668 100644 --- a/airflow-ctl/src/airflowctl/api/operations.py +++ b/airflow-ctl/src/airflowctl/api/operations.py @@ -300,17 +300,17 @@ def get_dag_queued_event(self, dag_id: str, asset_id: str) -> QueuedEventRespons def delete_queued_events(self, asset_id: str) -> str | ServerResponseError: """Delete a queued event for an asset.""" - self.client.delete(f"assets/{asset_id}/queuedEvents/") + self.client.delete(f"assets/{asset_id}/queuedEvents") return asset_id def delete_dag_queued_events(self, dag_id: str, before: str) -> str | ServerResponseError: """Delete a queued event for a Dag.""" - self.client.delete(f"assets/dags/{dag_id}/queuedEvents", params={"before": before}) + self.client.delete(f"dags/{dag_id}/assets/queuedEvents", params={"before": before}) return dag_id def delete_queued_event(self, dag_id: str, asset_id: str) -> str | ServerResponseError: """Delete a queued event for a Dag.""" - self.client.delete(f"assets/dags/{dag_id}/assets/{asset_id}/queuedEvents/") + self.client.delete(f"dags/{dag_id}/assets/{asset_id}/queuedEvents") return asset_id diff --git a/airflow-ctl/tests/airflow_ctl/api/test_operations.py b/airflow-ctl/tests/airflow_ctl/api/test_operations.py index 52faecee73ea0..cf61ed5dafa72 100644 --- a/airflow-ctl/tests/airflow_ctl/api/test_operations.py +++ b/airflow-ctl/tests/airflow_ctl/api/test_operations.py @@ -431,6 +431,37 @@ def handle_request(request: httpx.Request) -> httpx.Response: response = client.assets.get_dag_queued_event(dag_id=self.dag_id, asset_id=self.asset_id) assert response == self.asset_queued_event_response + def test_delete_queued_events(self): + def handle_request(request: httpx.Request) -> httpx.Response: + assert request.method == "DELETE" + assert request.url.path == f"/api/v2/assets/{self.asset_id}/queuedEvents" + return httpx.Response(204) + + client = make_api_client(transport=httpx.MockTransport(handle_request)) + response = client.assets.delete_queued_events(asset_id=self.asset_id) + assert response == self.asset_id + + def test_delete_dag_queued_events(self): + def handle_request(request: httpx.Request) -> httpx.Response: + assert request.method == "DELETE" + assert request.url.path == f"/api/v2/dags/{self.dag_id}/assets/queuedEvents" + assert request.url.params["before"] == self.before + return httpx.Response(204) + + client = make_api_client(transport=httpx.MockTransport(handle_request)) + response = client.assets.delete_dag_queued_events(dag_id=self.dag_id, before=self.before) + assert response == self.dag_id + + def test_delete_queued_event(self): + def handle_request(request: httpx.Request) -> httpx.Response: + assert request.method == "DELETE" + assert request.url.path == f"/api/v2/dags/{self.dag_id}/assets/{self.asset_id}/queuedEvents" + return httpx.Response(204) + + client = make_api_client(transport=httpx.MockTransport(handle_request)) + response = client.assets.delete_queued_event(dag_id=self.dag_id, asset_id=self.asset_id) + assert response == self.asset_id + class TestBackfillOperations: backfill_id: NonNegativeInt = 1