code: def test_spread_deployment_scheduling_policy_downscale_multiple_deployments(
      ray_start_cluster,
  ):
      """Test to make sure downscale prefers replicas without node id
      and then replicas on a node with fewest replicas of all deployments.
      """
      cluster = ray_start_cluster
      cluster.add_node(num_cpus=3)
      cluster.wait_for_nodes()
      ray.init(address=cluster.address)
  
      cluster_node_info_cache = create_cluster_node_info_cache(
          GcsClient(address=ray.get_runtime_context().gcs_address)
      )
      cluster_node_info_cache.update()
  
      scheduler = DefaultDeploymentScheduler(cluster_node_info_cache)
      d1_id = DeploymentID("deployment1", "default")
      d2_id = DeploymentID("deployment2", "default")
      scheduler.on_deployment_created(
          d1_id, SpreadDeploymentSchedulingPolicy()
      )
      scheduler.on_deployment_created(
          d2_id, SpreadDeploymentSchedulingPolicy()
      )
      scheduler.on_replica_running(d1_id, "replica1", "node1")
      scheduler.on_replica_running(d1_id, "replica2", "node2")
      scheduler.on_replica_running(d1_id, "replica3", "node2")
      scheduler.on_replica_running(d2_id, "replica1", "node1")
      scheduler.on_replica_running(d2_id, "replica2", "node2")
      scheduler.on_replica_running(d2_id, "replica3", "node1")
      scheduler.on_replica_running(d2_id, "replica4", "node1")
      deployment_to_replicas_to_stop = scheduler.schedule(
          upscales={},
          downscales={
              d1_id: DeploymentDownscaleRequest(
                  deployment_id=d1_id, num_to_stop=1
              )
          },
      )
      assert len(deployment_to_replicas_to_stop) == 1
      # Even though node1 has fewest replicas of deployment1
      # but it has more replicas of all deployments so
      # we should stop replicas from node2.
      assert len(deployment_to_replicas_to_stop[d1_id]) == 1
      assert deployment_to_replicas_to_stop[d1_id] < {
          "replica2",
          "replica3",
      }
  
      scheduler.on_replica_stopping(d1_id, "replica3")
      scheduler.on_replica_stopping(d2_id, "replica3")
      scheduler.on_replica_stopping(d2_id, "replica4")
  
      deployment_to_replicas_to_stop = scheduler.schedule(
          upscales={},
          downscales={
              d1_id: DeploymentDownscaleRequest(
                  deployment_id=d1_id, num_to_stop=1
              ),
              d2_id: DeploymentDownscaleRequest(
                  deployment_id=d2_id, num_to_stop=1
              ),
          },
      )
      assert len(deployment_to_replicas_to_stop) == 2
      # We should stop replicas from the same node.
      assert len(deployment_to_replicas_to_stop[d1_id]) == 1
      assert (
          deployment_to_replicas_to_stop[d1_id]
          == deployment_to_replicas_to_stop[d2_id]
      )
  
      scheduler.on_replica_stopping(d1_id, "replica1")
      scheduler.on_replica_stopping(d1_id, "replica2")
      scheduler.on_replica_stopping(d2_id, "replica1")
      scheduler.on_replica_stopping(d2_id, "replica2")
      scheduler.on_deployment_deleted(d1_id)
      scheduler.on_deployment_deleted(d2_id)
  
file_path: ray/python/ray/serve/tests/test_deployment_scheduler.py
imports: import sys
  import pytest
  import ray
  from ray._raylet import GcsClient
  from ray.serve._private.common import DeploymentID
  from ray.serve._private.default_impl import create_cluster_node_info_cache
  from ray.serve._private.deployment_scheduler import DefaultDeploymentScheduler, DeploymentDownscaleRequest, ReplicaSchedulingRequest, SpreadDeploymentSchedulingPolicy
  from ray.serve._private.utils import get_head_node_id
  from ray.tests.conftest import *
label_desc: Use image node_id 2 for calling a global function with example usage: test_spread_deployment_scheduling_policy_downscale_multiple_deployments(ray_start_cluster) without return types