code: def test_no_spurious_worker_startup(shutdown_only, runtime_env_class):
      """Test that no extra workers start up during a long env installation."""
  
      # Causes agent to sleep for 15 seconds to simulate creating a runtime env.
      os.environ["RAY_RUNTIME_ENV_SLEEP_FOR_TESTING_S"] = "15"
      ray.init(num_cpus=1)
  
      @ray.remote
      class Counter(object):
          def __init__(self):
              self.value = 0
  
          def get(self):
              return self.value
  
      # Set a nonempty runtime env so that the runtime env setup hook is called.
      runtime_env = runtime_env_class(env_vars={"a": "b"})
  
      # Instantiate an actor that requires the long runtime env installation.
      a = Counter.options(runtime_env=runtime_env).remote()
      assert ray.get(a.get.remote()) == 0
  
      # Check "debug_state.txt" to ensure no extra workers were started.
      session_dir = ray._private.worker.global_worker.node.address_info[
          "session_dir"
      ]
      session_path = Path(session_dir)
      debug_state_path = session_path / "logs" / "debug_state.txt"
  
      def get_num_workers():
          with open(debug_state_path) as f:
              for line in f.readlines():
                  num_workers_prefix = "- num PYTHON workers: "
                  if num_workers_prefix in line:
                      return int(line[len(num_workers_prefix) :])
          return None
  
      # Wait for "debug_state.txt" to be updated to reflect the started worker.
      start = time.time()
      wait_for_condition(
          lambda: get_num_workers() is not None
          and get_num_workers() > 0
      )
      time_waited = time.time() - start
      print(f"Waited {time_waited} for debug_state.txt to be updated")
  
      # If any workers were unnecessarily started during the initial env
      # installation, they will bypass the runtime env setup hook (because the
      # created env will have been cached) and should be added to num_workers
      # within a few seconds.  Adjusting the default update period for
      # debut_state.txt via this cluster_utils pytest fixture seems to be broken,
      # so just check it for the next 10 seconds (the default period).
      start = time.time()
      got_num_workers = False
      while time.time() - start < 10:
          # Check that no more than one extra worker is started. We add one
          # because Ray will prestart an idle worker for the one available CPU.
          num_workers = get_num_workers()
          if num_workers is not None:
              got_num_workers = True
              assert num_workers <= 2
          time.sleep(0.1)
      assert (
          got_num_workers
      ), "failed to read num workers for 10 seconds"
  
imports: from dataclasses import dataclass
  import dataclasses
  import json
  import logging
  import os
  import subprocess
  import sys
  import tempfile
  import time
  from pathlib import Path
  from typing import Any, Dict, List
  from unittest import mock
  import pytest
  from ray.runtime_env.runtime_env import RuntimeEnvConfig, _merge_runtime_env
  import requests
  import ray
  from ray._private.runtime_env.context import RuntimeEnvContext
  from ray._private.runtime_env.plugin import RuntimeEnvPlugin
  from ray._private.runtime_env.uri_cache import URICache
  from ray._private.runtime_env.utils import SubprocessCalledProcessError, check_output_cmd
  from ray._private.test_utils import chdir, get_error_message, get_log_sources, wait_for_condition
  from ray._private.utils import get_master_wheel_url, get_release_wheel_url, get_wheel_filename
  from ray.exceptions import RuntimeEnvSetupError
  from ray.runtime_env import RuntimeEnv
  import ray._private.ray_constants
label_desc: Use image node_id 11 for calling a global function with example usage: test_no_spurious_worker_startup(shutdown_only, runtime_env_class) and returns: None, self, int