code: def test_cpu_allocation(self):
      for (
          max_worker_nodes,
          num_cpus_worker_node,
          max_worker_nodes_arg,
      ) in [
          (
              self.max_spark_tasks // 2,
              self.num_cpus_per_spark_task,
              self.max_spark_tasks // 2,
          ),
          (
              self.max_spark_tasks,
              self.num_cpus_per_spark_task,
              MAX_NUM_WORKER_NODES,
          ),
          (
              self.max_spark_tasks // 2,
              self.num_cpus_per_spark_task * 2,
              MAX_NUM_WORKER_NODES,
          ),
          (
              self.max_spark_tasks // 2,
              self.num_cpus_per_spark_task * 2,
              self.max_spark_tasks // 2 + 1,
          ),  # Test case: requesting resources exceeding all cluster resources
      ]:
          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_arg,
              num_cpus_worker_node=num_cpus_worker_node,
              head_node_options={"include_dashboard": False},
          ):
              ray.init()
              worker_res_list = self.get_ray_worker_resources_list()
              assert len(worker_res_list) == max_worker_nodes
              for worker_res in worker_res_list:
                  assert (
                      worker_res["CPU"] == num_cpus_worker_node
                      and worker_res["memory"] == mem_worker_node
                      and worker_res["object_store_memory"]
                      == object_store_mem_worker_node
                  )
  
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 3 for calling the RayOnSparkCPUClusterTestBase obj's underlying member method code with example usage: obj.test_cpu_allocation() without return types