code: def test_e2e_preserve_prev_replicas(serve_instance):
      signal = SignalActor.remote()
  
      @serve.deployment(
          max_concurrent_queries=5,
          # The config makes the deployment scale up really quickly and then
          # wait nearly forever to downscale.
          autoscaling_config=AutoscalingConfig(
              min_replicas=1,
              max_replicas=2,
              downscale_delay_s=600,
              upscale_delay_s=0,
              metrics_interval_s=1,
              look_back_period_s=1,
          ),
      )
      def scaler():
          ray.get(signal.wait.remote())
          time.sleep(0.2)
          return os.getpid()
  
      handle = serve.run(scaler.bind())
      dep_id = DeploymentID("scaler", SERVE_DEFAULT_APP_NAME)
      responses = [handle.remote() for _ in range(10)]
  
      def check_two_replicas():
          actors = state_api.list_actors(
              filters=[
                  (
                      "class_name",
                      "=",
                      dep_id.to_replica_actor_class_name(),
                  ),
                  ("state", "=", "ALIVE"),
              ]
          )
          print(actors)
          return len(actors) == 2
  
      wait_for_condition(
          check_two_replicas, retry_interval_ms=1000, timeout=20
      )
  
      ray.get(signal.send.remote())
  
      pids = {r.result() for r in responses}
      assert len(pids) == 2
  
      # Now re-deploy the application, make sure it is still 2 replicas and it shouldn't
      # be scaled down.
      handle = serve.run(scaler.bind())
      responses = [handle.remote() for _ in range(10)]
      pids = {r.result() for r in responses}
      assert len(pids) == 2
  
      def check_num_replicas(live: int, dead: int):
          live_actors = state_api.list_actors(
              filters=[
                  (
                      "class_name",
                      "=",
                      dep_id.to_replica_actor_class_name(),
                  ),
                  ("state", "=", "ALIVE"),
              ]
          )
          dead_actors = state_api.list_actors(
              filters=[
                  (
                      "class_name",
                      "=",
                      dep_id.to_replica_actor_class_name(),
                  ),
                  ("state", "=", "DEAD"),
              ]
          )
  
          return len(live_actors) == live and len(dead_actors) == dead
  
      wait_for_condition(
          check_num_replicas,
          retry_interval_ms=1000,
          timeout=20,
          live=2,
          dead=2,
      )
      ray.get(signal.send.remote())
  
      # re-deploy the application with initial_replicas. This should override the
      # previous number of replicas.
      scaler = scaler.options(
          autoscaling_config=AutoscalingConfig(
              min_replicas=1,
              initial_replicas=3,
              max_replicas=5,
              downscale_delay_s=600,
              upscale_delay_s=600,
              metrics_interval_s=1,
              look_back_period_s=1,
          )
      )
      handle = serve.run(scaler.bind())
      responses = [handle.remote() for _ in range(15)]
      pids = {r.result() for r in responses}
      assert len(pids) == 3
  
      wait_for_condition(
          check_num_replicas,
          retry_interval_ms=1000,
          timeout=20,
          live=3,
          dead=4,
      )
  
imports: import asyncio
  import logging
  import os
  import sys
  import tempfile
  import time
  import zipfile
  from typing import Dict, Iterable, List
  from unittest import mock
  import pytest
  import requests
  import ray
  import ray.util.state
  from ray import serve
  from ray._private.test_utils import SignalActor, wait_for_condition
  from ray.serve._private.common import ApplicationStatus, DeploymentID, DeploymentStatus, DeploymentStatusInfo, DeploymentStatusTrigger, ReplicaState
  from ray.serve._private.constants import SERVE_DEFAULT_APP_NAME, SERVE_NAMESPACE
  from ray.serve._private.controller import ServeController
  from ray.serve.config import AutoscalingConfig
  from ray.serve.generated.serve_pb2 import DeploymentStatusInfo
  from ray.serve.schema import ServeDeploySchema
label_desc: Use image node_id 20 for calling a global function with example usage: test_e2e_preserve_prev_replicas(serve_instance) and returns: os, unknown, unknown