code: def test_proxy_manager_internal_kv(
      shutdown_only, with_specific_server, monkeypatch
  ):
      """
      Test that proxy manager can use internal kv with and without a
      SpecificServer and that once a SpecificServer is started up, it
      goes through it.
      """
      monkeypatch.setenv("TIMEOUT_FOR_SPECIFIC_SERVER_S", "5")
      pm, free_ports = start_ray_and_proxy_manager(n_ports=2)
      client = "client1"
  
      task_servicer = proxier.RayletServicerProxy(None, pm)
  
      def make_internal_kv_calls():
          response = task_servicer.KVPut(
              ray_client_pb2.KVPutRequest(key=b"key", value=b"val")
          )
          assert isinstance(response, ray_client_pb2.KVPutResponse)
          assert not response.already_exists
  
          response = task_servicer.KVPut(
              ray_client_pb2.KVPutRequest(key=b"key", value=b"val2")
          )
          assert isinstance(response, ray_client_pb2.KVPutResponse)
          assert response.already_exists
  
          response = task_servicer.KVGet(
              ray_client_pb2.KVGetRequest(key=b"key")
          )
          assert isinstance(response, ray_client_pb2.KVGetResponse)
          assert response.value == b"val"
  
          response = task_servicer.KVPut(
              ray_client_pb2.KVPutRequest(
                  key=b"key", value=b"val2", overwrite=True
              )
          )
          assert isinstance(response, ray_client_pb2.KVPutResponse)
          assert response.already_exists
  
          response = task_servicer.KVGet(
              ray_client_pb2.KVGetRequest(key=b"key")
          )
          assert isinstance(response, ray_client_pb2.KVGetResponse)
          assert response.value == b"val2"
  
      with patch(
          "ray.util.client.server.proxier._get_client_id_from_context"
      ) as mock_get_client_id:
          mock_get_client_id.return_value = client
  
          if with_specific_server:
              pm.create_specific_server(client)
              assert pm.start_specific_server(client, JobConfig())
              channel = pm.get_channel(client)
              assert channel is not None
              task_servicer.Init(
                  ray_client_pb2.InitRequest(
                      job_config=pickle.dumps(JobConfig())
                  )
              )
  
              # Mock out the internal kv calls in this process to raise an
              # exception if they're called. This verifies that we are not
              # making any calls in the proxier if there is a SpecificServer
              # started up.
              with patch(
                  "ray.experimental.internal_kv._internal_kv_put"
              ) as mock_put, patch(
                  "ray.experimental.internal_kv._internal_kv_get"
              ) as mock_get, patch(
                  "ray.experimental.internal_kv._internal_kv_initialized"
              ) as mock_initialized:
                  mock_put.side_effect = Exception(
                      "This shouldn't be called!"
                  )
                  mock_get.side_effect = Exception(
                      "This shouldn't be called!"
                  )
                  mock_initialized.side_effect = Exception(
                      "This shouldn't be called!"
                  )
                  make_internal_kv_calls()
          else:
              make_internal_kv_calls()
  
imports: import json
  import os
  import random
  import sys
  import time
  from glob import glob
  from unittest.mock import patch, MagicMock
  from itertools import chain
  import grpc
  import pytest
  import ray
  import ray.core.generated.ray_client_pb2
  import ray.util.client.server.proxier
  from ray._private.ray_constants import REDIS_DEFAULT_PASSWORD
  from ray._private.test_utils import run_string_as_driver, wait_for_condition
  from ray.cloudpickle.compat import pickle
  from ray.job_config import JobConfig
label_desc: Use image node_id 13 for calling a global function with example usage: test_proxy_manager_internal_kv(shutdown_only, with_specific_server, monkeypatch) without return types