code: def test_redis_cleanup(redis_replicas, shutdown_only):
      addr = ray.init(
          namespace="a",
          _system_config={"external_storage_namespace": "c1"},
      ).address_info["address"]
      gcs_client = GcsClient(address=addr)
      gcs_client.internal_kv_put(b"ABC", b"DEF", True, None)
  
      ray.shutdown()
      addr = ray.init(
          namespace="a",
          _system_config={"external_storage_namespace": "c2"},
      ).address_info["address"]
      gcs_client = GcsClient(address=addr)
      gcs_client.internal_kv_put(b"ABC", b"XYZ", True, None)
      ray.shutdown()
      redis_addr = os.environ["RAY_REDIS_ADDRESS"]
      host, port = redis_addr.split(":")
      if os.environ.get("TEST_EXTERNAL_REDIS_REPLICAS", "1") != "1":
          cli = redis.RedisCluster(host, int(port))
      else:
          cli = redis.Redis(host, int(port))
  
      assert set(cli.keys()) == {b"c1", b"c2"}
      gcs_utils.cleanup_redis_storage(host, int(port), "", False, "c1")
      assert set(cli.keys()) == {b"c2"}
      gcs_utils.cleanup_redis_storage(host, int(port), "", False, "c2")
      assert len(cli.keys()) == 0
  
imports: import contextlib
  import os
  import time
  import signal
  import sys
  import asyncio
  import async_timeout
  import pytest
  import ray
  import redis
  from ray._raylet import GcsClient
  import ray._private.gcs_utils
  from ray._private.test_utils import enable_external_redis, find_free_port, generate_system_config_map, async_wait_for_condition_async_predicate
  import ray._private.ray_constants
label_desc: Use image node_id 6 for calling a global function with example usage: test_redis_cleanup(redis_replicas, shutdown_only) without return types