code: def test_jobs_cluster_events(shutdown_only):
      ray.init()
      address = ray._private.worker._global_node.webui_url
      address = format_web_url(address)
      client = JobSubmissionClient(address)
      submission_id = client.submit_job(entrypoint="ls")
  
      def verify():
          events = list_cluster_events()
          assert len(list_cluster_events()) == 2
          start_event = events[0]
          completed_event = events[1]
  
          assert start_event["source_type"] == "JOBS"
          assert (
              f"Started a ray job {submission_id}"
              in start_event["message"]
          )
          assert start_event["severity"] == "INFO"
          assert completed_event["source_type"] == "JOBS"
          assert (
              f"Completed a ray job {submission_id} with a status SUCCEEDED."
              == completed_event["message"]
          )
          assert completed_event["severity"] == "INFO"
          return True
  
      print("Test successful job run.")
      wait_for_condition(verify)
      pprint(list_cluster_events())
  
      # Test the failure case. In this part, job fails because the runtime env
      # creation fails.
      submission_id = client.submit_job(
          entrypoint="ls",
          runtime_env={"pip": ["nonexistent_dep"]},
      )
  
      def verify():
          events = list_cluster_events(detail=True)
          failed_events = []
  
          for e in events:
              if (
                  "submission_id" in e["custom_fields"]
                  and e["custom_fields"]["submission_id"]
                  == submission_id
              ):
                  failed_events.append(e)
  
          assert len(failed_events) == 2
          failed_start = failed_events[0]
          failed_completed = failed_events[1]
  
          assert failed_start["source_type"] == "JOBS"
          assert (
              f"Started a ray job {submission_id}"
              in failed_start["message"]
          )
          assert failed_completed["source_type"] == "JOBS"
          assert failed_completed["severity"] == "ERROR"
          assert (
              f"Completed a ray job {submission_id} with a status FAILED."
              in failed_completed["message"]
          )
  
          # Make sure the error message is included.
          assert (
              "ERROR: No matching distribution found"
              in failed_completed["message"]
          )
          return True
  
      print("Test failed (runtime_env failure) job run.")
      wait_for_condition(verify, timeout=30)
      pprint(list_cluster_events())
  
imports: import os
  import sys
  import time
  import json
  import copy
  import logging
  import requests
  import asyncio
  import random
  import tempfile
  import socket
  from pprint import pprint
  from datetime import datetime
  import pytest
  import numpy
  import ray
  from ray.util.state import list_cluster_events
  from ray._private.utils import binary_to_hex
  from ray.cluster_utils import AutoscalingCluster
  from ray._private.event.event_logger import filter_event_by_level, get_event_logger
  from ray.dashboard.tests.conftest import *
  from ray.dashboard.modules.event import event_consts
  from ray.core.generated import event_pb2
  from ray._private.test_utils import format_web_url, wait_until_server_available, wait_for_condition
  from ray.dashboard.modules.event.event_utils import monitor_events
  from ray.job_submission import JobSubmissionClient
label_desc: Use image node_id 8 for calling a global function with example usage: test_jobs_cluster_events(shutdown_only) and returns: True, True