diff --git a/google/cloud/container_v1beta1/services/cluster_manager/async_client.py b/google/cloud/container_v1beta1/services/cluster_manager/async_client.py index 861a2814..ba69a990 100644 --- a/google/cloud/container_v1beta1/services/cluster_manager/async_client.py +++ b/google/cloud/container_v1beta1/services/cluster_manager/async_client.py @@ -2492,6 +2492,92 @@ async def sample_get_server_config(): # Done; return the response. return response + async def get_json_web_keys( + self, + request: Optional[Union[cluster_service.GetJSONWebKeysRequest, dict]] = None, + *, + retry: OptionalRetry = gapic_v1.method.DEFAULT, + timeout: Union[float, object] = gapic_v1.method.DEFAULT, + metadata: Sequence[Tuple[str, str]] = (), + ) -> cluster_service.GetJSONWebKeysResponse: + r"""Gets the public component of the cluster signing keys + in JSON Web Key format. + This API is not yet intended for general use, and is not + available for all clusters. + + .. code-block:: python + + # This snippet has been automatically generated and should be regarded as a + # code template only. + # It will require modifications to work: + # - It may require correct/in-range values for request initialization. + # - It may require specifying regional endpoints when creating the service + # client as shown in: + # https://googleapis.dev/python/google-api-core/latest/client_options.html + from google.cloud import container_v1beta1 + + async def sample_get_json_web_keys(): + # Create a client + client = container_v1beta1.ClusterManagerAsyncClient() + + # Initialize request argument(s) + request = container_v1beta1.GetJSONWebKeysRequest( + ) + + # Make the request + response = await client.get_json_web_keys(request=request) + + # Handle the response + print(response) + + Args: + request (Optional[Union[google.cloud.container_v1beta1.types.GetJSONWebKeysRequest, dict]]): + The request object. GetJSONWebKeysRequest gets the + public component of the keys used by the cluster to sign + token requests. This will be the jwks_uri for the + discover document returned by getOpenIDConfig. See the + OpenID Connect Discovery 1.0 specification for details. + retry (google.api_core.retry.Retry): Designation of what errors, if any, + should be retried. + timeout (float): The timeout for this request. + metadata (Sequence[Tuple[str, str]]): Strings which should be + sent along with the request as metadata. + + Returns: + google.cloud.container_v1beta1.types.GetJSONWebKeysResponse: + GetJSONWebKeysResponse is a valid + JSON Web Key Set as specififed in rfc + 7517 + + """ + # Create or coerce a protobuf request object. + request = cluster_service.GetJSONWebKeysRequest(request) + + # Wrap the RPC method; this adds retry and timeout information, + # and friendly error handling. + rpc = gapic_v1.method_async.wrap_method( + self._client._transport.get_json_web_keys, + default_timeout=None, + client_info=DEFAULT_CLIENT_INFO, + ) + + # Certain fields should be provided within the metadata header; + # add these here. + metadata = tuple(metadata) + ( + gapic_v1.routing_header.to_grpc_metadata((("parent", request.parent),)), + ) + + # Send the request. + response = await rpc( + request, + retry=retry, + timeout=timeout, + metadata=metadata, + ) + + # Done; return the response. + return response + async def list_node_pools( self, request: Optional[Union[cluster_service.ListNodePoolsRequest, dict]] = None, @@ -2633,92 +2719,6 @@ async def sample_list_node_pools(): # Done; return the response. return response - async def get_json_web_keys( - self, - request: Optional[Union[cluster_service.GetJSONWebKeysRequest, dict]] = None, - *, - retry: OptionalRetry = gapic_v1.method.DEFAULT, - timeout: Union[float, object] = gapic_v1.method.DEFAULT, - metadata: Sequence[Tuple[str, str]] = (), - ) -> cluster_service.GetJSONWebKeysResponse: - r"""Gets the public component of the cluster signing keys - in JSON Web Key format. - This API is not yet intended for general use, and is not - available for all clusters. - - .. code-block:: python - - # This snippet has been automatically generated and should be regarded as a - # code template only. - # It will require modifications to work: - # - It may require correct/in-range values for request initialization. - # - It may require specifying regional endpoints when creating the service - # client as shown in: - # https://googleapis.dev/python/google-api-core/latest/client_options.html - from google.cloud import container_v1beta1 - - async def sample_get_json_web_keys(): - # Create a client - client = container_v1beta1.ClusterManagerAsyncClient() - - # Initialize request argument(s) - request = container_v1beta1.GetJSONWebKeysRequest( - ) - - # Make the request - response = await client.get_json_web_keys(request=request) - - # Handle the response - print(response) - - Args: - request (Optional[Union[google.cloud.container_v1beta1.types.GetJSONWebKeysRequest, dict]]): - The request object. GetJSONWebKeysRequest gets the - public component of the keys used by the cluster to sign - token requests. This will be the jwks_uri for the - discover document returned by getOpenIDConfig. See the - OpenID Connect Discovery 1.0 specification for details. - retry (google.api_core.retry.Retry): Designation of what errors, if any, - should be retried. - timeout (float): The timeout for this request. - metadata (Sequence[Tuple[str, str]]): Strings which should be - sent along with the request as metadata. - - Returns: - google.cloud.container_v1beta1.types.GetJSONWebKeysResponse: - GetJSONWebKeysResponse is a valid - JSON Web Key Set as specififed in rfc - 7517 - - """ - # Create or coerce a protobuf request object. - request = cluster_service.GetJSONWebKeysRequest(request) - - # Wrap the RPC method; this adds retry and timeout information, - # and friendly error handling. - rpc = gapic_v1.method_async.wrap_method( - self._client._transport.get_json_web_keys, - default_timeout=None, - client_info=DEFAULT_CLIENT_INFO, - ) - - # Certain fields should be provided within the metadata header; - # add these here. - metadata = tuple(metadata) + ( - gapic_v1.routing_header.to_grpc_metadata((("parent", request.parent),)), - ) - - # Send the request. - response = await rpc( - request, - retry=retry, - timeout=timeout, - metadata=metadata, - ) - - # Done; return the response. - return response - async def get_node_pool( self, request: Optional[Union[cluster_service.GetNodePoolRequest, dict]] = None, diff --git a/google/cloud/container_v1beta1/services/cluster_manager/client.py b/google/cloud/container_v1beta1/services/cluster_manager/client.py index 10649aef..f0a8c26e 100644 --- a/google/cloud/container_v1beta1/services/cluster_manager/client.py +++ b/google/cloud/container_v1beta1/services/cluster_manager/client.py @@ -2659,6 +2659,93 @@ def sample_get_server_config(): # Done; return the response. return response + def get_json_web_keys( + self, + request: Optional[Union[cluster_service.GetJSONWebKeysRequest, dict]] = None, + *, + retry: OptionalRetry = gapic_v1.method.DEFAULT, + timeout: Union[float, object] = gapic_v1.method.DEFAULT, + metadata: Sequence[Tuple[str, str]] = (), + ) -> cluster_service.GetJSONWebKeysResponse: + r"""Gets the public component of the cluster signing keys + in JSON Web Key format. + This API is not yet intended for general use, and is not + available for all clusters. + + .. code-block:: python + + # This snippet has been automatically generated and should be regarded as a + # code template only. + # It will require modifications to work: + # - It may require correct/in-range values for request initialization. + # - It may require specifying regional endpoints when creating the service + # client as shown in: + # https://googleapis.dev/python/google-api-core/latest/client_options.html + from google.cloud import container_v1beta1 + + def sample_get_json_web_keys(): + # Create a client + client = container_v1beta1.ClusterManagerClient() + + # Initialize request argument(s) + request = container_v1beta1.GetJSONWebKeysRequest( + ) + + # Make the request + response = client.get_json_web_keys(request=request) + + # Handle the response + print(response) + + Args: + request (Union[google.cloud.container_v1beta1.types.GetJSONWebKeysRequest, dict]): + The request object. GetJSONWebKeysRequest gets the + public component of the keys used by the cluster to sign + token requests. This will be the jwks_uri for the + discover document returned by getOpenIDConfig. See the + OpenID Connect Discovery 1.0 specification for details. + retry (google.api_core.retry.Retry): Designation of what errors, if any, + should be retried. + timeout (float): The timeout for this request. + metadata (Sequence[Tuple[str, str]]): Strings which should be + sent along with the request as metadata. + + Returns: + google.cloud.container_v1beta1.types.GetJSONWebKeysResponse: + GetJSONWebKeysResponse is a valid + JSON Web Key Set as specififed in rfc + 7517 + + """ + # Create or coerce a protobuf request object. + # Minor optimization to avoid making a copy if the user passes + # in a cluster_service.GetJSONWebKeysRequest. + # There's no risk of modifying the input as we've already verified + # there are no flattened fields. + if not isinstance(request, cluster_service.GetJSONWebKeysRequest): + request = cluster_service.GetJSONWebKeysRequest(request) + + # Wrap the RPC method; this adds retry and timeout information, + # and friendly error handling. + rpc = self._transport._wrapped_methods[self._transport.get_json_web_keys] + + # Certain fields should be provided within the metadata header; + # add these here. + metadata = tuple(metadata) + ( + gapic_v1.routing_header.to_grpc_metadata((("parent", request.parent),)), + ) + + # Send the request. + response = rpc( + request, + retry=retry, + timeout=timeout, + metadata=metadata, + ) + + # Done; return the response. + return response + def list_node_pools( self, request: Optional[Union[cluster_service.ListNodePoolsRequest, dict]] = None, @@ -2790,93 +2877,6 @@ def sample_list_node_pools(): # Done; return the response. return response - def get_json_web_keys( - self, - request: Optional[Union[cluster_service.GetJSONWebKeysRequest, dict]] = None, - *, - retry: OptionalRetry = gapic_v1.method.DEFAULT, - timeout: Union[float, object] = gapic_v1.method.DEFAULT, - metadata: Sequence[Tuple[str, str]] = (), - ) -> cluster_service.GetJSONWebKeysResponse: - r"""Gets the public component of the cluster signing keys - in JSON Web Key format. - This API is not yet intended for general use, and is not - available for all clusters. - - .. code-block:: python - - # This snippet has been automatically generated and should be regarded as a - # code template only. - # It will require modifications to work: - # - It may require correct/in-range values for request initialization. - # - It may require specifying regional endpoints when creating the service - # client as shown in: - # https://googleapis.dev/python/google-api-core/latest/client_options.html - from google.cloud import container_v1beta1 - - def sample_get_json_web_keys(): - # Create a client - client = container_v1beta1.ClusterManagerClient() - - # Initialize request argument(s) - request = container_v1beta1.GetJSONWebKeysRequest( - ) - - # Make the request - response = client.get_json_web_keys(request=request) - - # Handle the response - print(response) - - Args: - request (Union[google.cloud.container_v1beta1.types.GetJSONWebKeysRequest, dict]): - The request object. GetJSONWebKeysRequest gets the - public component of the keys used by the cluster to sign - token requests. This will be the jwks_uri for the - discover document returned by getOpenIDConfig. See the - OpenID Connect Discovery 1.0 specification for details. - retry (google.api_core.retry.Retry): Designation of what errors, if any, - should be retried. - timeout (float): The timeout for this request. - metadata (Sequence[Tuple[str, str]]): Strings which should be - sent along with the request as metadata. - - Returns: - google.cloud.container_v1beta1.types.GetJSONWebKeysResponse: - GetJSONWebKeysResponse is a valid - JSON Web Key Set as specififed in rfc - 7517 - - """ - # Create or coerce a protobuf request object. - # Minor optimization to avoid making a copy if the user passes - # in a cluster_service.GetJSONWebKeysRequest. - # There's no risk of modifying the input as we've already verified - # there are no flattened fields. - if not isinstance(request, cluster_service.GetJSONWebKeysRequest): - request = cluster_service.GetJSONWebKeysRequest(request) - - # Wrap the RPC method; this adds retry and timeout information, - # and friendly error handling. - rpc = self._transport._wrapped_methods[self._transport.get_json_web_keys] - - # Certain fields should be provided within the metadata header; - # add these here. - metadata = tuple(metadata) + ( - gapic_v1.routing_header.to_grpc_metadata((("parent", request.parent),)), - ) - - # Send the request. - response = rpc( - request, - retry=retry, - timeout=timeout, - metadata=metadata, - ) - - # Done; return the response. - return response - def get_node_pool( self, request: Optional[Union[cluster_service.GetNodePoolRequest, dict]] = None, diff --git a/google/cloud/container_v1beta1/services/cluster_manager/transports/base.py b/google/cloud/container_v1beta1/services/cluster_manager/transports/base.py index 2b3b65c2..f705f455 100644 --- a/google/cloud/container_v1beta1/services/cluster_manager/transports/base.py +++ b/google/cloud/container_v1beta1/services/cluster_manager/transports/base.py @@ -268,6 +268,11 @@ def _prep_wrapped_messages(self, client_info): default_timeout=20.0, client_info=client_info, ), + self.get_json_web_keys: gapic_v1.method.wrap_method( + self.get_json_web_keys, + default_timeout=None, + client_info=client_info, + ), self.list_node_pools: gapic_v1.method.wrap_method( self.list_node_pools, default_retry=retries.Retry( @@ -283,11 +288,6 @@ def _prep_wrapped_messages(self, client_info): default_timeout=20.0, client_info=client_info, ), - self.get_json_web_keys: gapic_v1.method.wrap_method( - self.get_json_web_keys, - default_timeout=None, - client_info=client_info, - ), self.get_node_pool: gapic_v1.method.wrap_method( self.get_node_pool, default_retry=retries.Retry( @@ -574,25 +574,25 @@ def get_server_config( raise NotImplementedError() @property - def list_node_pools( + def get_json_web_keys( self, ) -> Callable[ - [cluster_service.ListNodePoolsRequest], + [cluster_service.GetJSONWebKeysRequest], Union[ - cluster_service.ListNodePoolsResponse, - Awaitable[cluster_service.ListNodePoolsResponse], + cluster_service.GetJSONWebKeysResponse, + Awaitable[cluster_service.GetJSONWebKeysResponse], ], ]: raise NotImplementedError() @property - def get_json_web_keys( + def list_node_pools( self, ) -> Callable[ - [cluster_service.GetJSONWebKeysRequest], + [cluster_service.ListNodePoolsRequest], Union[ - cluster_service.GetJSONWebKeysResponse, - Awaitable[cluster_service.GetJSONWebKeysResponse], + cluster_service.ListNodePoolsResponse, + Awaitable[cluster_service.ListNodePoolsResponse], ], ]: raise NotImplementedError() diff --git a/google/cloud/container_v1beta1/services/cluster_manager/transports/grpc.py b/google/cloud/container_v1beta1/services/cluster_manager/transports/grpc.py index a6657c5c..992ef6af 100644 --- a/google/cloud/container_v1beta1/services/cluster_manager/transports/grpc.py +++ b/google/cloud/container_v1beta1/services/cluster_manager/transports/grpc.py @@ -715,18 +715,21 @@ def get_server_config( return self._stubs["get_server_config"] @property - def list_node_pools( + def get_json_web_keys( self, ) -> Callable[ - [cluster_service.ListNodePoolsRequest], cluster_service.ListNodePoolsResponse + [cluster_service.GetJSONWebKeysRequest], cluster_service.GetJSONWebKeysResponse ]: - r"""Return a callable for the list node pools method over gRPC. + r"""Return a callable for the get json web keys method over gRPC. - Lists the node pools for a cluster. + Gets the public component of the cluster signing keys + in JSON Web Key format. + This API is not yet intended for general use, and is not + available for all clusters. Returns: - Callable[[~.ListNodePoolsRequest], - ~.ListNodePoolsResponse]: + Callable[[~.GetJSONWebKeysRequest], + ~.GetJSONWebKeysResponse]: A function that, when called, will call the underlying RPC on the server. """ @@ -734,30 +737,27 @@ def list_node_pools( # the request. # gRPC handles serialization and deserialization, so we just need # to pass in the functions for each. - if "list_node_pools" not in self._stubs: - self._stubs["list_node_pools"] = self.grpc_channel.unary_unary( - "/google.container.v1beta1.ClusterManager/ListNodePools", - request_serializer=cluster_service.ListNodePoolsRequest.serialize, - response_deserializer=cluster_service.ListNodePoolsResponse.deserialize, + if "get_json_web_keys" not in self._stubs: + self._stubs["get_json_web_keys"] = self.grpc_channel.unary_unary( + "/google.container.v1beta1.ClusterManager/GetJSONWebKeys", + request_serializer=cluster_service.GetJSONWebKeysRequest.serialize, + response_deserializer=cluster_service.GetJSONWebKeysResponse.deserialize, ) - return self._stubs["list_node_pools"] + return self._stubs["get_json_web_keys"] @property - def get_json_web_keys( + def list_node_pools( self, ) -> Callable[ - [cluster_service.GetJSONWebKeysRequest], cluster_service.GetJSONWebKeysResponse + [cluster_service.ListNodePoolsRequest], cluster_service.ListNodePoolsResponse ]: - r"""Return a callable for the get json web keys method over gRPC. + r"""Return a callable for the list node pools method over gRPC. - Gets the public component of the cluster signing keys - in JSON Web Key format. - This API is not yet intended for general use, and is not - available for all clusters. + Lists the node pools for a cluster. Returns: - Callable[[~.GetJSONWebKeysRequest], - ~.GetJSONWebKeysResponse]: + Callable[[~.ListNodePoolsRequest], + ~.ListNodePoolsResponse]: A function that, when called, will call the underlying RPC on the server. """ @@ -765,13 +765,13 @@ def get_json_web_keys( # the request. # gRPC handles serialization and deserialization, so we just need # to pass in the functions for each. - if "get_json_web_keys" not in self._stubs: - self._stubs["get_json_web_keys"] = self.grpc_channel.unary_unary( - "/google.container.v1beta1.ClusterManager/GetJSONWebKeys", - request_serializer=cluster_service.GetJSONWebKeysRequest.serialize, - response_deserializer=cluster_service.GetJSONWebKeysResponse.deserialize, + if "list_node_pools" not in self._stubs: + self._stubs["list_node_pools"] = self.grpc_channel.unary_unary( + "/google.container.v1beta1.ClusterManager/ListNodePools", + request_serializer=cluster_service.ListNodePoolsRequest.serialize, + response_deserializer=cluster_service.ListNodePoolsResponse.deserialize, ) - return self._stubs["get_json_web_keys"] + return self._stubs["list_node_pools"] @property def get_node_pool( diff --git a/google/cloud/container_v1beta1/services/cluster_manager/transports/grpc_asyncio.py b/google/cloud/container_v1beta1/services/cluster_manager/transports/grpc_asyncio.py index 2a4ce448..7613064d 100644 --- a/google/cloud/container_v1beta1/services/cluster_manager/transports/grpc_asyncio.py +++ b/google/cloud/container_v1beta1/services/cluster_manager/transports/grpc_asyncio.py @@ -742,35 +742,6 @@ def get_server_config( ) return self._stubs["get_server_config"] - @property - def list_node_pools( - self, - ) -> Callable[ - [cluster_service.ListNodePoolsRequest], - Awaitable[cluster_service.ListNodePoolsResponse], - ]: - r"""Return a callable for the list node pools method over gRPC. - - Lists the node pools for a cluster. - - Returns: - Callable[[~.ListNodePoolsRequest], - Awaitable[~.ListNodePoolsResponse]]: - A function that, when called, will call the underlying RPC - on the server. - """ - # Generate a "stub function" on-the-fly which will actually make - # the request. - # gRPC handles serialization and deserialization, so we just need - # to pass in the functions for each. - if "list_node_pools" not in self._stubs: - self._stubs["list_node_pools"] = self.grpc_channel.unary_unary( - "/google.container.v1beta1.ClusterManager/ListNodePools", - request_serializer=cluster_service.ListNodePoolsRequest.serialize, - response_deserializer=cluster_service.ListNodePoolsResponse.deserialize, - ) - return self._stubs["list_node_pools"] - @property def get_json_web_keys( self, @@ -803,6 +774,35 @@ def get_json_web_keys( ) return self._stubs["get_json_web_keys"] + @property + def list_node_pools( + self, + ) -> Callable[ + [cluster_service.ListNodePoolsRequest], + Awaitable[cluster_service.ListNodePoolsResponse], + ]: + r"""Return a callable for the list node pools method over gRPC. + + Lists the node pools for a cluster. + + Returns: + Callable[[~.ListNodePoolsRequest], + Awaitable[~.ListNodePoolsResponse]]: + A function that, when called, will call the underlying RPC + on the server. + """ + # Generate a "stub function" on-the-fly which will actually make + # the request. + # gRPC handles serialization and deserialization, so we just need + # to pass in the functions for each. + if "list_node_pools" not in self._stubs: + self._stubs["list_node_pools"] = self.grpc_channel.unary_unary( + "/google.container.v1beta1.ClusterManager/ListNodePools", + request_serializer=cluster_service.ListNodePoolsRequest.serialize, + response_deserializer=cluster_service.ListNodePoolsResponse.deserialize, + ) + return self._stubs["list_node_pools"] + @property def get_node_pool( self, diff --git a/tests/unit/gapic/container_v1beta1/test_cluster_manager.py b/tests/unit/gapic/container_v1beta1/test_cluster_manager.py index 9b550989..c299d152 100644 --- a/tests/unit/gapic/container_v1beta1/test_cluster_manager.py +++ b/tests/unit/gapic/container_v1beta1/test_cluster_manager.py @@ -5394,6 +5394,160 @@ async def test_get_server_config_flattened_error_async(): ) +@pytest.mark.parametrize( + "request_type", + [ + cluster_service.GetJSONWebKeysRequest, + dict, + ], +) +def test_get_json_web_keys(request_type, transport: str = "grpc"): + client = ClusterManagerClient( + credentials=ga_credentials.AnonymousCredentials(), + transport=transport, + ) + + # Everything is optional in proto3 as far as the runtime is concerned, + # and we are mocking out the actual API, so just send an empty request. + request = request_type() + + # Mock the actual call within the gRPC stub, and fake the request. + with mock.patch.object( + type(client.transport.get_json_web_keys), "__call__" + ) as call: + # Designate an appropriate return value for the call. + call.return_value = cluster_service.GetJSONWebKeysResponse() + response = client.get_json_web_keys(request) + + # Establish that the underlying gRPC stub method was called. + assert len(call.mock_calls) == 1 + _, args, _ = call.mock_calls[0] + assert args[0] == cluster_service.GetJSONWebKeysRequest() + + # Establish that the response is the type that we expect. + assert isinstance(response, cluster_service.GetJSONWebKeysResponse) + + +def test_get_json_web_keys_empty_call(): + # This test is a coverage failsafe to make sure that totally empty calls, + # i.e. request == None and no flattened fields passed, work. + client = ClusterManagerClient( + credentials=ga_credentials.AnonymousCredentials(), + transport="grpc", + ) + + # Mock the actual call within the gRPC stub, and fake the request. + with mock.patch.object( + type(client.transport.get_json_web_keys), "__call__" + ) as call: + client.get_json_web_keys() + call.assert_called() + _, args, _ = call.mock_calls[0] + assert args[0] == cluster_service.GetJSONWebKeysRequest() + + +@pytest.mark.asyncio +async def test_get_json_web_keys_async( + transport: str = "grpc_asyncio", request_type=cluster_service.GetJSONWebKeysRequest +): + client = ClusterManagerAsyncClient( + credentials=ga_credentials.AnonymousCredentials(), + transport=transport, + ) + + # Everything is optional in proto3 as far as the runtime is concerned, + # and we are mocking out the actual API, so just send an empty request. + request = request_type() + + # Mock the actual call within the gRPC stub, and fake the request. + with mock.patch.object( + type(client.transport.get_json_web_keys), "__call__" + ) as call: + # Designate an appropriate return value for the call. + call.return_value = grpc_helpers_async.FakeUnaryUnaryCall( + cluster_service.GetJSONWebKeysResponse() + ) + response = await client.get_json_web_keys(request) + + # Establish that the underlying gRPC stub method was called. + assert len(call.mock_calls) + _, args, _ = call.mock_calls[0] + assert args[0] == cluster_service.GetJSONWebKeysRequest() + + # Establish that the response is the type that we expect. + assert isinstance(response, cluster_service.GetJSONWebKeysResponse) + + +@pytest.mark.asyncio +async def test_get_json_web_keys_async_from_dict(): + await test_get_json_web_keys_async(request_type=dict) + + +def test_get_json_web_keys_field_headers(): + client = ClusterManagerClient( + credentials=ga_credentials.AnonymousCredentials(), + ) + + # Any value that is part of the HTTP/1.1 URI should be sent as + # a field header. Set these to a non-empty value. + request = cluster_service.GetJSONWebKeysRequest() + + request.parent = "parent_value" + + # Mock the actual call within the gRPC stub, and fake the request. + with mock.patch.object( + type(client.transport.get_json_web_keys), "__call__" + ) as call: + call.return_value = cluster_service.GetJSONWebKeysResponse() + client.get_json_web_keys(request) + + # Establish that the underlying gRPC stub method was called. + assert len(call.mock_calls) == 1 + _, args, _ = call.mock_calls[0] + assert args[0] == request + + # Establish that the field header was sent. + _, _, kw = call.mock_calls[0] + assert ( + "x-goog-request-params", + "parent=parent_value", + ) in kw["metadata"] + + +@pytest.mark.asyncio +async def test_get_json_web_keys_field_headers_async(): + client = ClusterManagerAsyncClient( + credentials=ga_credentials.AnonymousCredentials(), + ) + + # Any value that is part of the HTTP/1.1 URI should be sent as + # a field header. Set these to a non-empty value. + request = cluster_service.GetJSONWebKeysRequest() + + request.parent = "parent_value" + + # Mock the actual call within the gRPC stub, and fake the request. + with mock.patch.object( + type(client.transport.get_json_web_keys), "__call__" + ) as call: + call.return_value = grpc_helpers_async.FakeUnaryUnaryCall( + cluster_service.GetJSONWebKeysResponse() + ) + await client.get_json_web_keys(request) + + # Establish that the underlying gRPC stub method was called. + assert len(call.mock_calls) + _, args, _ = call.mock_calls[0] + assert args[0] == request + + # Establish that the field header was sent. + _, _, kw = call.mock_calls[0] + assert ( + "x-goog-request-params", + "parent=parent_value", + ) in kw["metadata"] + + @pytest.mark.parametrize( "request_type", [ @@ -5640,160 +5794,6 @@ async def test_list_node_pools_flattened_error_async(): ) -@pytest.mark.parametrize( - "request_type", - [ - cluster_service.GetJSONWebKeysRequest, - dict, - ], -) -def test_get_json_web_keys(request_type, transport: str = "grpc"): - client = ClusterManagerClient( - credentials=ga_credentials.AnonymousCredentials(), - transport=transport, - ) - - # Everything is optional in proto3 as far as the runtime is concerned, - # and we are mocking out the actual API, so just send an empty request. - request = request_type() - - # Mock the actual call within the gRPC stub, and fake the request. - with mock.patch.object( - type(client.transport.get_json_web_keys), "__call__" - ) as call: - # Designate an appropriate return value for the call. - call.return_value = cluster_service.GetJSONWebKeysResponse() - response = client.get_json_web_keys(request) - - # Establish that the underlying gRPC stub method was called. - assert len(call.mock_calls) == 1 - _, args, _ = call.mock_calls[0] - assert args[0] == cluster_service.GetJSONWebKeysRequest() - - # Establish that the response is the type that we expect. - assert isinstance(response, cluster_service.GetJSONWebKeysResponse) - - -def test_get_json_web_keys_empty_call(): - # This test is a coverage failsafe to make sure that totally empty calls, - # i.e. request == None and no flattened fields passed, work. - client = ClusterManagerClient( - credentials=ga_credentials.AnonymousCredentials(), - transport="grpc", - ) - - # Mock the actual call within the gRPC stub, and fake the request. - with mock.patch.object( - type(client.transport.get_json_web_keys), "__call__" - ) as call: - client.get_json_web_keys() - call.assert_called() - _, args, _ = call.mock_calls[0] - assert args[0] == cluster_service.GetJSONWebKeysRequest() - - -@pytest.mark.asyncio -async def test_get_json_web_keys_async( - transport: str = "grpc_asyncio", request_type=cluster_service.GetJSONWebKeysRequest -): - client = ClusterManagerAsyncClient( - credentials=ga_credentials.AnonymousCredentials(), - transport=transport, - ) - - # Everything is optional in proto3 as far as the runtime is concerned, - # and we are mocking out the actual API, so just send an empty request. - request = request_type() - - # Mock the actual call within the gRPC stub, and fake the request. - with mock.patch.object( - type(client.transport.get_json_web_keys), "__call__" - ) as call: - # Designate an appropriate return value for the call. - call.return_value = grpc_helpers_async.FakeUnaryUnaryCall( - cluster_service.GetJSONWebKeysResponse() - ) - response = await client.get_json_web_keys(request) - - # Establish that the underlying gRPC stub method was called. - assert len(call.mock_calls) - _, args, _ = call.mock_calls[0] - assert args[0] == cluster_service.GetJSONWebKeysRequest() - - # Establish that the response is the type that we expect. - assert isinstance(response, cluster_service.GetJSONWebKeysResponse) - - -@pytest.mark.asyncio -async def test_get_json_web_keys_async_from_dict(): - await test_get_json_web_keys_async(request_type=dict) - - -def test_get_json_web_keys_field_headers(): - client = ClusterManagerClient( - credentials=ga_credentials.AnonymousCredentials(), - ) - - # Any value that is part of the HTTP/1.1 URI should be sent as - # a field header. Set these to a non-empty value. - request = cluster_service.GetJSONWebKeysRequest() - - request.parent = "parent_value" - - # Mock the actual call within the gRPC stub, and fake the request. - with mock.patch.object( - type(client.transport.get_json_web_keys), "__call__" - ) as call: - call.return_value = cluster_service.GetJSONWebKeysResponse() - client.get_json_web_keys(request) - - # Establish that the underlying gRPC stub method was called. - assert len(call.mock_calls) == 1 - _, args, _ = call.mock_calls[0] - assert args[0] == request - - # Establish that the field header was sent. - _, _, kw = call.mock_calls[0] - assert ( - "x-goog-request-params", - "parent=parent_value", - ) in kw["metadata"] - - -@pytest.mark.asyncio -async def test_get_json_web_keys_field_headers_async(): - client = ClusterManagerAsyncClient( - credentials=ga_credentials.AnonymousCredentials(), - ) - - # Any value that is part of the HTTP/1.1 URI should be sent as - # a field header. Set these to a non-empty value. - request = cluster_service.GetJSONWebKeysRequest() - - request.parent = "parent_value" - - # Mock the actual call within the gRPC stub, and fake the request. - with mock.patch.object( - type(client.transport.get_json_web_keys), "__call__" - ) as call: - call.return_value = grpc_helpers_async.FakeUnaryUnaryCall( - cluster_service.GetJSONWebKeysResponse() - ) - await client.get_json_web_keys(request) - - # Establish that the underlying gRPC stub method was called. - assert len(call.mock_calls) - _, args, _ = call.mock_calls[0] - assert args[0] == request - - # Establish that the field header was sent. - _, _, kw = call.mock_calls[0] - assert ( - "x-goog-request-params", - "parent=parent_value", - ) in kw["metadata"] - - @pytest.mark.parametrize( "request_type", [ @@ -10430,8 +10430,8 @@ def test_cluster_manager_base_transport(): "get_operation", "cancel_operation", "get_server_config", - "list_node_pools", "get_json_web_keys", + "list_node_pools", "get_node_pool", "create_node_pool", "delete_node_pool",