code: def test_kv_basic(ray_start_regular, monkeypatch):
      monkeypatch.setenv("TEST_RAY_COLLECT_KV_FREQUENCY", "1")
      gcs_address = ray._private.worker.global_worker.gcs_client.address
      gcs_client = ray._raylet.GcsClient(
          address=gcs_address, nums_reconnect_retry=0
      )
      # Wait until all other calls finished
      time.sleep(2)
      # reset the counter
      ray._private.utils._CALLED_FREQ.clear()
      assert gcs_client.internal_kv_get(b"A", b"NS") is None
      assert gcs_client.internal_kv_put(b"A", b"B", False, b"NS") == 1
      assert gcs_client.internal_kv_get(b"A", b"NS") == b"B"
      assert gcs_client.internal_kv_put(b"A", b"C", False, b"NS") == 0
      assert gcs_client.internal_kv_get(b"A", b"NS") == b"B"
      assert gcs_client.internal_kv_put(b"A", b"C", True, b"NS") == 0
      assert gcs_client.internal_kv_get(b"A", b"NS") == b"C"
      assert gcs_client.internal_kv_put(b"AA", b"B", False, b"NS") == 1
      assert gcs_client.internal_kv_put(b"AB", b"B", False, b"NS") == 1
      assert set(gcs_client.internal_kv_keys(b"A", b"NS")) == {
          b"A",
          b"AA",
          b"AB",
      }
      assert gcs_client.internal_kv_del(b"A", False, b"NS") == 1
      assert set(gcs_client.internal_kv_keys(b"A", b"NS")) == {
          b"AA",
          b"AB",
      }
      assert gcs_client.internal_kv_keys(b"A", b"NSS") == []
      assert gcs_client.internal_kv_del(b"A", True, b"NS") == 2
      assert gcs_client.internal_kv_keys(b"A", b"NS") == []
      assert gcs_client.internal_kv_del(b"A", False, b"NSS") == 0
      assert ray._private.utils._CALLED_FREQ["internal_kv_get"] == 4
      assert ray._private.utils._CALLED_FREQ["internal_kv_put"] == 5
  
      # Test internal_kv_multi_get
      assert gcs_client.internal_kv_multi_get([b"A", b"B"], b"NS") == {}
      assert gcs_client.internal_kv_put(b"A", b"B", False, b"NS") == 1
      assert gcs_client.internal_kv_put(b"B", b"C", False, b"NS") == 1
      assert gcs_client.internal_kv_multi_get([b"A", b"B"], b"NS") == {
          b"A": b"B",
          b"B": b"C",
      }
      assert (
          gcs_client.internal_kv_multi_get([b"A", b"B"], b"NSS") == {}
      )
  
      # Test internal_kv_multi_get where some keys don't exist
      assert gcs_client.internal_kv_multi_get(
          [b"A", b"B", b"C"], b"NS"
      ) == {
          b"A": b"B",
          b"B": b"C",
      }
  
      assert (
          ray._private.utils._CALLED_FREQ["internal_kv_multi_get"] == 4
      )
  
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 2 for calling a global function with example usage: test_kv_basic(ray_start_regular, monkeypatch) without return types