docstr: Tests that actor-level working_dir is GC'd when the actor exits.
docstr_tok: ["Tests","that","actor-level","working_dir","is","GC","'d","when","the","actor","exits","."]
code: def test_actor_level_gc(
      self,
      start_cluster,
      working_dir_and_pymodules_disable_URI_cache,
      disable_temporary_uri_pinning,
      option: str,
  ):
      """Tests that actor-level working_dir is GC'd when the actor exits."""
      NUM_NODES = 5
      cluster, address = start_cluster
      for i in range(NUM_NODES - 1):  # Head node already added.
          cluster.add_node(
              num_cpus=1,
              runtime_env_dir_name=f"node_{i}_runtime_resources",
          )
          print(
              f'Added node with runtime_env_dir_name "node_{i}_runtime_resources".'
          )
  
      print(f"Added all {NUM_NODES} nodes.")
  
      ray.init(address)
      print(f'Initialized Ray at address "{address}".')
  
      @ray.remote(num_cpus=1)
      class A:
          def check(self):
              import test_module
  
              test_module.one()
  
      if option == "working_dir":
          A = A.options(runtime_env={"working_dir": S3_PACKAGE_URI})
      else:
          A = A.options(
              runtime_env={
                  "py_modules": [
                      S3_PACKAGE_URI,
                  ]
              }
          )
      print(f'Created deployment A with option "{option}".')
  
      num_cpus = int(ray.available_resources()["CPU"])
      print(f"{num_cpus} cpus available.")
  
      actors = [A.remote() for _ in range(num_cpus)]
      print(f"Created {len(actors)} actors.")
  
      ray.get([a.check.remote() for a in actors])
      print("Got responses from all actors.")
  
      for i in range(num_cpus):
          assert not check_local_files_gced(cluster)
          print(f"check_local_files_gced assertion passed for cpu {i}.")
  
          ray.kill(actors[i])
          print(f"Issued ray.kill for actor {i}.")
  
      whitelist = get_local_file_whitelist(cluster, option)
      wait_for_condition(
          lambda: check_local_files_gced(cluster, whitelist)
      )
      print("check_local_files_gced passed wait_for_condition block.")
  
imports: import os
  import sys
  import tempfile
  import time
  from pathlib import Path
  from unittest import mock
  import pytest
  from pytest_lazyfixture import lazy_fixture
  import ray
  import ray.experimental.internal_kv
  from ray._private.ray_constants import RAY_RUNTIME_ENV_URI_PIN_EXPIRATION_S_ENV_VAR
  from ray._private.utils import get_directory_size_bytes
  from ray._private.test_utils import chdir, check_local_files_gced, wait_for_condition, find_free_port
label_desc: Use image node_id 2 for calling the TestGC obj's underlying member method code with example usage: obj.test_actor_level_gc(start_cluster, working_dir_and_pymodules_disable_URI_cache, disable_temporary_uri_pinning, option) without return types