code: def test_replica_startup_status_transitions(ray_cluster):
      cluster = ray_cluster
      cluster.add_node(num_cpus=1)
      cluster.connect(namespace=SERVE_NAMESPACE)
      serve.start()
      client = _get_global_client()
  
      signal = SignalActor.remote()
  
      @serve.deployment(version="1", ray_actor_options={"num_cpus": 2})
      class E:
          async def __init__(self):
              await signal.wait.remote()
  
      serve.run(E.bind(), _blocking=False)
  
      def get_replicas(replica_state):
          controller = client._controller
          replicas = ray.get(
              controller._dump_replica_states_for_testing.remote(
                  DeploymentID(E.name, "default")
              )
          )
          return replicas.get([replica_state])
  
      # wait for serve to start the replica, and catch a reference to it.
      wait_for_condition(
          lambda: len(get_replicas(ReplicaState.STARTING)) > 0
      )
      replica = get_replicas(ReplicaState.STARTING)[0]
  
      # currently there are no resources to allocate the replica
      assert (
          replica.check_started()[0]
          == ReplicaStartupStatus.PENDING_ALLOCATION
      )
  
      # add the necessary resources to allocate the replica
      cluster.add_node(num_cpus=4)
      wait_for_condition(
          lambda: (ray.cluster_resources().get("CPU", 0) >= 4)
      )
      wait_for_condition(
          lambda: (ray.available_resources().get("CPU", 0) >= 2)
      )
  
      def is_replica_pending_initialization():
          status, _ = replica.check_started()
          print(status)
          return status == ReplicaStartupStatus.PENDING_INITIALIZATION
  
      wait_for_condition(is_replica_pending_initialization, timeout=25)
  
      # send signal to complete replica intialization
      signal.send.remote()
      wait_for_condition(
          lambda: replica.check_started()[0]
          == ReplicaStartupStatus.SUCCEEDED
      )
  
imports: import os
  import sys
  import time
  from collections import defaultdict
  import pytest
  import requests
  import ray
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.cluster_utils import Cluster
  from ray.exceptions import RayActorError
  from ray.serve._private.common import DeploymentID, ReplicaState
  from ray.serve._private.constants import SERVE_NAMESPACE
  from ray.serve._private.deployment_state import ReplicaStartupStatus
  from ray.serve._private.utils import get_head_node_id
  from ray.serve.context import _get_global_client
  from ray.serve.handle import RayServeHandle
  from ray.serve.schema import ServeDeploySchema
label_desc: Use image node_id 5 for calling a global function with example usage: test_replica_startup_status_transitions(ray_cluster) and returns: replicas, unknown