code: def test_external_flow_with_grpc_metadata():
      import grpc
      from grpc_health.v1 import health_pb2, health_pb2_grpc
      from grpc_reflection.v1alpha import reflection
  
      from jina.proto import jina_pb2, jina_pb2_grpc
      from jina.serve.runtimes.gateway.grpc import GRPCGateway
  
      class DummyInterceptor(grpc.aio.ServerInterceptor):
          def __init__(self):
              def deny(_, context):
                  context.abort(
                      grpc.StatusCode.UNAUTHENTICATED, "Invalid key"
                  )
  
              self._deny = grpc.unary_unary_rpc_method_handler(deny)
  
          async def intercept_service(
              self, continuation, handler_call_details
          ):
              meta = handler_call_details.invocation_metadata
              method = handler_call_details.method
              if method != "/jina.JinaRPC/Call":
                  return await continuation(handler_call_details)
  
              for m in meta:
                  if m == ("authorization", "valid"):
                      return await continuation(handler_call_details)
  
              return self._deny
  
      class DummyGRPCGateway(GRPCGateway):
          def __int__(self, *args, **kwargs):
              super().__init__(*args, **kwargs)
  
          def grpc_extra_server_interceptors(self):
              return [DummyInterceptor()]
  
          async def setup_server(self):
              """
              setup GRPC server
              """
              server_interceptors = (
                  self.grpc_tracing_server_interceptors or []
              )
              extra_interceptors = self.grpc_extra_server_interceptors()
              if extra_interceptors:
                  server_interceptors.extend(extra_interceptors)
  
              self.server = grpc.aio.server(
                  options=get_server_side_grpc_options(
                      self.grpc_server_options
                  ),
                  interceptors=server_interceptors,
              )
  
              jina_pb2_grpc.add_JinaRPCServicer_to_server(
                  self.streamer._streamer, self.server
              )
  
              jina_pb2_grpc.add_JinaGatewayDryRunRPCServicer_to_server(
                  self._request_handler, self.server
              )
              jina_pb2_grpc.add_JinaInfoRPCServicer_to_server(
                  self._request_handler, self.server
              )
  
              service_names = (
                  jina_pb2.DESCRIPTOR.services_by_name[
                      "JinaRPC"
                  ].full_name,
                  jina_pb2.DESCRIPTOR.services_by_name[
                      "JinaGatewayDryRunRPC"
                  ].full_name,
                  jina_pb2.DESCRIPTOR.services_by_name[
                      "JinaInfoRPC"
                  ].full_name,
                  reflection.SERVICE_NAME,
              )
              # Mark all services as healthy.
              health_pb2_grpc.add_HealthServicer_to_server(
                  self.health_servicer, self.server
              )
  
              for service in service_names:
                  await self.health_servicer.set(
                      service, health_pb2.HealthCheckResponse.SERVING
                  )
              reflection.enable_server_reflection(
                  service_names, self.server
              )
  
              bind_addr = f"{self.host}:{self.port}"
  
              if self.ssl_keyfile and self.ssl_certfile:
                  with open(self.ssl_keyfile, "rb") as f:
                      private_key = f.read()
                  with open(self.ssl_certfile, "rb") as f:
                      certificate_chain = f.read()
  
                  server_credentials = grpc.ssl_server_credentials(
                      (
                          (
                              private_key,
                              certificate_chain,
                          ),
                      )
                  )
                  self.server.add_secure_port(
                      bind_addr, server_credentials
                  )
              elif (
                  self.ssl_keyfile != self.ssl_certfile
              ):  # if we have only ssl_keyfile and not ssl_certfile or vice versa
                  raise ValueError(
                      f"you can't pass a ssl_keyfile without a ssl_certfile and vice versa"
                  )
              else:
                  self.server.add_insecure_port(bind_addr)
              self.logger.debug(f"start server bound to {bind_addr}")
              await self.server.start()
  
      class ExtExecutor(Executor):
          @requests
          def foo(self, docs, **kwargs):
              for doc in docs:
                  doc.text = "external"
  
      external_flow = (
          Flow()
          .config_gateway(uses=DummyGRPCGateway)
          .add(uses=ExtExecutor)
      )
  
      with external_flow:
          d = Document(text="sunset with green landscape by the river")
          f = Flow().add(
              port=external_flow.port,
              external=True,
              name="external_executor",
              grpc_metadata={"authorization": "valid"},
          )
          with f:
              response = f.post(
                  on="/", inputs=d, target_executor="external_executor"
              )
  
              assert response[0].text == "external"
  
          f = Flow().add(
              port=external_flow.port,
              external=True,
              name="external_executor",
              grpc_metadata={"authorization": "invalid"},
          )
          with f:
              with pytest.raises(
                  jina.excepts.BadServerFlow
              ) as error_info:
                  response = f.post(
                      on="/",
                      inputs=d,
                      target_executor="external_executor",
                  )
              assert "Invalid key" in str(error_info.value)
  
imports: import uuid
  import pytest
  import jina.excepts
  from jina import Document, DocumentArray, Executor, Flow, requests
  from jina.helper import random_port
  from jina.orchestrate.deployments import Deployment
  from jina.parsers import set_deployment_parser
  from jina.serve.helper import get_server_side_grpc_options
label_desc: Use image node_id 21 for calling a global function with example usage: test_external_flow_with_grpc_metadata() and returns: self, list, unknown, unknown