fix: raise ValueError if api_endpoint is unset when using AnonymousCredentials in AsyncGrpcClient. (#1778)

chandra-siri · web-flow · commit 17828ea31687 · 2026-03-20T03:58:43.000-07:00
Added validation to ensure `api_endpoint` is provided when using
`AnonymousCredentials` in `AsyncGrpcClient`.
diff --git a/google/cloud/storage/asyncio/async_grpc_client.py b/google/cloud/storage/asyncio/async_grpc_client.py
@@ -55,6 +55,10 @@ def __init__(
         attempt_direct_path=True,
     ):
         if isinstance(credentials, auth_credentials.AnonymousCredentials):
+            if client_options is None or client_options.api_endpoint is None:
+                raise ValueError(
+                    "Either client_options or `client_option.api_endpoint` is None. Please provide api_endpoint when `AnonymousCredentials` is used "
+                )
             self._grpc_client = self._create_anonymous_client(
                 client_options, credentials
             )
diff --git a/tests/unit/asyncio/test_async_grpc_client.py b/tests/unit/asyncio/test_async_grpc_client.py
@@ -184,6 +184,23 @@ def test_grpc_client_with_anon_creds(
         transport = kwargs["transport"]
         assert isinstance(transport._credentials, AnonymousCredentials)
 
+    def test_grpc_client_with_anon_creds_no_client_options(self):
+        # Act & Assert
+        message = "Either client_options or `client_option.api_endpoint` is None. Please provide api_endpoint when `AnonymousCredentials` is used "
+        with pytest.raises(ValueError, match=message):
+            async_grpc_client.AsyncGrpcClient(
+                credentials=AnonymousCredentials(),
+            )
+
+    def test_grpc_client_with_anon_creds_empty_client_options(self):
+        # Act & Assert
+        message = "Either client_options or `client_option.api_endpoint` is None. Please provide api_endpoint when `AnonymousCredentials` is used "
+        with pytest.raises(ValueError, match=message):
+            async_grpc_client.AsyncGrpcClient(
+                client_options=client_options.ClientOptions(),
+                credentials=AnonymousCredentials(),
+            )
+
     @mock.patch("google.cloud._storage_v2.StorageAsyncClient")
     def test_user_agent_with_custom_client_info(self, mock_async_storage_client):
         """Test that gcloud-python user agent is appended to existing user agent.
