code: def test_public_api(self):
      try:
          ray_temp_root_dir = tempfile.mkdtemp(dir="/tmp")
          collect_log_to_path = tempfile.mkdtemp(dir="/tmp")
          # Test the case that `collect_log_to_path` directory does not exist.
          shutil.rmtree(collect_log_to_path, ignore_errors=True)
          setup_ray_cluster(
              max_worker_nodes=MAX_NUM_WORKER_NODES,
              collect_log_to_path=collect_log_to_path,
              ray_temp_root_dir=ray_temp_root_dir,
              head_node_options={"include_dashboard": True},
          )
  
          assert (
              os.environ["RAY_ADDRESS"]
              == ray.util.spark.cluster_init._active_ray_cluster.address
          )
  
          ray.init()
  
          @ray.remote
          def f(x):
              return x * x
  
          futures = [f.remote(i) for i in range(32)]
          results = ray.get(futures)
          assert results == [i * i for i in range(32)]
  
          shutdown_ray_cluster()
  
          assert "RAY_ADDRESS" not in os.environ
  
          time.sleep(7)
          # assert temp dir is removed.
          assert len(os.listdir(ray_temp_root_dir)) == 1 and os.listdir(
              ray_temp_root_dir
          )[0].endswith(".lock")
  
          # assert logs are copied to specified path
          listed_items = os.listdir(collect_log_to_path)
          assert len(listed_items) == 1 and listed_items[0].startswith(
              "ray-"
          )
          log_dest_dir = os.path.join(
              collect_log_to_path, listed_items[0], socket.gethostname()
          )
          assert (
              os.path.exists(log_dest_dir)
              and len(os.listdir(log_dest_dir)) > 0
          )
      finally:
          if (
              ray.util.spark.cluster_init._active_ray_cluster
              is not None
          ):
              # if the test raised error and does not destroy cluster,
              # destroy it here.
              ray.util.spark.cluster_init._active_ray_cluster.shutdown()
              time.sleep(5)
          shutil.rmtree(ray_temp_root_dir, ignore_errors=True)
          shutil.rmtree(collect_log_to_path, ignore_errors=True)
  
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 4 for calling the RayOnSparkCPUClusterTestBase obj's underlying member method code with example usage: obj.test_public_api() and returns: unknown