code: def test_jobs_run_on_head_by_default_E2E(
      ray_start_cluster_head_with_env_vars,
  ):
      allow_driver_on_worker_nodes = (
          os.environ.get(RAY_JOB_ALLOW_DRIVER_ON_WORKER_NODES_ENV_VAR)
          == "1"
      )
      # Cluster setup
      cluster = ray_start_cluster_head_with_env_vars
      cluster.add_node(dashboard_agent_listen_port=52366)
      cluster.add_node(dashboard_agent_listen_port=52367)
      assert wait_until_server_available(cluster.webui_url) is True
      webui_url = cluster.webui_url
      webui_url = format_web_url(webui_url)
      client = JobSubmissionClient(webui_url)
  
      def _check_nodes(num_nodes):
          try:
              assert len(list_nodes()) == num_nodes
              return True
          except Exception as ex:
              print(ex)
              return False
  
      wait_for_condition(lambda: _check_nodes(num_nodes=3), timeout=15)
      wait_for_condition(
          lambda: get_register_agents_number(webui_url) == 3, timeout=20
      )
  
      # Submit 20 simple jobs.
      for i in range(20):
          client.submit_job(
              entrypoint="echo hi", submission_id=f"job_{i}"
          )
      import pprint
  
      def check_all_jobs_succeeded():
          submission_jobs = [
              job
              for job in client.list_jobs()
              if job.type == JobType.SUBMISSION
          ]
          for job in submission_jobs:
              pprint.pprint(job)
              if job.status != JobStatus.SUCCEEDED:
                  return False
          return True
  
      # Wait until all jobs have finished.
      wait_for_condition(
          check_all_jobs_succeeded, timeout=60, retry_interval_ms=1000
      )
  
      # Check driver_node_id of all jobs.
      submission_jobs = [
          job
          for job in client.list_jobs()
          if job.type == JobType.SUBMISSION
      ]
      driver_node_ids = [job.driver_node_id for job in submission_jobs]
  
      # Spuriously fails with probability (1/3)^20.
      pprint.pprint(driver_node_ids)
      num_ids = len(set(driver_node_ids))
      assert (
          (num_ids > 1)
          if allow_driver_on_worker_nodes
          else (num_ids == 1)
      ), [id[:5] for id in driver_node_ids]
  
imports: from pathlib import Path
  import tempfile
  import time
  import os
  from ray.dashboard.modules.job.pydantic_models import JobType
  import requests
  import pytest
  import psutil
  import sys
  from typing import Dict, Optional, Tuple
  from unittest.mock import Mock, patch
  from ray._private.ray_constants import DEFAULT_DASHBOARD_AGENT_LISTEN_PORT
  from ray._private.test_utils import format_web_url, wait_for_condition, wait_until_server_available
  from ray.dashboard.modules.dashboard_sdk import ClusterInfo, DEFAULT_DASHBOARD_ADDRESS, parse_cluster_info
  from ray.dashboard.modules.job.sdk import JobSubmissionClient, JobStatus
  from ray.dashboard.tests.conftest import *
  from ray.dashboard.consts import RAY_JOB_ALLOW_DRIVER_ON_WORKER_NODES_ENV_VAR
  from ray.tests.conftest import _ray_start
  import ray
  import ray.experimental.internal_kv
  from ray.util.state import list_nodes
label_desc: Use image node_id 9 for calling a global function with example usage: test_jobs_run_on_head_by_default_E2E(ray_start_cluster_head_with_env_vars) and returns: True, True, False, False