docstr: Tests that task-level working_dir is GC'd when the task exits.
docstr_tok: ["Tests","that","task-level","working_dir","is","GC","'d","when","the","task","exits","."]
code: def test_task_level_gc(
      self,
      runtime_env_disable_URI_cache,
      ray_start_cluster,
      field,
      spec_format,
      tmp_path,
  ):
      """Tests that task-level working_dir is GC'd when the task exits."""
  
      cluster = ray_start_cluster
  
      soft_limit_zero = 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
  
      runtime_env = generate_runtime_env_dict(
          field, spec_format, tmp_path
      )
  
      @ray.remote
      def f():
          import pip_install_test  # noqa: F401
  
          return True
  
      @ray.remote
      class A:
          def test_import(self):
              import pip_install_test  # noqa: F401
  
              return True
  
      # Start a task with runtime env
      ray.get(f.options(runtime_env=runtime_env).remote())
      if soft_limit_zero:
          # 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()
      ray.get(actor.test_import.remote())
      # Local files should not be gced
      assert not check_local_files_gced(cluster)
  
      # Kill actor
      ray.kill(actor)
      if soft_limit_zero:
          # 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
      ray.get(f.options(runtime_env=runtime_env).remote())
      if soft_limit_zero:
          # 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_conda_and_pip_3.py
imports: import os
  import pytest
  import sys
  import time
  from ray._private.test_utils import wait_for_condition, check_local_files_gced, generate_runtime_env_dict
  import ray
  from unittest import mock
label_desc: Use image node_id 2 for calling the TestGC obj's underlying member method code with example usage: obj.test_task_level_gc(runtime_env_disable_URI_cache, ray_start_cluster, field, spec_format, tmp_path) and returns: True, True