code: def main():
      opts = parse_opts()
      if opts.verbose:
          log_level = logging.DEBUG
      else:
          log_level = logging.INFO
      should_test_modules = opts.testnames is None
      logging.basicConfig(
          stream=sys.stdout, level=log_level, format="%(message)s"
      )
      LOGGER.info("Running PySpark tests. Output is in %s", LOG_FILE)
      if os.path.exists(LOG_FILE):
          os.remove(LOG_FILE)
      python_execs = opts.python_executables.split(",")
      LOGGER.info(
          "Will test against the following Python executables: %s",
          python_execs,
      )
  
      if should_test_modules:
          modules_to_test = []
          for module_name in opts.modules.split(","):
              if module_name in python_modules:
                  modules_to_test.append(python_modules[module_name])
              else:
                  print(
                      "Error: unrecognized module '%s'. Supported modules: %s"
                      % (module_name, ", ".join(python_modules))
                  )
                  sys.exit(-1)
          LOGGER.info(
              "Will test the following Python modules: %s",
              [x.name for x in modules_to_test],
          )
      else:
          testnames_to_test = opts.testnames.split(",")
          LOGGER.info(
              "Will test the following Python tests: %s",
              testnames_to_test,
          )
  
      task_queue = Queue.PriorityQueue()
      for python_exec in python_execs:
          # Check if the python executable has coverage installed when 'COVERAGE_PROCESS_START'
          # environmental variable is set.
          if "COVERAGE_PROCESS_START" in os.environ:
              _check_coverage(python_exec)
  
          python_implementation = subprocess_check_output(
              [
                  python_exec,
                  "-c",
                  "import platform; print(platform.python_implementation())",
              ],
              universal_newlines=True,
          ).strip()
          LOGGER.info(
              "%s python_implementation is %s",
              python_exec,
              python_implementation,
          )
          LOGGER.info(
              "%s version is: %s",
              python_exec,
              subprocess_check_output(
                  [python_exec, "--version"],
                  stderr=subprocess.STDOUT,
                  universal_newlines=True,
              ).strip(),
          )
          if should_test_modules:
              for module in modules_to_test:
                  if (
                      python_implementation
                      not in module.excluded_python_implementations
                  ):
                      for test_goal in module.python_test_goals:
                          heavy_tests = [
                              "pyspark.streaming.tests",
                              "pyspark.mllib.tests",
                              "pyspark.tests",
                              "pyspark.sql.tests",
                              "pyspark.ml.tests",
                              "pyspark.pandas.tests",
                          ]
                          if any(
                              map(
                                  lambda prefix: test_goal.startswith(
                                      prefix
                                  ),
                                  heavy_tests,
                              )
                          ):
                              priority = 0
                          else:
                              priority = 100
                          task_queue.put(
                              (priority, (python_exec, test_goal))
                          )
          else:
              for test_goal in testnames_to_test:
                  task_queue.put((0, (python_exec, test_goal)))
  
      # Create the target directory before starting tasks to avoid races.
      target_dir = os.path.abspath(
          os.path.join(os.path.dirname(__file__), "target")
      )
      if not os.path.isdir(target_dir):
          os.mkdir(target_dir)
  
      def process_queue(task_queue):
          while True:
              try:
                  (
                      priority,
                      (python_exec, test_goal),
                  ) = task_queue.get_nowait()
              except Queue.Empty:
                  break
              try:
                  run_individual_python_test(
                      target_dir,
                      test_goal,
                      python_exec,
                      opts.keep_test_output,
                  )
              finally:
                  task_queue.task_done()
  
      start_time = time.time()
      for _ in range(opts.parallelism):
          worker = Thread(target=process_queue, args=(task_queue,))
          worker.daemon = True
          worker.start()
      try:
          task_queue.join()
      except (KeyboardInterrupt, SystemExit):
          print_red("Exiting due to interrupt")
          sys.exit(-1)
      total_duration = time.time() - start_time
      LOGGER.info("Tests passed in %i seconds", total_duration)
  
      for key, lines in sorted(SKIPPED_TESTS.items()):
          pyspark_python, test_name = key
          LOGGER.info(
              "\nSkipped tests in %s with %s:"
              % (test_name, pyspark_python)
          )
          for line in lines:
              LOGGER.info("    %s" % line.rstrip())
  
imports: import logging
  from argparse import ArgumentParser
  import os
  import platform
  import re
  import shutil
  import subprocess
  import sys
  import tempfile
  from threading import Thread, Lock
  import time
  import uuid
  import queue
  from multiprocessing import Manager
  from sparktestsupport import SPARK_HOME
  from sparktestsupport.shellutils import which, subprocess_check_output
  from sparktestsupport.modules import all_modules, pyspark_sql
label_desc: Use image node_id 7 for calling a global function with example usage: main() without return types