docstr: Check how Serve's status updates when target_capacity changes.
docstr_tok: ["Check","how","Serve","'s","status","updates","when","target_capacity","changes","."]
code: def test_static_num_replicas_target_capacity_update(
      self, shutdown_ray_and_serve, client: ServeControllerClient
  ):
      """Check how Serve's status updates when target_capacity changes."""
  
      app_name = "controlled_app"
      deployment_name = "controlled"
      num_replicas = 20
  
      signal = SignalActor.options(
          name="lifecycle_signal", namespace=SERVE_NAMESPACE
      ).remote()
  
      config = ServeDeploySchema(
          applications=[
              ServeApplicationSchema(
                  name=app_name,
                  import_path=(
                      "ray.serve.tests.test_target_capacity:create_controlled_app"
                  ),
                  args={"num_replicas": num_replicas},
              )
          ]
      )
  
      # Initially deploy at target_capacity 0, and check status.
      self.apply_config_and_check_status(
          client,
          target_capacity=0.0,
          config=config,
          app_name=app_name,
          deployment_name=deployment_name,
          expected_app_status=ApplicationStatus.RUNNING,
          expected_deployment_status=DeploymentStatus.HEALTHY,
          expected_deployment_status_trigger=(
              DeploymentStatusTrigger.CONFIG_UPDATE_COMPLETED
          ),
      )
      self.check_num_replicas(0, app_name, deployment_name)
  
      # Increase the target_capacity, and check again.
      self.apply_config_and_check_status(
          client,
          target_capacity=50.0,
          config=config,
          app_name=app_name,
          deployment_name=deployment_name,
          expected_app_status=ApplicationStatus.DEPLOYING,
          expected_deployment_status=DeploymentStatus.UPSCALING,
          expected_deployment_status_trigger=(
              DeploymentStatusTrigger.CONFIG_UPDATE_STARTED
          ),
      )
      self.check_num_replicas(0, app_name, deployment_name)
      self.unblock_replica_creation_and_deletion(signal, app_name)
      self.check_num_replicas(
          int(0.5 * num_replicas), app_name, deployment_name
      )
  
      # Decrease the target_capacity, and check again.
      self.apply_config_and_check_status(
          client,
          target_capacity=10.0,
          config=config,
          app_name=app_name,
          deployment_name=deployment_name,
          expected_app_status=ApplicationStatus.DEPLOYING,
          expected_deployment_status=DeploymentStatus.DOWNSCALING,
          expected_deployment_status_trigger=(
              DeploymentStatusTrigger.CONFIG_UPDATE_STARTED
          ),
      )
      # DeploymentStateManager marks replicas as STOPPING once the deployment
      # starts downscaling.
      wait_for_condition(
          self.check_num_replicas,
          expected_num_replicas=int(0.1 * num_replicas),
          app_name=app_name,
          deployment_name=deployment_name,
          replica_state=ReplicaState.RUNNING,
          timeout=20,
      )
      self.check_num_replicas(
          int(0.4 * num_replicas),
          app_name,
          deployment_name,
          replica_state=ReplicaState.STOPPING,
      )
      self.unblock_replica_creation_and_deletion(signal, app_name)
      self.check_num_replicas(
          int(0.1 * num_replicas), app_name, deployment_name
      )
      self.check_num_replicas(
          0,
          app_name,
          deployment_name,
          replica_state=ReplicaState.STOPPING,
      )
  
      # Send a signal so all replicas shut down quickly when the test finishes.
      ray.get(signal.send.remote())
  
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 4 for calling the TestTargetCapacityUpdateAndServeStatus obj's underlying member method code with example usage: obj.test_static_num_replicas_target_capacity_update(shutdown_ray_and_serve, client) without return types