code: def test_autoscaling(self):
      for max_worker_nodes, num_cpus_worker_node, min_worker_nodes in [
          (self.max_spark_tasks, self.num_cpus_per_spark_task, 0),
          (
              self.max_spark_tasks // 2,
              self.num_cpus_per_spark_task * 2,
              0,
          ),
          (self.max_spark_tasks, self.num_cpus_per_spark_task, 1),
      ]:
          num_ray_task_slots = self.max_spark_tasks // (
              num_cpus_worker_node // self.num_cpus_per_spark_task
          )
          (
              mem_worker_node,
              object_store_mem_worker_node,
              _,
          ) = _calc_mem_per_ray_worker_node(
              num_task_slots=num_ray_task_slots,
              physical_mem_bytes=_RAY_ON_SPARK_WORKER_PHYSICAL_MEMORY_BYTES,
              shared_mem_bytes=_RAY_ON_SPARK_WORKER_SHARED_MEMORY_BYTES,
              configured_object_store_bytes=None,
          )
  
          with _setup_ray_cluster(
              max_worker_nodes=max_worker_nodes,
              min_worker_nodes=min_worker_nodes,
              num_cpus_worker_node=num_cpus_worker_node,
              head_node_options={"include_dashboard": False},
              autoscale_idle_timeout_minutes=0.1,
          ):
              ray.init()
              worker_res_list = self.get_ray_worker_resources_list()
              assert len(worker_res_list) == min_worker_nodes
  
              @ray.remote(num_cpus=num_cpus_worker_node)
              def f(x):
                  import time
  
                  time.sleep(5)
                  return x * x
  
              # Test scale up
              futures = [f.remote(i) for i in range(8)]
              results = ray.get(futures)
              assert results == [i * i for i in range(8)]
  
              worker_res_list = self.get_ray_worker_resources_list()
              assert len(worker_res_list) == max_worker_nodes and all(
                  worker_res_list[i]["CPU"] == num_cpus_worker_node
                  and worker_res_list[i]["memory"] == mem_worker_node
                  and worker_res_list[i]["object_store_memory"]
                  == object_store_mem_worker_node
                  for i in range(max_worker_nodes)
              )
  
              # Test scale down
              wait_for_condition(
                  lambda: len(self.get_ray_worker_resources_list())
                  == min_worker_nodes,
                  timeout=60,
                  retry_interval_ms=1000,
              )
              if min_worker_nodes > 0:
                  # Test scaling down keeps nodes number >= min_worker_nodes
                  time.sleep(30)
                  assert (
                      len(self.get_ray_worker_resources_list())
                      == min_worker_nodes
                  )
  
imports: import os
  import shutil
  import tempfile
  import socket
  import threading
  import re
  import pytest
  import sys
  from unittest import mock
  from abc import ABC
  import ray
  import ray.util.spark.cluster_init
  from ray.util.spark import setup_ray_cluster, shutdown_ray_cluster, setup_global_ray_cluster, MAX_NUM_WORKER_NODES
  from ray.util.spark.utils import is_port_in_use, _calc_mem_per_ray_worker_node
  from pyspark.sql import SparkSession
  import time
  import logging
  from contextlib import contextmanager
  from ray._private.test_utils import wait_for_condition
label_desc: Use image node_id 7 for calling the RayOnSparkCPUClusterTestBase obj's underlying member method code with example usage: obj.test_autoscaling() and returns: unknown