code: def test_handle_driver_tasks(shutdown_only):
      ray.init(_system_config=_SYSTEM_CONFIG)
  
      job_id = ray.get_runtime_context().get_job_id()
      script = """
  import ray
  import time
  ray.init("auto")
  
  @ray.remote
  def f():
      time.sleep(3)
  
  
  ray.get(f.remote())
  """
      run_string_as_driver_nonblocking(script)
  
      client = StateApiClient()
  
      def list_tasks(exclude_driver):
          return client.list(
              StateResource.TASKS,
              # Filter out this driver
              options=ListApiOptions(
                  exclude_driver=exclude_driver,
                  filters=[("job_id", "!=", job_id)],
              ),
              raise_on_missing_output=True,
          )
  
      # Check driver running
      def verify():
          tasks_with_driver = list_tasks(exclude_driver=False)
          assert len(tasks_with_driver) == 2, tasks_with_driver
          task_types = {task["type"] for task in tasks_with_driver}
          assert task_types == {"NORMAL_TASK", "DRIVER_TASK"}
  
          for task in tasks_with_driver:
              if task["type"] == "DRIVER_TASK":
                  assert task["state"] == "RUNNING", task
  
          return True
  
      wait_for_condition(verify, timeout=15, retry_interval_ms=1000)
  
      # Check driver finishes
      def verify():
          tasks_with_driver = list_tasks(exclude_driver=False)
          assert len(tasks_with_driver) == 2, tasks_with_driver
          for task in tasks_with_driver:
              if task["type"] == "DRIVER_TASK":
                  assert task["state"] == "FINISHED", task
  
          tasks_no_driver = list_tasks(exclude_driver=True)
          assert len(tasks_no_driver) == 1, tasks_no_driver
          return True
  
      wait_for_condition(verify)
  
imports: import asyncio
  from collections import defaultdict
  import os
  from typing import Dict
  import pytest
  import sys
  import time
  from ray._private import ray_constants
  from functools import reduce
  import ray
  from ray._private.state_api_test_utils import PidActor, get_state_api_manager, verify_tasks_running_or_terminated, verify_failed_task
  from ray.util.state.common import ListApiOptions, StateResource
  from ray._private.test_utils import async_wait_for_condition_async_predicate, run_string_as_driver, run_string_as_driver_nonblocking, wait_for_condition
  from ray.util.state import StateApiClient, list_actors, list_jobs, list_tasks
label_desc: Use image node_id 5 for calling a global function with example usage: test_handle_driver_tasks(shutdown_only) and returns: client, True, True