code: def test_load_batch(
      printer: Callable,
      client: cx.Client,
      api: str,
      test_s3_path: str,
      load_config: Dict[str, Union[int, float]],
      deploy_timeout: int = None,
      retry_attempts: int = 0,
      api_config_name: str = "cortex_cpu.yaml",
      node_groups: List[str] = [],
  ):
      jobs = load_config["jobs"]
      workers_per_job = load_config["workers_per_job"]
      items_per_job = load_config["items_per_job"]
      batch_size = load_config["batch_size"]
      workload_timeout = load_config["workload_timeout"]
  
      bucket, key = re.match("s3://(.+?)/(.+)", test_s3_path).groups()
      s3 = boto3.client("s3")
      paginator = s3.get_paginator("list_objects_v2")
  
      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
  
      if len(node_groups) > 0:
          api_specs[0]["node_groups"] = node_groups
  
      sample_generator_path = api_dir / "sample_generator.py"
      assert (
          sample_generator_path.exists()
      ), "sample_generator.py must be present for the batch load test"
      sample_generator = load_generator(sample_generator_path)
  
      api_name = api_specs[0]["name"]
      client.deploy(api_spec=api_specs[0])
      api_endpoint = client.get_api(api_name)["endpoint"]
      failed = False
      try:
          assert endpoint_ready(
              client=client, api_name=api_name, timeout=deploy_timeout
          ), f"api {api_name} not ready"
  
          # submit jobs
          printer(f"submitting {jobs} jobs")
          job_specs = []
          for _ in range(jobs):
              for _ in range(retry_attempts + 1):
                  response = request_batch_prediction(
                      client,
                      api_name,
                      item_list=[
                          sample_generator()
                          for _ in range(items_per_job)
                      ],
                      batch_size=batch_size,
                      workers=workers_per_job,
                      config={"dest_s3_dir": test_s3_path},
                  )
                  if response.status_code == HTTPStatus.OK:
                      break
                  time.sleep(1)
              # retries are only required once
              retry_attempts = 0
  
              assert (
                  response.status_code == HTTPStatus.OK
              ), f"status code: got {response.status_code}, expected {HTTPStatus.OK} ({response.text})"
              job_specs.append(response.json())
  
          # wait the jobs to finish
          printer("waiting on the jobs")
          assert jobs_done(
              client,
              api_name,
              [job_spec["job_id"] for job_spec in job_specs],
              workload_timeout,
          ), f"not all jobs succeed in {workload_timeout}s"
  
          # assert jobs
          printer("checking the jobs' responses")
          for job_spec in job_specs:
              job_id: str = job_spec["job_id"]
              job = requests.get(
                  f"{api_endpoint}?jobID={job_id}"
              ).json()
              job_status = job["job_status"]
              job_metrics = job["metrics"]
  
              assert (
                  job_status["batches_in_queue"] == 0
              ), f"there are still batches in queue ({job_status['batches_in_queue']}) for job ID {job_id}"
              assert job_metrics["succeeded"] == math.ceil(
                  items_per_job / batch_size
              )
  
              num_objects = 0
              for page in paginator.paginate(
                  Bucket=bucket, Prefix=os.path.join(key, job_id)
              ):
                  num_objects += len(page["Contents"])
              assert num_objects == 1
  
      except:
          # best effort
          failed = True
          try:
              api_info = client.get_api(api_name)
  
              # only get the last 10 job statuses
              if (
                  "batch_job_statuses" in api_info
                  and len(api_info["batch_job_statuses"]) > 10
              ):
                  api_info["batch_job_statuses"] = api_info[
                      "batch_job_statuses"
                  ][-10:]
  
              printer(json.dumps(api_info, indent=2))
          finally:
              raise
  
      finally:
          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 9 for calling a global function with example usage: test_load_batch(printer, client, api, test_s3_path, load_config, deploy_timeout, retry_attempts, api_config_name, node_groups) without return types