code: def test_proxy_prefers_replicas_on_same_node(
      ray_cluster: Cluster, set_flag
  ):
      """When the feature flag is turned on via env var, verify that http proxy routes to
      replicas on the same node when possible. Otherwise if env var is not set, http proxy
      should route to all replicas equally.
      """
  
      if set_flag:
          os.environ["RAY_SERVE_PROXY_PREFER_LOCAL_NODE_ROUTING"] = "1"
  
      cluster = ray_cluster
      cluster.add_node(num_cpus=1)
      cluster.add_node(num_cpus=1)
  
      # Only start one HTTP proxy on the head node.
      serve.start(http_options={"location": "HeadOnly"})
      head_node_id = get_head_node_id()
  
      @serve.deployment(num_replicas=2, max_concurrent_queries=1)
      def f():
          return ray.get_runtime_context().get_node_id()
  
      # The deployment's two replicas will be spread across the two nodes
      serve.run(f.bind())
  
      # Since they're sent sequentially, all requests should be routed to
      # the replica on the head node
      responses = [
          requests.post("http://localhost:8000").text for _ in range(10)
      ]
      if set_flag:
          assert all(resp == head_node_id for resp in responses)
      else:
          assert len(set(responses)) == 2
  
      if "RAY_SERVE_PROXY_PREFER_LOCAL_NODE_ROUTING" in os.environ:
          del os.environ["RAY_SERVE_PROXY_PREFER_LOCAL_NODE_ROUTING"]
  
imports: import os
  import sys
  import time
  from collections import defaultdict
  import pytest
  import requests
  import ray
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.cluster_utils import Cluster
  from ray.exceptions import RayActorError
  from ray.serve._private.common import DeploymentID, ReplicaState
  from ray.serve._private.constants import SERVE_NAMESPACE
  from ray.serve._private.deployment_state import ReplicaStartupStatus
  from ray.serve._private.utils import get_head_node_id
  from ray.serve.context import _get_global_client
  from ray.serve.handle import RayServeHandle
  from ray.serve.schema import ServeDeploySchema
label_desc: Use image node_id 10 for calling a global function with example usage: test_proxy_prefers_replicas_on_same_node(ray_cluster, set_flag) and returns: ray