code: def test_batch_api(
      printer: Callable,
      client: cx.Client,
      api: str,
      test_s3_path: str,
      deploy_timeout: int = None,
      job_timeout: int = None,
      retry_attempts: int = 0,
      api_config_name: str = "cortex_cpu.yaml",
      node_groups: List[str] = [],
      local_operator: bool = False,
  ):
      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
  
      api_name = api_specs[0]["name"]
      client.deploy(api_spec=api_specs[0])
  
      try:
          endpoint_override = (
              f"http://localhost:8888/batch/{api_name}"
              if local_operator
              else None
          )
          assert endpoint_ready(
              client=client,
              api_name=api_name,
              timeout=deploy_timeout,
              endpoint_override=endpoint_override,
          ), f"api {api_name} not ready"
  
          with open(str(api_dir / "sample.json")) as f:
              payload = json.load(f)
  
          response = None
          for _ in range(retry_attempts + 1):
              response = request_batch_prediction(
                  client,
                  api_name,
                  item_list=payload,
                  batch_size=2,
                  config={"dest_s3_dir": test_s3_path},
                  local_operator=local_operator,
              )
              if response.status_code == HTTPStatus.OK:
                  break
  
              time.sleep(1)
  
          assert (
              response.status_code == HTTPStatus.OK
          ), f"status code: got {response.status_code}, expected {HTTPStatus.OK} ({response.text})"
  
          job_spec = response.json()
  
          # monitor job progress
          job_id = job_spec["job_id"]
          endpoint_override = (
              f"http://localhost:8888/batch/{api_name}?jobID={job_id}"
              if local_operator
              else None
          )
          assert job_done(
              client=client,
              api_name=api_name,
              job_id=job_id,
              timeout=job_timeout,
              endpoint_override=endpoint_override,
          ), f"job did not succeed (api_name: {api_name}, job_id: {job_spec['job_id']})"
  
      except:
          # best effort
          try:
              api_info = client.get_api(api_name)
              printer(json.dumps(api_info, indent=2))
  
              job_status = client.get_job(api_name, job_spec["job_id"])
              printer(json.dumps(job_status, indent=2))
  
              td.Thread(
                  target=lambda: stream_job_logs(
                      client, api_name, job_spec["job_id"]
                  ),
                  daemon=True,
              ).start()
              time.sleep(5)
          finally:
              raise
      finally:
          delete_apis(client, [api_name])
  
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 3 for calling a global function with example usage: test_batch_api(printer, client, api, test_s3_path, deploy_timeout, job_timeout, retry_attempts, api_config_name, node_groups, local_operator) without return types