docstr: Unit test for upscale_delay_s and downscale_delay_s.
docstr_tok: ["Unit","test","for","upscale_delay_s","and","downscale_delay_s","."]
code: def test_upscale_downscale_delay(self):
      """Unit test for upscale_delay_s and downscale_delay_s."""
  
      upscale_delay_s = 30.0
      downscale_delay_s = 600.0
  
      config = AutoscalingConfig(
          min_replicas=0,
          max_replicas=2,
          target_num_ongoing_requests_per_replica=1,
          upscale_delay_s=30.0,
          downscale_delay_s=600.0,
      )
  
      policy_manager = AutoscalingPolicyManager(config)
  
      upscale_wait_periods = int(
          upscale_delay_s / CONTROL_LOOP_PERIOD_S
      )
      downscale_wait_periods = int(
          downscale_delay_s / CONTROL_LOOP_PERIOD_S
      )
  
      overload_requests = [100]
  
      # Scale up when there are 0 replicas and current_handle_queued_queries > 0
      new_num_replicas = policy_manager.get_decision_num_replicas(
          current_num_ongoing_requests=[],
          curr_target_num_replicas=0,
          current_handle_queued_queries=1,
      )
      assert new_num_replicas == 1
  
      # We should scale up only after enough consecutive scale-up decisions.
      for i in range(upscale_wait_periods):
          new_num_replicas = policy_manager.get_decision_num_replicas(
              current_num_ongoing_requests=overload_requests,
              curr_target_num_replicas=1,
              current_handle_queued_queries=0,
          )
          assert new_num_replicas == 1, i
  
      new_num_replicas = policy_manager.get_decision_num_replicas(
          current_num_ongoing_requests=overload_requests,
          curr_target_num_replicas=1,
          current_handle_queued_queries=0,
      )
      assert new_num_replicas == 2
  
      no_requests = [0, 0]
  
      # We should scale down only after enough consecutive scale-down decisions.
      for i in range(downscale_wait_periods):
          new_num_replicas = policy_manager.get_decision_num_replicas(
              current_num_ongoing_requests=no_requests,
              curr_target_num_replicas=2,
              current_handle_queued_queries=0,
          )
          assert new_num_replicas == 2, i
  
      new_num_replicas = policy_manager.get_decision_num_replicas(
          current_num_ongoing_requests=no_requests,
          curr_target_num_replicas=2,
          current_handle_queued_queries=0,
      )
      assert new_num_replicas == 0
  
      # Get some scale-up decisions, but not enough to trigger a scale up.
      for i in range(int(upscale_wait_periods / 2)):
          new_num_replicas = policy_manager.get_decision_num_replicas(
              current_num_ongoing_requests=overload_requests,
              curr_target_num_replicas=1,
              current_handle_queued_queries=0,
          )
          assert new_num_replicas == 1, i
  
      # Interrupt with a scale-down decision.
      policy_manager.get_decision_num_replicas(
          current_num_ongoing_requests=[0],
          curr_target_num_replicas=1,
          current_handle_queued_queries=0,
      )
  
      # The counter should be reset, so it should require `upscale_wait_periods`
      # more periods before we actually scale up.
      for i in range(upscale_wait_periods):
          new_num_replicas = policy_manager.get_decision_num_replicas(
              current_num_ongoing_requests=overload_requests,
              curr_target_num_replicas=1,
              current_handle_queued_queries=0,
          )
          assert new_num_replicas == 1, i
  
      new_num_replicas = policy_manager.get_decision_num_replicas(
          current_num_ongoing_requests=overload_requests,
          curr_target_num_replicas=1,
          current_handle_queued_queries=0,
      )
      assert new_num_replicas == 2
  
      # Get some scale-down decisions, but not enough to trigger a scale down.
      for i in range(int(downscale_wait_periods / 2)):
          new_num_replicas = policy_manager.get_decision_num_replicas(
              current_num_ongoing_requests=no_requests,
              curr_target_num_replicas=2,
              current_handle_queued_queries=0,
          )
          assert new_num_replicas == 2, i
  
      # Interrupt with a scale-up decision.
      policy_manager.get_decision_num_replicas(
          current_num_ongoing_requests=[100, 100],
          curr_target_num_replicas=2,
          current_handle_queued_queries=0,
      )
  
      # The counter should be reset so it should require `downscale_wait_periods`
      # more periods before we actually scale down.
      for i in range(downscale_wait_periods):
          new_num_replicas = policy_manager.get_decision_num_replicas(
              current_num_ongoing_requests=no_requests,
              curr_target_num_replicas=2,
              current_handle_queued_queries=0,
          )
          assert new_num_replicas == 2, i
  
      new_num_replicas = policy_manager.get_decision_num_replicas(
          current_num_ongoing_requests=no_requests,
          curr_target_num_replicas=2,
          current_handle_queued_queries=0,
      )
      assert new_num_replicas == 0
  
imports: import sys
  import numpy
  import pytest
  from ray.serve._private.autoscaling_policy import AutoscalingPolicyManager
  from ray.serve._private.constants import CONTROL_LOOP_PERIOD_S
  from ray.serve.autoscaling_policy import _calculate_desired_num_replicas
  from ray.serve.config import AutoscalingConfig
label_desc: Use image node_id 3 for calling the TestGetDecisionNumReplicas obj's underlying member method code with example usage: obj.test_upscale_downscale_delay() without return types