Implementation:Pytorch Serve Api Utils

Overview

Api_Utils is the API testing framework for TorchServe that orchestrates automated integration tests using Postman collections executed via Newman (the Postman CLI runner). It provides a unified entry point (test_api) that triggers management, inference, workflow, explanation, and HTTPS test suites against running TorchServe instances, supporting both standard REST and KServe (v1 and v2) inference protocols.

Knowledge Sources	Pytorch_Serve
Domains	Testing, API_Design
Last Updated	2026-02-13 18:52 GMT

Description

The api_utils.py module (465 lines) serves as the central test orchestration layer for TorchServe's API integration testing. It defines constants for artifact directories and Postman collection file paths, then provides a series of trigger_* functions that execute Newman against specific Postman collections to validate API behavior.

Key Responsibilities

• Test Orchestration: Coordinates execution of multiple test suites (management, inference, workflow, explanation, HTTPS) in a structured sequence
• KServe Protocol Support: Provides dedicated test triggers for both KServe v1 (trigger_*_kf) and KServe v2 (trigger_*_kfv2) inference protocols
• Postman/Newman Integration: Uses Newman CLI to run Postman collection files, capturing results and exit codes
• Artifact Management: Manages paths to model archives, test data, and Postman collection files used during testing

Architecture

The module follows a layered structure:

1. Constants Layer: Defines paths to artifact directories and Postman collection JSON files
2. Trigger Functions Layer: Individual functions that invoke Newman with specific collections and environment configurations
3. Aggregation Layer: trigger_all() runs the complete test suite; test_api() serves as the CLI entry point

Code Reference

Source Location

File	Lines	Repository
ts_scripts/api_utils.py	L1-465	pytorch/serve

Key Functions

def trigger_management_tests(connection_params):
    """
    Run Postman collection for Management API tests.
    Validates model registration, scaling, description, and unregistration.
    """
    ...

def trigger_inference_tests(connection_params):
    """
    Run Postman collection for Inference API tests.
    Validates prediction endpoints for registered models.
    """
    ...

def trigger_workflow_tests(connection_params):
    """
    Run Postman collection for Workflow API tests.
    Validates workflow DAG registration and execution.
    """
    ...

def trigger_explanation_tests(connection_params):
    """
    Run Postman collection for Explanation API tests.
    Validates Captum-based model explanations endpoint.
    """
    ...

def trigger_https_tests(connection_params):
    """
    Run Postman collection for HTTPS tests.
    Validates TLS-secured endpoint communication.
    """
    ...

KServe Protocol Test Functions

# KServe v1 Protocol triggers
def trigger_management_tests_kf(connection_params):
    """Run management tests using KServe v1 protocol."""
    ...

def trigger_inference_tests_kf(connection_params):
    """Run inference tests using KServe v1 protocol."""
    ...

# KServe v2 Protocol triggers
def trigger_management_tests_kfv2(connection_params):
    """Run management tests using KServe v2 protocol."""
    ...

def trigger_inference_tests_kfv2(connection_params):
    """Run inference tests using KServe v2 protocol."""
    ...

Entry Points

def trigger_all(connection_params):
    """
    Run the complete API test suite.
    Executes all trigger functions in sequence:
    management -> inference -> workflow -> explanation -> HTTPS.

    Args:
        connection_params: Connection configuration for the TorchServe instance.

    Returns:
        int: Exit code (0 for success, non-zero for failure).
    """
    ...

def test_api():
    """
    CLI entry point for the API testing framework.
    Parses command-line arguments, configures connection parameters,
    and dispatches to the appropriate trigger function(s).
    """
    ...

Import

# Typically invoked as a script rather than imported as a module:
# python ts_scripts/api_utils.py [OPTIONS]

# When imported:
from ts_scripts.api_utils import trigger_all, test_api

I/O Contract

Function	Input	Output	Notes
test_api()	CLI arguments (test type, connection params)	Exit code (0 = pass, non-zero = fail)	Main entry point; parses sys.argv
trigger_all()	connection_params dict	Aggregated exit code	Runs all test suites sequentially
trigger_management_tests()	connection_params dict	Newman exit code	Executes management Postman collection
trigger_inference_tests()	connection_params dict	Newman exit code	Executes inference Postman collection
trigger_workflow_tests()	connection_params dict	Newman exit code	Executes workflow Postman collection
trigger_explanation_tests()	connection_params dict	Newman exit code	Executes explanation Postman collection
trigger_https_tests()	connection_params dict	Newman exit code	Executes HTTPS Postman collection

Constants

Constant	Description
Artifact directories	Paths to directories containing model archives (.mar files) and test data
Postman collection files	Paths to .json Postman collection files for each test suite (management, inference, workflow, explanation, HTTPS)
KServe collection files	Separate Postman collections for KServe v1 and v2 protocol testing

Usage Examples

Example 1: Run the complete test suite from CLI

# Run all API tests against a local TorchServe instance
python ts_scripts/api_utils.py --all

# Run only management tests
python ts_scripts/api_utils.py --management

# Run only inference tests
python ts_scripts/api_utils.py --inference

Example 2: Run KServe v2 protocol tests

# Run KServe v2 management and inference tests
python ts_scripts/api_utils.py --management-kfv2
python ts_scripts/api_utils.py --inference-kfv2

Example 3: Programmatic invocation

from ts_scripts.api_utils import trigger_management_tests, trigger_inference_tests

# Define connection parameters for TorchServe
connection_params = {
    "management_port": 8081,
    "inference_port": 8080,
    "host": "localhost",
}

# Run management tests
mgmt_result = trigger_management_tests(connection_params)
print(f"Management tests exit code: {mgmt_result}")

# Run inference tests
infer_result = trigger_inference_tests(connection_params)
print(f"Inference tests exit code: {infer_result}")

Example 4: Test execution flow

# Internal execution flow of trigger_management_tests:
# 1. Construct Newman CLI command with collection path and environment variables
# 2. Execute Newman as a subprocess
# 3. Capture stdout/stderr and exit code
# 4. Return exit code (0 = all assertions passed, non-zero = failures)

# Newman command pattern:
# newman run <collection.json> -e <environment.json> --reporters cli,json

Related Pages

• Principle:Pytorch_Serve_API_Integration_Testing - The principle of automated API integration testing for model serving
• Implementation:Pytorch_Serve_Management_API - Management API endpoints validated by the management test suite
• Implementation:Pytorch_Serve_Install_Dependencies_Script - Installs Newman and Node.js required for test execution
• Implementation:Pytorch_Serve_Model_Server_Start - Server must be running before API tests can execute