code: def test_task_level_gc(
      runtime_env_disable_URI_cache, ray_start_cluster, option
  ):
      """Tests that task-level working_dir is GC'd when the worker exits."""
  
      cluster = ray_start_cluster
  
      soft_limit_zero = False
      worker_register_timeout = False
      system_config = cluster.list_all_nodes()[
          0
      ]._ray_params._system_config
      if (
          "num_workers_soft_limit" in system_config
          and system_config["num_workers_soft_limit"] == 0
      ):
          soft_limit_zero = True
      if (
          "worker_register_timeout_seconds" in system_config
          and system_config["worker_register_timeout_seconds"] != 0
      ):
          worker_register_timeout = True
  
      @ray.remote
      def f():
          import test_module
  
          test_module.one()
  
      @ray.remote(num_cpus=1)
      class A:
          def check(self):
              import test_module
  
              test_module.one()
  
      if option == "working_dir":
          runtime_env = {"working_dir": S3_PACKAGE_URI}
      else:
          runtime_env = {"py_modules": [S3_PACKAGE_URI]}
  
      # Note: We should set a bigger timeout if downloads the s3 package slowly.
      get_timeout = 10
  
      # Start a task with runtime env
      if worker_register_timeout:
          obj_ref = f.options(runtime_env=runtime_env).remote()
          with pytest.raises(GetTimeoutError):
              ray.get(obj_ref, timeout=get_timeout)
          ray.cancel(obj_ref)
      else:
          ray.get(f.options(runtime_env=runtime_env).remote())
      if soft_limit_zero or worker_register_timeout:
          # Wait for worker exited and local files gced
          wait_for_condition(lambda: check_local_files_gced(cluster))
      else:
          # Local files should not be gced because of an enough soft limit.
          assert not check_local_files_gced(cluster)
  
      # Start a actor with runtime env
      actor = A.options(runtime_env=runtime_env).remote()
      if worker_register_timeout:
          with pytest.raises(GetTimeoutError):
              ray.get(actor.check.remote(), timeout=get_timeout)
      else:
          ray.get(actor.check.remote())
  
      # Kill actor
      ray.kill(actor)
      if soft_limit_zero or worker_register_timeout:
          # Wait for worker exited and local files gced
          wait_for_condition(lambda: check_local_files_gced(cluster))
      else:
          # Local files should not be gced because of an enough soft limit.
          assert not check_local_files_gced(cluster)
  
      # Start a task with runtime env
      if worker_register_timeout:
          obj_ref = f.options(runtime_env=runtime_env).remote()
          with pytest.raises(GetTimeoutError):
              ray.get(obj_ref, timeout=get_timeout)
          ray.cancel(obj_ref)
      else:
          ray.get(f.options(runtime_env=runtime_env).remote())
      if soft_limit_zero or worker_register_timeout:
          # Wait for worker exited and local files gced
          wait_for_condition(lambda: check_local_files_gced(cluster))
      else:
          # Local files should not be gced because of an enough soft limit.
          assert not check_local_files_gced(cluster)
  
file_path: ray/python/ray/tests/test_runtime_env_working_dir_4.py
imports: import os
  from pathlib import Path
  import sys
  import pytest
  from pytest_lazyfixture import lazy_fixture
  import ray
  from ray._private.test_utils import wait_for_condition, check_local_files_gced
  from ray.exceptions import GetTimeoutError
label_desc: Use image node_id 2 for calling a global function with example usage: test_task_level_gc(runtime_env_disable_URI_cache, ray_start_cluster, option) without return types