code: def test_grpc_proxy_timeouts(
      ray_instance, ray_shutdown, streaming: bool
  ):
      """Test gRPC request timed out.
  
      When the request timed out, gRPC proxy should return timeout response for both
      unary and streaming request.
      """
      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,
          ),
      )
  
      signal_actor = SignalActor.remote()
  
      @serve.deployment
      class HelloModel:
          def __call__(self, user_message):
              ray.get(signal_actor.wait.remote())
              return serve_pb2.UserDefinedResponse(greeting="hello")
  
          def Streaming(self, user_message):
              for i in range(10):
                  ray.get(signal_actor.wait.remote())
                  yield serve_pb2.UserDefinedResponse(greeting="hello")
  
      serve.run(HelloModel.bind())
  
      channel = grpc.insecure_channel("localhost:9000")
      stub = serve_pb2_grpc.UserDefinedServiceStub(channel)
      request = serve_pb2.UserDefinedMessage(
          name="foo", num=30, foo="bar"
      )
  
      timeout_response = "timed out after 0.1s."
  
      if streaming:
          # Ensure streaming request respond with timeout response when running longer
          # than the serve request timeout setting
          with pytest.raises(grpc.RpcError) as exception_info:
              list(stub.Streaming(request=request))
      else:
          # Ensure unary request respond with timeout response when running longer than
          # the serve request timeout setting
          with pytest.raises(grpc.RpcError) as exception_info:
              stub.__call__(request=request)
  
      rpc_error = exception_info.value
      assert rpc_error.code() == grpc.StatusCode.DEADLINE_EXCEEDED
      assert timeout_response in rpc_error.details()
  
      # Unblock the handlers to avoid graceful shutdown time.
      ray.get(signal_actor.send.remote())
  
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 6 for calling a global function with example usage: test_grpc_proxy_timeouts(ray_instance, ray_shutdown, streaming) and returns: serve_pb2