code: def test_temporary_uri_reference(monkeypatch, expiration_s):
      """Test that temporary GCS URI references are deleted after expiration_s."""
      monkeypatch.setenv(
          "RAY_RUNTIME_ENV_TEMPORARY_REFERENCE_EXPIRATION_S",
          str(expiration_s),
      )
      # We can't use a fixture with a shared Ray runtime because we need to set the
      # expiration_s env var before Ray starts.
      with _ray_start(include_dashboard=True, num_cpus=1) as ctx:
          headers = {
              "Connection": "keep-alive",
              "Authorization": "TOK:<MY_TOKEN>",
          }
          address = ctx.address_info["webui_url"]
          assert wait_until_server_available(address)
          client = JobSubmissionClient(
              format_web_url(address), headers=headers
          )
          with tempfile.TemporaryDirectory() as tmp_dir:
              path = Path(tmp_dir)
  
              hello_file = path / "hi.txt"
              with hello_file.open(mode="w") as f:
                  f.write("hi\n")
  
              start = time.time()
  
              client.submit_job(
                  entrypoint="echo hi",
                  runtime_env={"working_dir": tmp_dir},
              )
  
              # Give time for deletion to occur if expiration_s is 0.
              time.sleep(2)
              # Need to connect to Ray to check internal_kv.
              # ray.init(address="auto")
  
              print(
                  "Starting Internal KV checks at time ",
                  time.time() - start,
              )
              if expiration_s > 0:
                  assert not check_internal_kv_gced()
                  wait_for_condition(
                      check_internal_kv_gced, timeout=2 * expiration_s
                  )
                  assert (
                      expiration_s
                      < time.time() - start
                      < 2 * expiration_s
                  )
                  print(
                      "Internal KV was GC'ed at time ",
                      time.time() - start,
                  )
              else:
                  wait_for_condition(check_internal_kv_gced)
                  print(
                      "Internal KV was GC'ed at time ",
                      time.time() - start,
                  )
  
imports: from pathlib import Path
  import tempfile
  import time
  import os
  from ray.dashboard.modules.job.pydantic_models import JobType
  import requests
  import pytest
  import psutil
  import sys
  from typing import Dict, Optional, Tuple
  from unittest.mock import Mock, patch
  from ray._private.ray_constants import DEFAULT_DASHBOARD_AGENT_LISTEN_PORT
  from ray._private.test_utils import format_web_url, wait_for_condition, wait_until_server_available
  from ray.dashboard.modules.dashboard_sdk import ClusterInfo, DEFAULT_DASHBOARD_ADDRESS, parse_cluster_info
  from ray.dashboard.modules.job.sdk import JobSubmissionClient, JobStatus
  from ray.dashboard.tests.conftest import *
  from ray.dashboard.consts import RAY_JOB_ALLOW_DRIVER_ON_WORKER_NODES_ENV_VAR
  from ray.tests.conftest import _ray_start
  import ray
  import ray.experimental.internal_kv
  from ray.util.state import list_nodes
label_desc: Use image node_id 5 for calling a global function with example usage: test_temporary_uri_reference(monkeypatch, expiration_s) without return types