code: def test_uri_caching(
      self,
      set_runtime_env_plugins,
      start_cluster,
      uri_cache_size_100_gb,
  ):
      cluster, address = start_cluster
  
      ray.init(address=address)
  
      def reinit():
          ray.shutdown()
          # TODO(architkulkarni): Currently, reinit the driver will generate the same
          # job id. And if we reinit immediately after shutdown, raylet may
          # process new job started before old job finished in some cases. This
          # inconsistency could disorder the URI reference and delete a valid
          # runtime env. We sleep here to walk around this issue.
          time.sleep(5)
          ray.init(address=address)
  
      @ray.remote
      def f():
          return True
  
      # Run a task to trigger runtime_env creation.
      ref1 = f.options(
          runtime_env={
              URI_CACHING_TEST_PLUGIN_NAME: {
                  "uri": "file:///tmp/test_uri_1",
                  "size_bytes": gb_to_bytes(50),
              }
          }
      ).remote()
      ray.get(ref1)
      # Check that the URI was "created on disk".
      print(get_plugin_usage_data())
      wait_for_condition(
          lambda: get_plugin_usage_data()
          == {
              "uris_to_sizes": {
                  "file:///tmp/test_uri_1": gb_to_bytes(50)
              },
              "modify_context_call_count": 1,
              "create_call_count": 1,
          }
      )
  
      # Shutdown ray to stop the worker and remove the runtime_env reference.
      reinit()
  
      # Run a task with a different runtime env.
      ref2 = f.options(
          runtime_env={
              URI_CACHING_TEST_PLUGIN_NAME: {
                  "uri": "file:///tmp/test_uri_2",
                  "size_bytes": gb_to_bytes(51),
              }
          }
      ).remote()
      ray.get(ref2)
      # This should delete the old URI and create a new one, because 50 + 51 > 100
      # and the cache size limit is 100.
      wait_for_condition(
          lambda: get_plugin_usage_data()
          == {
              "uris_to_sizes": {
                  "file:///tmp/test_uri_2": gb_to_bytes(51)
              },
              "modify_context_call_count": 2,
              "create_call_count": 2,
          }
      )
  
      reinit()
  
      # Run a task with the cached runtime env, to check that the runtime env is not
      # created anew.
      ref3 = f.options(
          runtime_env={
              URI_CACHING_TEST_PLUGIN_NAME: {
                  "uri": "file:///tmp/test_uri_2",
                  "size_bytes": gb_to_bytes(51),
              }
          }
      ).remote()
      ray.get(ref3)
      # modify_context should still be called even if create() is not called.
      # Example: for a "conda" plugin, even if the conda env is already created
      # and cached, we still need to call modify_context to add "conda activate" to
      # the RuntimeEnvContext.command_prefix for the worker.
      wait_for_condition(
          lambda: get_plugin_usage_data()
          == {
              "uris_to_sizes": {
                  "file:///tmp/test_uri_2": gb_to_bytes(51)
              },
              "modify_context_call_count": 3,
              "create_call_count": 2,
          }
      )
  
      reinit()
  
      # Run a task with a new runtime env
      ref4 = f.options(
          runtime_env={
              URI_CACHING_TEST_PLUGIN_NAME: {
                  "uri": "file:///tmp/test_uri_3",
                  "size_bytes": gb_to_bytes(10),
              }
          }
      ).remote()
      ray.get(ref4)
      # The last two URIs should still be present in the cache, because 51 + 10 < 100.
      wait_for_condition(
          lambda: get_plugin_usage_data()
          == {
              "uris_to_sizes": {
                  "file:///tmp/test_uri_2": gb_to_bytes(51),
                  "file:///tmp/test_uri_3": gb_to_bytes(10),
              },
              "modify_context_call_count": 4,
              "create_call_count": 3,
          }
      )
  
imports: import asyncio
  import logging
  import os
  from pathlib import Path
  import time
  from unittest import mock
  import tempfile
  import json
  from typing import List
  import pytest
  import ray
  from ray._private.runtime_env.context import RuntimeEnvContext
  from ray._private.runtime_env.plugin import RuntimeEnvPlugin
  from ray._private.test_utils import enable_external_redis, wait_for_condition
  from ray.exceptions import RuntimeEnvSetupError
  from ray.runtime_env.runtime_env import RuntimeEnv
label_desc: Use image node_id 1 for calling the TestGC obj's underlying member method code with example usage: obj.test_uri_caching(set_runtime_env_plugins, start_cluster, uri_cache_size_100_gb) and returns: True