code: def test_e2e_intermediate_downscaling(serve_instance):
      """
      Scales up, then down, and up again.
      """
  
      controller = serve_instance._controller
      signal = SignalActor.remote()
  
      @serve.deployment(
          autoscaling_config={
              "metrics_interval_s": 0.1,
              "min_replicas": 0,
              "max_replicas": 20,
              "look_back_period_s": 0.2,
              "downscale_delay_s": 0.2,
              "upscale_delay_s": 0.2,
          },
          # We will send over a lot of queries. This will make sure replicas are
          # killed quickly during cleanup.
          graceful_shutdown_timeout_s=1,
          max_concurrent_queries=1000,
          version="v1",
      )
      class A:
          def __call__(self):
              ray.get(signal.wait.remote())
  
      handle = serve.run(A.bind())
      wait_for_condition(
          lambda: get_deployment_status(controller, "A")
          == DeploymentStatus.HEALTHY
      )
      start_time = get_deployment_start_time(controller, "A")
  
      [handle.remote() for _ in range(50)]
  
      wait_for_condition(
          check_autoscale_num_replicas_gte,
          controller=controller,
          name="A",
          target=20,
          timeout=30,
      )
      signal.send.remote()
  
      wait_for_condition(
          check_autoscale_num_replicas_lte,
          controller=controller,
          name="A",
          target=1,
          timeout=30,
      )
      signal.send.remote(clear=True)
  
      [handle.remote() for _ in range(50)]
      wait_for_condition(
          check_autoscale_num_replicas_gte,
          controller=controller,
          name="A",
          target=20,
          timeout=30,
      )
  
      signal.send.remote()
      # As the queue is drained, we should scale back down.
      wait_for_condition(
          check_autoscale_num_replicas_eq,
          controller=controller,
          name="A",
          target=0,
          timeout=30,
      )
  
      # Make sure start time did not change for the deployment
      assert get_deployment_start_time(controller, "A") == start_time
  
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 16 for calling a global function with example usage: test_e2e_intermediate_downscaling(serve_instance) without return types