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:

Constants Layer: Defines paths to artifact directories and Postman collection JSON files
Trigger Functions Layer: Individual functions that invoke Newman with specific collections and environment configurations
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

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment