docstr: Check Serve's status when target_capacity changes while autoscaling.
docstr_tok: ["Check","Serve","'s","status","when","target_capacity","changes","while","autoscaling","."]
code: def test_autoscaling_target_capacity_update(
      self, shutdown_ray_and_serve, client: ServeControllerClient
  ):
      """Check Serve's status when target_capacity changes while autoscaling."""
  
      app_name = "controlled_app"
      deployment_name = "controlled"
      min_replicas = 10
      initial_replicas = 30
      max_replicas = 70
  
      lifecycle_signal = SignalActor.options(
          name="lifecycle_signal", namespace=SERVE_NAMESPACE
      ).remote()
      request_signal = SignalActor.options(
          name="request_signal", namespace=SERVE_NAMESPACE
      ).remote()
  
      config = ServeDeploySchema(
          applications=[
              ServeApplicationSchema(
                  name=app_name,
                  import_path=(
                      "ray.serve.tests.test_target_capacity:"
                      "create_autoscaling_controlled_app"
                  ),
                  args=dict(
                      min_replicas=min_replicas,
                      initial_replicas=initial_replicas,
                      max_replicas=max_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,
          timeout=20,
          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(
          lifecycle_signal, app_name
      )
      self.check_num_replicas(
          int(0.5 * initial_replicas), app_name, deployment_name
      )
  
      # Send requests and check that the application scales up.
      requests = []
      handle = serve.get_app_handle(app_name)
      for _ in range(max_replicas):
          requests.append(handle.remote())
      ray.get(lifecycle_signal.send.remote())
      wait_for_condition(
          self.check_num_replicas,
          expected_num_replicas=int(0.5 * max_replicas),
          app_name=app_name,
          deployment_name=deployment_name,
          timeout=20,
      )
  
      # Clear requests and check that application scales down.
      ray.get(request_signal.send.remote())
      results = [request.result() for request in requests]
      assert results == ["Hello world!"] * (max_replicas)
      wait_for_condition(
          self.check_num_replicas,
          expected_num_replicas=int(0.5 * initial_replicas),
          app_name=app_name,
          deployment_name=deployment_name,
          replica_state=ReplicaState.RUNNING,
          timeout=20,
          retry_interval_ms=1000,
      )
      wait_for_condition(
          self.check_num_replicas,
          expected_num_replicas=0,
          app_name=app_name,
          deployment_name=deployment_name,
          replica_state=ReplicaState.STOPPING,
          timeout=20,
          retry_interval_ms=1000,
      )
      ray.get(lifecycle_signal.send.remote(clear=True))
      ray.get(request_signal.send.remote(clear=True))
  
      # Decrease the target_capacity, and check that min_replicas is used
      # to create the lower bound.
      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
          ),
      )
      self.unblock_replica_creation_and_deletion(
          lifecycle_signal, app_name
      )
      self.check_num_replicas(
          int(0.1 * min_replicas),
          app_name,
          deployment_name,
          replica_state=ReplicaState.RUNNING,
      )
      self.check_num_replicas(
          0,
          app_name,
          deployment_name,
          replica_state=ReplicaState.STOPPING,
      )
  
      # Check that target_capacity * max_replicas is still the upper bound.
      requests = []
      handle = serve.get_app_handle(app_name)
      for _ in range(max_replicas):
          requests.append(handle.remote())
      ray.get(lifecycle_signal.send.remote())
      wait_for_condition(
          self.check_num_replicas,
          expected_num_replicas=int(0.1 * max_replicas),
          app_name=app_name,
          deployment_name=deployment_name,
          controller_handle=client._controller,
          timeout=25,
          retry_interval_ms=2000,
      )
  
      # Clear requests and check that application scales down to
      # target_capacity * min_replicas.
      ray.get(request_signal.send.remote())
      results = [request.result() for request in requests]
      assert results == ["Hello world!"] * (max_replicas)
  
      wait_for_condition(
          self.check_num_replicas,
          expected_num_replicas=int(0.1 * min_replicas),
          app_name=app_name,
          deployment_name=deployment_name,
          controller_handle=client._controller,
          timeout=25,
          retry_interval_ms=2000,
      )
      ray.get(lifecycle_signal.send.remote(clear=True))
      ray.get(request_signal.send.remote(clear=True))
  
      # Scaling up to 100% target_capacity should make Serve use
      # initial_replicas as lower bound.
      self.apply_config_and_check_status(
          client,
          target_capacity=100.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(
          int(0.1 * min_replicas), app_name, deployment_name
      )
      self.unblock_replica_creation_and_deletion(
          lifecycle_signal, app_name
      )
      self.check_num_replicas(
          initial_replicas, app_name, deployment_name
      )
  
      # Unsetting target_capacity should make Serve use
      # min_replicas as lower bound. The current number of
      # replicas is already a valid number, so the application stays
      # RUNNING.
      self.apply_config_and_check_status(
          client,
          target_capacity=None,
          config=config,
          app_name=app_name,
          deployment_name=deployment_name,
          expected_app_status=ApplicationStatus.RUNNING,
          expected_deployment_status=DeploymentStatus.DOWNSCALING,
          expected_deployment_status_trigger=(
              DeploymentStatusTrigger.AUTOSCALING
          ),
      )
      self.unblock_replica_creation_and_deletion(
          lifecycle_signal, app_name
      )
      wait_for_condition(
          self.check_num_replicas,
          expected_num_replicas=min_replicas,
          app_name=app_name,
          deployment_name=deployment_name,
      )
  
      # Send a signal so all replicas shut down quickly when the test finishes.
      ray.get(request_signal.send.remote())
      ray.get(lifecycle_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 5 for calling the TestTargetCapacityUpdateAndServeStatus obj's underlying member method code with example usage: obj.test_autoscaling_target_capacity_update(shutdown_ray_and_serve, client) without return types