code: def test_multiplexed_e2e(serve_instance):
      """Test multiplexed function end to end"""
  
      @serve.deployment(num_replicas=2)
      class Model:
          @serve.multiplexed(max_num_models_per_replica=1)
          async def get_model(self, tag):
              return tag
  
          async def __call__(self, request):
              tag = serve.get_multiplexed_model_id()
              await self.get_model(tag)
              # return pid to check if the same model is used
              return os.getpid()
  
      model_id = "1"
      handle = serve.run(Model.bind())
      headers = {SERVE_MULTIPLEXED_MODEL_ID: model_id}
      resp = requests.get("http://localhost:8000", headers=headers)
      initial_pid = resp.json()
  
      wait_for_condition(
          check_model_id_in_replicas, handle=handle, model_id=model_id
      )
  
      # Check that the same replica is used repeatedly for the same model_id.
      for _ in range(10):
          resp = requests.get("http://localhost:8000", headers=headers)
          assert resp.json() == initial_pid
  
      for _ in range(10):
          assert (
              handle.options(multiplexed_model_id="1")
              .remote("blabla")
              .result()
              == initial_pid
          )
  
imports: import asyncio
  import os
  from typing import List
  import pytest
  import requests
  import ray
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray._private.utils import get_or_create_event_loop
  from ray.serve._private.constants import SERVE_MULTIPLEXED_MODEL_ID
  from ray.serve.context import _get_internal_replica_context
  from ray.serve.handle import RayServeHandle
  from ray.serve.multiplex import _ModelMultiplexWrapper
label_desc: Use image node_id 5 for calling a global function with example usage: test_multiplexed_e2e(serve_instance) and returns: tag, os