docstr: Tests that URIs for detached actors are GC'd only when they exit.
docstr_tok: ["Tests","that","URIs","for","detached","actors","are","GC","'d","only","when","they","exit","."]
code: def test_detached_actor_gc(
      self,
      start_cluster,
      working_dir_and_pymodules_disable_URI_cache,
      disable_temporary_uri_pinning,
      option: str,
      source: str,
  ):
      """Tests that URIs for detached actors are GC'd only when they exit."""
      cluster, address = start_cluster
      # Wait until agent is ready.
      # TODO(sang): There's a bug where is runtime env creation request is
      # sent before agent is ready, it fails. We will fix this issue soon.
      time.sleep(2)
  
      if option == "working_dir":
          ray.init(
              address,
              namespace="test",
              runtime_env={"working_dir": source},
          )
      elif option == "py_modules":
          if source != S3_PACKAGE_URI:
              source = str(Path(source) / "test_module")
          ray.init(
              address,
              namespace="test",
              runtime_env={
                  "py_modules": [
                      source,
                      Path(os.path.dirname(__file__))
                      / "pip_install_test-0.5-py3-none-any.whl",
                  ]
              },
          )
      print(f'Initialized Ray with option "{option}".')
  
      # For a local directory, the package should be in the GCS.
      # For an S3 URI, there should be nothing in the GCS because
      # it will be downloaded from S3 directly on each node.
      # In the "py_modules" case, a local wheel file will be in the GCS.
      if source == S3_PACKAGE_URI and option != "py_modules":
          assert check_internal_kv_gced()
      else:
          assert not check_internal_kv_gced()
      print(
          f'kv check 1 passed with source "{source}" and option "{option}".'
      )
  
      @ray.remote
      class A:
          def test_import(self):
              import test_module
  
              if option == "py_modules":
                  import pip_install_test  # noqa: F401
              test_module.one()
  
      a = A.options(name="test", lifetime="detached").remote()
      print('Created detached actor with name "test".')
  
      ray.get(a.test_import.remote())
      print('Got response from "test" actor.')
  
      if source == S3_PACKAGE_URI and option != "py_modules":
          assert check_internal_kv_gced()
      else:
          assert not check_internal_kv_gced()
      print(
          f'kv check 2 passed with source "{source}" and option "{option}".'
      )
  
      assert not check_local_files_gced(cluster)
      print("check_local_files_gced() check passed.")
  
      ray.shutdown()
      print("Ray has been shut down.")
  
      ray.init(address, namespace="test")
      print(
          f'Reconnected to Ray at address "{address}" and namespace "test".'
      )
  
      if source == S3_PACKAGE_URI and option != "py_modules":
          assert check_internal_kv_gced()
      else:
          assert not check_internal_kv_gced()
      print(
          f'kv check 3 passed with source "{source}" and option "{option}".'
      )
  
      assert not check_local_files_gced(cluster)
      print("check_local_files_gced() check passed.")
  
      a = ray.get_actor("test")
      print('Got "test" actor.')
  
      ray.get(a.test_import.remote())
      print('Got response from "test" actor.')
  
      ray.kill(a)
      print('Issued ray.kill() request to "test" actor.')
  
      wait_for_condition(check_internal_kv_gced)
      print("check_internal_kv_gced passed wait_for_condition block.")
  
      whitelist = get_local_file_whitelist(cluster, option)
      wait_for_condition(
          lambda: check_local_files_gced(cluster, whitelist=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 3 for calling the TestGC obj's underlying member method code with example usage: obj.test_detached_actor_gc(start_cluster, working_dir_and_pymodules_disable_URI_cache, disable_temporary_uri_pinning, option, source) without return types