code: def test_async_api(
      printer: Callable,
      client: cx.Client,
      api: str,
      deploy_timeout: int = None,
      poll_retries: int = 5,
      poll_sleep_seconds: int = 1,
      api_config_name: str = "cortex_cpu.yaml",
      node_groups: List[str] = [],
  ):
      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
  
      expectations = None
      expectations_file = api_dir / "expectations.yaml"
      if expectations_file.exists():
          expectations = parse_expectations(str(expectations_file))
  
      api_name = api_specs[0]["name"]
      client.deploy(api_spec=api_specs[0])
  
      try:
          assert apis_ready(
              client=client,
              api_names=[api_name],
              timeout=deploy_timeout,
          ), f"apis {api_name} not ready"
  
          with open(str(api_dir / "sample.json")) as f:
              payload = json.load(f)
  
          response = post_request(client, api_name, payload)
  
          assert (
              response.status_code == HTTPStatus.OK
          ), f"workload submission status code: got {response.status_code}, expected {HTTPStatus.OK}"
  
          response_json = response.json()
          assert "id" in response_json
  
          request_id = response_json["id"]
  
          result_response = None
          for _ in range(poll_retries + 1):
              result_response = retrieve_async_result(
                  client=client,
                  api_name=api_name,
                  request_id=request_id,
              )
  
              if result_response.status_code != HTTPStatus.OK:
                  time.sleep(poll_sleep_seconds)
                  continue
  
              result_response_json = result_response.json()
              assert (
                  "id" in result_response_json
              ), f"id key was not present in result response (response: {result_response_json})"
              assert (
                  "status" in result_response_json
              ), f"status key was not present in result response (response: {result_response_json})"
  
              if result_response_json["status"] != "completed":
                  time.sleep(poll_sleep_seconds)
                  continue
              break
  
          assert (
              result_response.status_code == HTTPStatus.OK
          ), f"result retrieval status code: got {result_response.status_code}, expected {HTTPStatus.OK}"
  
          # validate keys are in the result json response
          assert (
              "result" in result_response_json
          ), f"result key was not present in result response (response: {result_response_json})"
  
          assert (
              "timestamp" in result_response_json
          ), f"timestamp key was not present in result response (response: {result_response_json})"
  
          # validate result json response has valid values
          assert (
              result_response_json["id"] == request_id
          ), f"result 'id' and request 'id' mismatch ({result_response_json['id']} != {request_id})"
          assert (
              result_response_json["status"] == "completed"
          ), f"async workload did not complete (response: {result_response_json})"
          assert (
              result_response_json["timestamp"] != ""
          ), "result 'timestamp' value was empty"
          assert (
              result_response_json["result"] != ""
          ), "result 'result' value was empty"
  
          # assert result expectations
          if expectations:
              assert_json_expectations(
                  result_response_json["result"],
                  expectations["response"],
              )
  
      except:
          # best effort
          try:
              api_info = client.get_api(api_name)
              printer(json.dumps(api_info, indent=2))
              printer(json.dumps(result_response_json, indent=2))
              td.Thread(
                  target=lambda: stream_api_logs(client, api_name),
                  daemon=True,
              ).start()
              time.sleep(5)
          except:
              pass
          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 4 for calling a global function with example usage: test_async_api(printer, client, api, deploy_timeout, poll_retries, poll_sleep_seconds, api_config_name, node_groups) without return types