code: def test_controller_recover_target_capacity(
      shutdown_ray_and_serve, client: ServeControllerClient
  ):
      config = ServeDeploySchema(
          target_capacity=100,
          applications=[
              ServeApplicationSchema(
                  import_path="ray.serve.tests.test_target_capacity:test_app"
              )
          ],
      )
  
      client.deploy_apps(config)
      wait_for_condition(
          lambda: serve.status().target_capacity == 100.0
      )
      assert (
          ray.get(
              client._controller._get_target_capacity_direction.remote()
          )
          == TargetCapacityDirection.UP
      )
  
      # Scale down to target_capacity 50, both deployments should be at half scale.
      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={
              INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS
              / 2,
              DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS
              / 2,
          },
      )
      assert (
          ray.get(
              client._controller._get_target_capacity_direction.remote()
          )
          == TargetCapacityDirection.DOWN
      )
  
      ray.kill(client._controller, no_restart=False)
  
      # Verify that the target_capacity is recovered after the controller comes back.
      wait_for_condition(lambda: serve.status().target_capacity == 50.0)
      wait_for_condition(
          check_expected_num_replicas,
          deployment_to_num_replicas={
              INGRESS_DEPLOYMENT_NAME: INGRESS_DEPLOYMENT_NUM_REPLICAS
              / 2,
              DOWNSTREAM_DEPLOYMENT_NAME: DOWNSTREAM_DEPLOYMENT_NUM_REPLICAS
              / 2,
          },
      )
      assert (
          ray.get(
              client._controller._get_target_capacity_direction.remote()
          )
          == TargetCapacityDirection.DOWN
      )
  
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 6 for calling a global function with example usage: test_controller_recover_target_capacity(shutdown_ray_and_serve, client) without return types