code: def test_autoscaling_scale_to_zero(
      shutdown_ray_and_serve, client: ServeControllerClient
  ):
      config = ServeDeploySchema(
          applications=[
              ServeApplicationSchema(
                  import_path="ray.serve.tests.test_target_capacity:scale_to_zero_app"
              )
          ]
      )
  
      # Initially deploy at target_capacity 0, should have 0 replicas.
      config.target_capacity = 0.0
      client.deploy_apps(config)
      wait_for_condition(lambda: serve.status().target_capacity == 0.0)
      wait_for_condition(
          check_expected_num_replicas,
          deployment_to_num_replicas={
              SCALE_TO_ZERO_DEPLOYMENT_NAME: 0,
          },
      )
  
      # Increase to target_capacity 1, should remain at 0 replicas initially.
      config.target_capacity = 1.0
      client.deploy_apps(config)
      wait_for_condition(lambda: serve.status().target_capacity == 1.0)
      wait_for_condition(
          check_expected_num_replicas,
          deployment_to_num_replicas={
              SCALE_TO_ZERO_DEPLOYMENT_NAME: 0,
          },
      )
  
      # Send a bunch of requests. Autoscaler will want to scale it up to max replicas,
      # but it should stay at one due to the target_capacity.
      handle = serve.get_app_handle(SERVE_DEFAULT_APP_NAME)
      responses = [
          handle.remote()
          for _ in range(5 * SCALE_TO_ZERO_DEPLOYMENT_MAX_REPLICAS)
      ]
      wait_for_condition(
          check_expected_num_replicas,
          deployment_to_num_replicas={
              SCALE_TO_ZERO_DEPLOYMENT_NAME: 1,
          },
      )
  
      # Increase to target_capacity 100, should scale all the way up.
      config.target_capacity = 100.0
      client.deploy_apps(config)
      wait_for_condition(
          lambda: serve.status().target_capacity == 100.0
      )
      wait_for_condition(
          check_expected_num_replicas,
          deployment_to_num_replicas={
              SCALE_TO_ZERO_DEPLOYMENT_NAME: SCALE_TO_ZERO_DEPLOYMENT_MAX_REPLICAS,
          },
      )
  
      # Decrease to target_capacity 50, should scale down.
      config.target_capacity = 50.0
      client.deploy_apps(config)
      wait_for_condition(lambda: serve.status().target_capacity == 50.0)
      wait_for_condition(
          check_expected_num_replicas,
          deployment_to_num_replicas={
              SCALE_TO_ZERO_DEPLOYMENT_NAME: SCALE_TO_ZERO_DEPLOYMENT_MAX_REPLICAS
              / 2,
          },
      )
  
      # TODO(edoakes): for some reason, the deployment does not actually scale down to
      # zero here, so skipping this part for now. Instead it repeatedly scales down and
      # then back up. Seems to have something to do with the handle-side queue metric.
      return
  
      # Cancel all of the requests, should scale down to zero.
      [r.cancel() for r in responses]
      wait_for_condition(
          check_expected_num_replicas,
          deployment_to_num_replicas={
              SCALE_TO_ZERO_DEPLOYMENT_NAME: 0,
          },
      )
  
imports: import asyncio
  import sys
  from copy import deepcopy
  from typing import Dict, Optional
  import pytest
  import ray
  from ray import serve
  from ray._private.pydantic_compat import BaseModel
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.exceptions import RayActorError
  from ray.serve import Application
  from ray.serve._private.client import ServeControllerClient
  from ray.serve._private.common import ApplicationStatus, DeploymentStatus, DeploymentStatusTrigger, ReplicaState, TargetCapacityDirection
  from ray.serve._private.constants import SERVE_DEFAULT_APP_NAME, SERVE_NAMESPACE
  from ray.serve.config import AutoscalingConfig
  from ray.serve.context import _get_global_client
  from ray.serve.schema import ServeApplicationSchema, ServeDeploySchema
label_desc: Use image node_id 7 for calling a global function with example usage: test_autoscaling_scale_to_zero(shutdown_ray_and_serve, client) without return types