code: def test_ray_cluster_shutdown(self):
      with _setup_ray_cluster(
          max_worker_nodes=self.max_spark_tasks
      ) as cluster:
          ray.init()
          assert (
              len(self.get_ray_worker_resources_list())
              == self.max_spark_tasks
          )
  
          # Test: cancel background spark job will cause all ray worker nodes exit.
          cluster._cancel_background_spark_job()
          time.sleep(8)
  
          assert len(self.get_ray_worker_resources_list()) == 0
  
      time.sleep(2)  # wait ray head node exit.
      # assert ray head node exit by checking head port being closed.
      hostname, port = cluster.address.split(":")
      assert not is_port_in_use(hostname, int(port))
  
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 5 for calling the RayOnSparkCPUClusterTestBase obj's underlying member method code with example usage: obj.test_ray_cluster_shutdown() without return types