code: def test_job_observability(ray_start_regular):
      driver_template = """
  import ray
  from time import sleep
  
  ray.init(address="{}")
  open("{}", "w+").close()
  
  print("My job id: ", str(ray.get_runtime_context().job_id))
  
  {}
  ray.shutdown()
      """
      tmpfile1 = tempfile.NamedTemporaryFile(
          "w+", suffix=".tmp", prefix="_"
      )
      tmpfile2 = tempfile.NamedTemporaryFile(
          "w+", suffix=".tmp", prefix="_"
      )
      tmpfiles = [tmpfile1.name, tmpfile2.name]
      tmpfile1.close()
      tmpfile2.close()
      for tmpfile in tmpfiles:
          if os.path.exists(tmpfile):
              os.unlink(tmpfile)
  
      non_hanging = driver_template.format(
          ray_start_regular["address"], tmpfiles[0], "sleep(1)"
      )
      hanging_driver = driver_template.format(
          ray_start_regular["address"], tmpfiles[1], "sleep(60)"
      )
  
      out = run_string_as_driver(non_hanging)
      p = run_string_as_driver_nonblocking(hanging_driver)
      # The nonblocking process needs time to connect.
      while not os.path.exists(tmpfiles[1]):
          time.sleep(1)
  
      jobs = list(ray._private.state.jobs())
      jobs.sort(key=lambda x: x["JobID"])
  
      driver = jobs[0]
      finished = jobs[1]
      running = jobs[2]
  
      # The initial driver timestamp/start time go down a different code path.
      assert driver["Timestamp"] == driver["StartTime"]
      assert finished["Timestamp"] == finished["EndTime"]
      assert running["Timestamp"] == running["StartTime"]
  
      assert finished["EndTime"] > finished["StartTime"] > 0, out
      lapsed = finished["EndTime"] - finished["StartTime"]
      assert 0 < lapsed < 5000, f"Job should've taken ~1s, {finished}"
  
      assert running["StartTime"] > 0
      assert running["EndTime"] == 0
  
      assert len(running["DriverIPAddress"]) > 0
      assert running["DriverPid"] > 0
      assert len(finished["DriverIPAddress"]) > 0
      assert finished["DriverPid"] > 0
  
      p.kill()
      # Give the second job time to clean itself up.
      time.sleep(1)
  
      jobs = list(ray._private.state.jobs())
      jobs.sort(key=lambda x: x["JobID"])
  
      # jobs[0] is the test case driver.
      finished = jobs[1]
      prev_running = jobs[2]
  
      assert (
          finished["EndTime"] > finished["StartTime"] > 0
      ), f"{finished}"
      assert finished["EndTime"] == finished["Timestamp"]
      lapsed = finished["EndTime"] - finished["StartTime"]
      assert 0 < lapsed < 5000, f"Job should've taken ~1s {finished}"
  
      assert prev_running["EndTime"] > prev_running["StartTime"] > 0
  
      assert len(prev_running["DriverIPAddress"]) > 0
      assert prev_running["DriverPid"] > 0
  
imports: import os
  import subprocess
  import sys
  import tempfile
  import time
  import re
  import json
  from subprocess import Popen, PIPE, STDOUT, list2cmdline
  from typing import List
  from pathlib import Path
  import pytest
  import ray
  from ray._private.test_utils import run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition, format_web_url
  from ray.job_config import JobConfig
  from ray.job_submission import JobSubmissionClient
  from ray.dashboard.modules.job.pydantic_models import JobDetails
label_desc: Use image node_id 5 for calling a global function with example usage: test_job_observability(ray_start_regular) without return types