code: def test_spread_deployment_scheduling_policy_downscale_single_deployment(
      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)
      dep_id = DeploymentID("deployment1", "my_app")
      scheduler.on_deployment_created(
          dep_id, SpreadDeploymentSchedulingPolicy()
      )
      scheduler.on_replica_running(dep_id, "replica1", "node1")
      scheduler.on_replica_running(dep_id, "replica2", "node1")
      scheduler.on_replica_running(dep_id, "replica3", "node2")
      scheduler.on_replica_recovering(dep_id, "replica4")
      deployment_to_replicas_to_stop = scheduler.schedule(
          upscales={},
          downscales={
              dep_id: DeploymentDownscaleRequest(
                  deployment_id=dep_id, num_to_stop=1
              )
          },
      )
      assert len(deployment_to_replicas_to_stop) == 1
      # Prefer replica without node id
      assert deployment_to_replicas_to_stop[dep_id] == {"replica4"}
      scheduler.on_replica_stopping(dep_id, "replica4")
  
      deployment_to_replicas_to_stop = scheduler.schedule(
          upscales={
              dep_id: [
                  ReplicaSchedulingRequest(
                      deployment_id=dep_id,
                      replica_name="replica5",
                      actor_def=Replica,
                      actor_resources={"CPU": 1},
                      actor_options={},
                      actor_init_args=(),
                      on_scheduled=lambda actor_handle, placement_group: actor_handle,
                  ),
              ]
          },
          downscales={},
      )
      assert not deployment_to_replicas_to_stop
      deployment_to_replicas_to_stop = scheduler.schedule(
          upscales={},
          downscales={
              dep_id: DeploymentDownscaleRequest(
                  deployment_id=dep_id, num_to_stop=1
              )
          },
      )
      assert len(deployment_to_replicas_to_stop) == 1
      # Prefer replica without node id
      assert deployment_to_replicas_to_stop[dep_id] == {"replica5"}
      scheduler.on_replica_stopping(dep_id, "replica5")
  
      deployment_to_replicas_to_stop = scheduler.schedule(
          upscales={},
          downscales={
              dep_id: DeploymentDownscaleRequest(
                  deployment_id=dep_id, num_to_stop=1
              )
          },
      )
      assert len(deployment_to_replicas_to_stop) == 1
      # Prefer replica on a node with fewest replicas of all deployments.
      assert deployment_to_replicas_to_stop[dep_id] == {"replica3"}
      scheduler.on_replica_stopping(dep_id, "replica3")
  
      deployment_to_replicas_to_stop = scheduler.schedule(
          upscales={},
          downscales={
              dep_id: DeploymentDownscaleRequest(
                  deployment_id=dep_id, num_to_stop=2
              )
          },
      )
      assert len(deployment_to_replicas_to_stop) == 1
      assert deployment_to_replicas_to_stop[dep_id] == {
          "replica1",
          "replica2",
      }
      scheduler.on_replica_stopping(dep_id, "replica1")
      scheduler.on_replica_stopping(dep_id, "replica2")
      scheduler.on_deployment_deleted(dep_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 3 for calling a global function with example usage: test_spread_deployment_scheduling_policy_downscale_single_deployment(ray_start_cluster) without return types