code: def test_e2e_preserve_prev_replicas_rest_api(serve_instance):
      client = serve_instance
      signal = SignalActor.options(
          name="signal", namespace="serve"
      ).remote()
  
      # Step 1: Prepare the script in a zip file so it can be submitted via REST API.
      with tempfile.NamedTemporaryFile(
          suffix=".zip", delete=False
      ) as tmp_path:
          with zipfile.ZipFile(tmp_path, "w") as zip_obj:
              with zip_obj.open("app.py", "w") as f:
                  f.write(
                      """
  from ray import serve
  import ray
  import os
  
  @serve.deployment
  def g():
      signal = ray.get_actor("signal", namespace="serve")
      ray.get(signal.wait.remote())
      return os.getpid()
  
  
  app = g.bind()
  """.encode()
                  )
  
      # Step 2: Deploy it with max_replicas=1
      app_config = {
          "import_path": "app:app",
          "runtime_env": {"working_dir": f"file://{tmp_path.name}"},
          "deployments": [
              {
                  "name": "g",
                  "autoscaling_config": {
                      "min_replicas": 0,
                      "max_replicas": 1,
                      "downscale_delay_s": 600,
                      "upscale_delay_s": 0,
                      "metrics_interval_s": 1,
                      "look_back_period_s": 1,
                  },
              }
          ],
      }
  
      client.deploy_apps(
          ServeDeploySchema(**{"applications": [app_config]})
      )
      dep_id = DeploymentID("g", SERVE_DEFAULT_APP_NAME)
      wait_for_condition(
          lambda: serve.status()
          .applications[SERVE_DEFAULT_APP_NAME]
          .status
          == "RUNNING"
      )
  
      # Step 3: Verify that it can scale from 0 to 1.
      @ray.remote
      def send_request():
          return requests.get("http://localhost:8000/").text
  
      ref = send_request.remote()
  
      def check_num_replicas(num: int):
          actors = state_api.list_actors(
              filters=[
                  (
                      "class_name",
                      "=",
                      dep_id.to_replica_actor_class_name(),
                  ),
                  ("state", "=", "ALIVE"),
              ]
          )
          return len(actors) == num
  
      wait_for_condition(
          check_num_replicas, retry_interval_ms=1000, timeout=20, num=1
      )
  
      signal.send.remote()
      existing_pid = ray.get(ref)
  
      # Step 4: Change the max replicas to 2
      app_config["deployments"][0]["autoscaling_config"][
          "max_replicas"
      ] = 2
      client.deploy_apps(
          ServeDeploySchema(**{"applications": [app_config]})
      )
      wait_for_condition(
          lambda: serve.status()
          .applications[SERVE_DEFAULT_APP_NAME]
          .status
          == "RUNNING"
      )
      wait_for_condition(
          check_num_replicas, retry_interval_ms=1000, timeout=20, num=1
      )
  
      # Step 5: Make sure it is the same replica (lightweight change).
      for _ in range(10):
          other_pid = ray.get(send_request.remote())
          assert other_pid == existing_pid
  
      # Step 6: Make sure initial_replicas overrides previous replicas
      app_config["deployments"][0]["autoscaling_config"][
          "max_replicas"
      ] = 5
      app_config["deployments"][0]["autoscaling_config"][
          "initial_replicas"
      ] = 3
      app_config["deployments"][0]["autoscaling_config"][
          "upscale_delay"
      ] = 600
      client.deploy_apps(
          ServeDeploySchema(**{"applications": [app_config]})
      )
      wait_for_condition(
          lambda: serve.status()
          .applications[SERVE_DEFAULT_APP_NAME]
          .status
          == "RUNNING"
      )
      wait_for_condition(
          check_num_replicas, retry_interval_ms=1000, timeout=20, num=3
      )
  
      # Step 7: Make sure original replica is still running (lightweight change)
      pids = set()
      for _ in range(15):
          pids.add(ray.get(send_request.remote()))
      assert existing_pid in pids
  
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 21 for calling a global function with example usage: test_e2e_preserve_prev_replicas_rest_api(serve_instance) and returns: requests, unknown