code: def test_intelligent_scale_down(ray_cluster):
      cluster = ray_cluster
      # Head node
      cluster.add_node(num_cpus=0)
      cluster.connect(namespace=SERVE_NAMESPACE)
      cluster.add_node(num_cpus=2)
      cluster.add_node(num_cpus=2)
      serve.start()
      client = _get_global_client()
  
      def get_actor_distributions():
          actors = ray._private.state.actors()
          node_to_actors = defaultdict(list)
          for actor in actors.values():
              if "ServeReplica" not in actor["ActorClassName"]:
                  continue
              if actor["State"] != "ALIVE":
                  continue
              node_to_actors[actor["Address"]["NodeID"]].append(actor)
  
          return set(map(len, node_to_actors.values()))
  
      def check_app_running_with_replicas(num_replicas):
          status = serve.status().applications["default"]
          assert status.status == "RUNNING"
          assert (
              status.deployments["f"].replica_states["RUNNING"]
              == num_replicas
          )
          return True
  
      app_config = {
          "name": "default",
          "import_path": "ray.serve.tests.test_cluster.f_app",
          "deployments": [{"name": "f", "num_replicas": 3}],
      }
      client.deploy_apps(
          ServeDeploySchema(**{"applications": [app_config]})
      )
      wait_for_condition(
          check_app_running_with_replicas, num_replicas=3
      )
      assert get_actor_distributions() == {2, 1}
  
      app_config["deployments"][0]["num_replicas"] = 2
      client.deploy_apps(
          ServeDeploySchema(**{"applications": [app_config]})
      )
      wait_for_condition(
          check_app_running_with_replicas, num_replicas=2
      )
      assert get_actor_distributions() == {2}
  
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 7 for calling a global function with example usage: test_intelligent_scale_down(ray_cluster) and returns: set, True