fix: propagate quota_project_id and api_endpoint in AsyncGrpcClient (#16731)

This PR fixes b/503990490 where AsyncGrpcClient fails to propagate
quota_project_id and api_endpoint from ClientOptions to the gRPC
channel.

Author: chandra-siri

packages/google-cloud-storage/google/cloud/storage/asyncio/async_grpc_client.py

@@ -23,6 +23,8 @@
 )
 from google.cloud.storage import __version__
 
+_DEFAULT_HOST = "storage.googleapis.com"
+
 
 class AsyncGrpcClient:
     """An asynchronous client for interacting with Google Cloud Storage using the gRPC API.
@@ -109,7 +111,15 @@ def _create_async_grpc_client(
 
         primary_user_agent = client_info.to_user_agent()
 
+        host = _DEFAULT_HOST
+        quota_project_id = None
+        if client_options:
+            host = getattr(client_options, "api_endpoint", None) or _DEFAULT_HOST
+            quota_project_id = getattr(client_options, "quota_project_id", None)
+
         channel = transport_cls.create_channel(
+            host=host,
+            quota_project_id=quota_project_id,
             attempt_direct_path=attempt_direct_path,
             credentials=credentials,
             options=(("grpc.primary_user_agent", primary_user_agent),),

packages/google-cloud-storage/tests/unit/asyncio/test_async_grpc_client.py

@@ -53,6 +53,8 @@ def test_constructor_default_options(self, mock_async_storage_client):
         expected_options = (("grpc.primary_user_agent", primary_user_agent),)
 
         mock_transport_cls.create_channel.assert_called_once_with(
+            host="storage.googleapis.com",
+            quota_project_id=None,
             attempt_direct_path=True,
             credentials=mock_creds,
             options=expected_options,
@@ -82,6 +84,39 @@ def test_constructor_with_client_info(self, mock_async_storage_client):
         expected_options = (("grpc.primary_user_agent", primary_user_agent),)
 
         mock_transport_cls.create_channel.assert_called_once_with(
+            host="storage.googleapis.com",
+            quota_project_id=None,
+            attempt_direct_path=True,
+            credentials=mock_creds,
+            options=expected_options,
+        )
+
+    @mock.patch("google.cloud._storage_v2.StorageAsyncClient")
+    def test_constructor_with_quota_project_and_endpoint(
+        self, mock_async_storage_client
+    ):
+        mock_transport_cls = mock.MagicMock()
+        mock_async_storage_client.get_transport_class.return_value = mock_transport_cls
+        mock_creds = _make_credentials()
+
+        from google.api_core import client_options
+
+        mock_client_options = client_options.ClientOptions(
+            api_endpoint="custom-endpoint.com", quota_project_id="my-quota-project"
+        )
+
+        async_grpc_client.AsyncGrpcClient(
+            credentials=mock_creds, client_options=mock_client_options
+        )
+
+        kwargs = mock_async_storage_client.call_args.kwargs
+        client_info = kwargs["client_info"]
+        primary_user_agent = client_info.to_user_agent()
+        expected_options = (("grpc.primary_user_agent", primary_user_agent),)
+
+        mock_transport_cls.create_channel.assert_called_once_with(
+            host="custom-endpoint.com",
+            quota_project_id="my-quota-project",
             attempt_direct_path=True,
             credentials=mock_creds,
             options=expected_options,
@@ -105,6 +140,8 @@ def test_constructor_disables_directpath(self, mock_async_storage_client):
         expected_options = (("grpc.primary_user_agent", primary_user_agent),)
 
         mock_transport_cls.create_channel.assert_called_once_with(
+            host="storage.googleapis.com",
+            quota_project_id=None,
             attempt_direct_path=False,
             credentials=mock_creds,
             options=expected_options,
@@ -147,6 +184,8 @@ def test_grpc_client_property(self, mock_grpc_gapic_client):
         expected_options = (("grpc.primary_user_agent", primary_user_agent),)
 
         mock_transport_cls.create_channel.assert_called_once_with(
+            host="storage.googleapis.com",
+            quota_project_id=None,
             attempt_direct_path=mock_attempt_direct_path,
             credentials=mock_creds,
             options=expected_options,