code: def test_load_realtime(
      printer: Callable,
      client: cx.Client,
      api: str,
      load_config: Dict[str, Union[int, float]],
      deploy_timeout: int = None,
      api_config_name: str = "cortex_cpu.yaml",
      node_groups: List[str] = [],
  ):
      total_requests = load_config["total_requests"]
      desired_replicas = load_config["desired_replicas"]
      concurrency = load_config["concurrency"]
      status_code_timeout = load_config["status_code_timeout"]
  
      api_dir = TEST_APIS_DIR / api
      with open(str(api_dir / api_config_name)) as f:
          api_specs = yaml.safe_load(f)
      assert len(api_specs) == 1
  
      api_specs[0]["autoscaling"] = {
          "min_replicas": desired_replicas,
          "max_replicas": desired_replicas,
      }
      if len(node_groups) > 0:
          api_specs[0]["node_groups"] = node_groups
  
      api_name = api_specs[0]["name"]
      client.deploy(api_spec=api_specs[0])
  
      # controls the flow of requests
      request_stopper = td.Event()
      failed = False
      try:
          printer(f"getting {desired_replicas} replicas ready")
          assert apis_ready(
              client=client,
              api_names=[api_name],
              timeout=deploy_timeout,
          ), f"api {api_name} not ready"
  
          # give the APIs some time to prevent getting high latency spikes in the beginning
          time.sleep(5)
  
          with open(str(api_dir / "sample.json")) as f:
              payload = json.load(f)
  
          printer("start making requests concurrently")
          threads_futures = make_requests_concurrently(
              client,
              api_name,
              concurrency,
              request_stopper,
              max_total_requests=total_requests,
              payload=payload,
          )
  
          while not request_stopper.is_set():
              # check if the requesting threads are still healthy
              # if not, they'll raise an exception
              check_futures_healthy(threads_futures)
  
              # don't stress the CPU too hard
              time.sleep(1)
      except:
          # best effort
          failed = True
          try:
              api_info = client.get_api(api_name)
              printer(json.dumps(api_info, indent=2))
          finally:
              raise
  
      finally:
          request_stopper.set()
          delete_apis(client, [api_name])
          if failed:
              time.sleep(30)
  
imports: import json
  import math
  import os
  import re
  import threading
  import time
  from http import HTTPStatus
  from pathlib import Path
  from typing import Callable, Dict, Any, List, Union
  import boto3
  import cortex
  import requests
  import yaml
  from e2e.expectations import parse_expectations, assert_response_expectations, assert_json_expectations
  from e2e.generator import load_generator
  from e2e.utils import apis_ready, api_updated, wait_on_event, wait_on_futures, endpoint_ready, post_request, get_request, job_done, jobs_done, request_batch_prediction, request_task, retrieve_async_result, make_requests_concurrently, check_futures_healthy, retrieve_results_concurrently, stream_api_logs, stream_job_logs, wait_for
label_desc: Use image node_id 7 for calling a global function with example usage: test_load_realtime(printer, client, api, load_config, deploy_timeout, api_config_name, node_groups) without return types