code: def test_healthz_and_routes_on_head_and_worker_nodes(
      shutdown_ray, call_ray_stop_only  # noqa: F811
  ):
      """Test `/-/healthz` and `/-/routes` return the correct responses for head and
      worker nodes.
  
      When there are replicas on all nodes, `/-/routes` and `/-/routes` on all nodes
      should return 200. When there are no replicas on any nodes, `/-/routes` and
      `/-/routes` on the head node should continue to return 200. `/-/routes` and
      `/-/routes` on the worker node should start to return 503
      """
      # Setup worker http proxy to be pointing to port 8001. Head node http proxy will
      # continue to be pointing to the default port 8000.
      os.environ["TEST_WORKER_NODE_HTTP_PORT"] = "8001"
  
      # Setup a cluster with 2 nodes
      cluster = Cluster()
      cluster.add_node(num_cpus=0)
      cluster.add_node(num_cpus=2)
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
      serve.start(http_options={"location": "EveryNode"})
  
      # Deploy 2 replicas, both should be on the worker node.
      @serve.deployment(num_replicas=2)
      class HelloModel:
          def __call__(self):
              return "hello"
  
      model = HelloModel.bind()
      serve.run(target=model)
  
      # Ensure worker node has both replicas.
      def check_replicas_on_worker_nodes():
          _actors = ray._private.state.actors().values()
          replica_nodes = [
              a["Address"]["NodeID"]
              for a in _actors
              if a["ActorClassName"].startswith("ServeReplica")
          ]
          return len(set(replica_nodes)) == 1
  
      wait_for_condition(check_replicas_on_worker_nodes)
  
      # Ensure total actors of 2 proxies, 1 controller, and 2 replicas, and 2 nodes exist.
      wait_for_condition(lambda: len(ray._private.state.actors()) == 5)
      assert len(ray.nodes()) == 2
  
      # Ensure `/-/healthz` and `/-/routes` return 200 and expected responses
      # on both nodes.
      def check_request(
          url: str, expected_code: int, expected_text: str
      ):
          req = requests.get(url)
          return (
              req.status_code == expected_code
              and req.text == expected_text
          )
  
      wait_for_condition(
          condition_predictor=check_request,
          url="http://127.0.0.1:8000/-/healthz",
          expected_code=200,
          expected_text="success",
      )
      assert (
          requests.get("http://127.0.0.1:8000/-/routes").status_code
          == 200
      )
      assert (
          requests.get("http://127.0.0.1:8000/-/routes").text
          == '{"/":"default"}'
      )
      wait_for_condition(
          condition_predictor=check_request,
          url="http://127.0.0.1:8001/-/healthz",
          expected_code=200,
          expected_text="success",
      )
      assert (
          requests.get("http://127.0.0.1:8001/-/routes").status_code
          == 200
      )
      assert (
          requests.get("http://127.0.0.1:8001/-/routes").text
          == '{"/":"default"}'
      )
  
      # Delete the deployment should bring the active actors down to 3 and drop
      # replicas on all nodes.
      serve.delete(name=SERVE_DEFAULT_APP_NAME)
  
      def _check():
          _actors = ray._private.state.actors().values()
          return (
              len(
                  list(
                      filter(
                          lambda a: a["State"] == "ALIVE",
                          _actors,
                      )
                  )
              )
              == 3
          )
  
      wait_for_condition(_check)
  
      # Ensure head node `/-/healthz` and `/-/routes` continue to return 200 and expected
      # responses. Also, the worker node `/-/healthz` and `/-/routes` should return 503
      # and unavailable responses.
      wait_for_condition(
          condition_predictor=check_request,
          url="http://127.0.0.1:8000/-/healthz",
          expected_code=200,
          expected_text="success",
      )
      assert (
          requests.get("http://127.0.0.1:8000/-/routes").status_code
          == 200
      )
      assert requests.get("http://127.0.0.1:8000/-/routes").text == "{}"
      wait_for_condition(
          condition_predictor=check_request,
          url="http://127.0.0.1:8001/-/healthz",
          expected_code=503,
          expected_text="This node is being drained.",
      )
      assert (
          requests.get("http://127.0.0.1:8001/-/routes").status_code
          == 503
      )
      assert (
          requests.get("http://127.0.0.1:8001/-/routes").text
          == "This node is being drained."
      )
  
      # Clean up serve.
      serve.shutdown()
  
imports: import os
  import subprocess
  import sys
  from contextlib import contextmanager
  from tempfile import NamedTemporaryFile
  import pytest
  import requests
  import ray
  import ray._private.state
  import ray.actor
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.cluster_utils import AutoscalingCluster, Cluster
  from ray.exceptions import RayActorError
  from ray.serve._private.common import ProxyStatus
  from ray.serve._private.constants import SERVE_DEFAULT_APP_NAME
  from ray.serve._private.logging_utils import get_serve_logs_dir
  from ray.serve._private.utils import get_head_node_id
  from ray.serve.context import _get_global_client
  from ray.serve.schema import ServeInstanceDetails
  from ray.tests.conftest import call_ray_stop_only
label_desc: Use image node_id 10 for calling a global function with example usage: test_healthz_and_routes_on_head_and_worker_nodes(shutdown_ray, call_ray_stop_only) and returns: unknown, unknown, unknown, str