[v3-0-test] fix: task-sdk AssetEventOperations.get to use alias_name when specified (#52303)

task-sdk/src/airflow/sdk/api/client.py

@@ -546,7 +546,7 @@ def get(
         if name or uri:
             resp = self.client.get("asset-events/by-asset", params={"name": name, "uri": uri})
         elif alias_name:
-            resp = self.client.get("asset-events/by-asset-alias", params={"name": name})
+            resp = self.client.get("asset-events/by-asset-alias", params={"name": alias_name})
         else:
             raise ValueError("Either `name`, `uri` or `alias_name` must be provided")
 

task-sdk/tests/task_sdk/api/test_client.py

@@ -28,6 +28,7 @@
 
 from airflow.sdk.api.client import RemoteValidationError, ServerResponseError
 from airflow.sdk.api.datamodels._generated import (
+    AssetEventsResponse,
     AssetResponse,
     ConnectionResponse,
     DagRunState,
@@ -875,6 +876,52 @@ def handle_request(request: httpx.Request) -> httpx.Response:
         assert result.error == ErrorType.CONNECTION_NOT_FOUND
 
 
+class TestAssetEventOperations:
+    @pytest.mark.parametrize(
+        "request_params",
+        [
+            ({"name": "this_asset", "uri": "s3://bucket/key"}),
+            ({"alias_name": "this_asset_alias"}),
+        ],
+    )
+    def test_by_name_get_success(self, request_params):
+        def handle_request(request: httpx.Request) -> httpx.Response:
+            params = request.url.params
+            if request.url.path == "/asset-events/by-asset":
+                assert params.get("name") == request_params.get("name")
+                assert params.get("uri") == request_params.get("uri")
+            elif request.url.path == "/asset-events/by-asset-alias":
+                assert params.get("name") == request_params.get("alias_name")
+            else:
+                return httpx.Response(status_code=400, json={"detail": "Bad Request"})
+
+            return httpx.Response(
+                status_code=200,
+                json={
+                    "asset_events": [
+                        {
+                            "id": 1,
+                            "asset": {
+                                "name": "this_asset",
+                                "uri": "s3://bucket/key",
+                                "group": "asset",
+                            },
+                            "created_dagruns": [],
+                            "timestamp": "2023-01-01T00:00:00Z",
+                        }
+                    ]
+                },
+            )
+
+        client = make_client(transport=httpx.MockTransport(handle_request))
+        result = client.asset_events.get(**request_params)
+
+        assert isinstance(result, AssetEventsResponse)
+        assert len(result.asset_events) == 1
+        assert result.asset_events[0].asset.name == "this_asset"
+        assert result.asset_events[0].asset.uri == "s3://bucket/key"
+
+
 class TestAssetOperations:
     @pytest.mark.parametrize(
         "request_params",