code: def test_serving_request_through_grpc_proxy(ray_cluster):
      """Test serving request through gRPC proxy.
  
      When Serve runs with a gRPC deployment, the app should be deployed successfully,
      both ListApplications and Healthz methods returning successful responses, and
      registered gRPC methods are routing to the correct replica and return the correct
      response.
      """
      cluster = ray_cluster
      cluster.add_node(num_cpus=2)
      cluster.connect(namespace=SERVE_NAMESPACE)
  
      grpc_port = 9000
      grpc_servicer_functions = [
          "ray.serve.generated.serve_pb2_grpc.add_UserDefinedServiceServicer_to_server",
          "ray.serve.generated.serve_pb2_grpc.add_FruitServiceServicer_to_server",
      ]
  
      serve.start(
          grpc_options=gRPCOptions(
              port=grpc_port,
              grpc_servicer_functions=grpc_servicer_functions,
          ),
      )
      replicas = ray.get(
          serve.context._global_client._controller._all_running_replicas.remote()
      )
  
      # Ensures the app is not yet deployed.
      app_name = "default"
      deployment_name = "grpc-deployment"
      replica_name = DeploymentID(deployment_name, app_name)
      assert replica_name not in replicas
  
      channel = grpc.insecure_channel("localhost:9000")
  
      # Ensures the not found is responding correctly.
      ping_grpc_call_method(channel, app_name, test_not_found=True)
  
      serve.run(target=g)
      replicas = ray.get(
          serve.context._global_client._controller._all_running_replicas.remote()
      )
  
      # Ensures the app is deployed.
      assert len(replicas[replica_name]) == 1
  
      # Ensures ListApplications method succeeding.
      ping_grpc_list_applications(channel, [app_name])
  
      # Ensures Healthz method succeeding.
      ping_grpc_healthz(channel)
  
      # Ensures a custom defined method is responding correctly.
      ping_grpc_call_method(channel, app_name)
  
      # Ensures another custom defined method is responding correctly.
      ping_grpc_another_method(channel, app_name)
  
      # Ensures model multiplexing is responding correctly.
      ping_grpc_model_multiplexing(channel, app_name)
  
      # Ensure Streaming method is responding correctly.
      ping_grpc_streaming(channel, app_name)
  
      serve.run(target=g2)
      replicas = ray.get(
          serve.context._global_client._controller._all_running_replicas.remote()
      )
  
      # Ensures the app is deployed.
      deployment_name = "grpc-deployment-model-composition"
      replica_name = DeploymentID(deployment_name, app_name)
      assert len(replicas[replica_name]) == 1
  
      # Ensure model composition is responding correctly.
      ping_fruit_stand(channel, app_name)
  
imports: import os
  import sys
  from typing import Any
  import grpc
  import pytest
  import ray
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.cluster_utils import Cluster
  from ray.serve._private.common import DeploymentID
  from ray.serve._private.constants import SERVE_NAMESPACE
  from ray.serve._private.test_utils import ping_fruit_stand, ping_grpc_another_method, ping_grpc_call_method, ping_grpc_healthz, ping_grpc_list_applications, ping_grpc_model_multiplexing, ping_grpc_streaming, send_signal_on_cancellation
  from ray.serve.config import gRPCOptions
  from ray.serve.generated import serve_pb2, serve_pb2_grpc
  from ray.serve.grpc_util import RayServegRPCContext
  from ray.serve.tests.test_config_files.grpc_deployment import g, g2
label_desc: Use image node_id 1 for calling a global function with example usage: test_serving_request_through_grpc_proxy(ray_cluster) without return types