docstr: Checks that the number of replicas are as expected.
  
  Args:
      expected_num_replicas: the expected number of replicas.
      app_name: the deployment's application name.
      deployment_name: the deployment's name.
      replica_state: only replicas in this state are counted.
      controller_handle: this is an optional argument. If provided, the
          controller handle is used to get the current autoscaling
          metrics and print them if the assertion fails.
docstr_tok: ["Checks","that","the","number","of","replicas","are","as","expected",".","Args",":","expected_num_replicas",":","the","expected","number","of","replicas",".","app_name",":","the","deployment","'s","application","name",".","deployment_name",":","the","deployment","'s","name",".","replica_state",":","only","replicas","in","this","state","are","counted",".","controller_handle",":","this","is","an","optional","argument",".","If","provided",",","the","controller","handle","is","used","to","get","the","current","autoscaling","metrics","and","print","them","if","the","assertion","fails","."]
code: def check_num_replicas(
      self,
      expected_num_replicas: int,
      app_name: str,
      deployment_name: str,
      replica_state: ReplicaState = ReplicaState.RUNNING,
      controller_handle=None,
  ) -> bool:
      """Checks that the number of replicas are as expected.
  
      Args:
          expected_num_replicas: the expected number of replicas.
          app_name: the deployment's application name.
          deployment_name: the deployment's name.
          replica_state: only replicas in this state are counted.
          controller_handle: this is an optional argument. If provided, the
              controller handle is used to get the current autoscaling
              metrics and print them if the assertion fails.
      """
      deployment = (
          serve.status()
          .applications[app_name]
          .deployments[deployment_name]
      )
      num_running_replicas = deployment.replica_states.get(
          replica_state, 0
      )
      if controller_handle is None:
          assert (
              num_running_replicas == expected_num_replicas
          ), f"{deployment}"
      else:
          autoscaling_metrics = ray.get(
              controller_handle._dump_autoscaling_metrics_for_testing.remote()
          )
          assert num_running_replicas == expected_num_replicas, (
              f"Status: {deployment}"
              f"\nAutoscaling metrics: {autoscaling_metrics}"
          )
      return True
  
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 1 for calling the TestTargetCapacityUpdateAndServeStatus obj's underlying member method code with example usage: obj.check_num_replicas(expected_num_replicas, app_name, deployment_name, replica_state, controller_handle) and returns: True