Evaluating API service performance under stress requires realistic user behavior simulation rather than simple, static HTTP flooding. Developers and DevOps
Evaluating API service performance under stress requires realistic user behavior simulation rather than simple, static HTTP flooding. Developers and DevOps engineers need to measure response latencies, identify throughput bottlenecks, verify stateful operation sequences, and test system stability prior to major deployments.
This tutorial provides a complete Python load-testing solution built on Locust. It handles token authentication, performs stateful API operations (such as item creation followed by cleanup), enforces custom response validations using assertions, logs summary metrics, and supports dual execution modes—either driven via the standard locust CLI or executed programmatically inside custom automation pipelines.
Locust uses standard Python code to define user behavior. Under the hood, it uses gevent coroutines (lightweight green threads) to simulate thousands of concurrent users per node without requiring standard OS thread overhead.
+-----------------------------------+
| Locust Environment / Runner |
+-----------------+-----------------+
|
+-------------------------+-------------------------+
| |
+-------v-------+ +-------v-------+
| Virtual User 1| | Virtual User N|
| (APIUser) | | (APIUser) |
+-------+-------+ +-------+-------+
| |
|-- HTTP GET /health ------------------------------>|
|-- HTTP GET /api/v1/items (Validate JSON) -------->| Target API Host
|-- HTTP POST /api/v1/items ----------------------->| (https://api.example.com)
|-- HTTP DELETE /api/v1/items/{id} ---------------->|
The script defines an APIUser class inheriting from HttpUser. During initialization (on_start), each virtual user sets up stateful session headers (such as Authorization tokens). Tasks are decorated with weights (@task(5), @task(3), @task(1)) to establish relative traffic distribution. Response handlers use catch_response=True context managers to evaluate HTTP status codes and response bodies, marking requests as successes or failures explicitly. Finally, lifecycle hooks (@events.test_start, @events.test_stop) provide assertions on performance criteria (like max acceptable failure rates) suitable for CI/CD gates.
Save this script as locustfile.py:
#!/usr/bin/env python3
"""
Load Testing with Locust
========================
A production-ready load testing script featuring dynamic auth, weighted
user journeys, custom failure thresholds, response assertions, and dual-mode
execution (CLI & programmatic API).
"""
import logging
import os
import sys
import time
from typing import Any, Dict, List, Optional
from locust import HttpUser, task, between, events
from locust.env import Environment
# Configure structured logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("locust_suite")
# Configuration via environment variables with safe defaults
TARGET_HOST: str = os.environ.get("LOCUST_HOST", "https://api.example.com")
AUTH_TOKEN: str = os.environ.get("LOCUST_AUTH_TOKEN", "default-bearer-token")
REQUEST_TIMEOUT: float = float(os.environ.get("LOCUST_REQUEST_TIMEOUT", "5.0"))
SPAWN_USERS: int = int(os.environ.get("LOCUST_USERS", "10"))
SPAWN_RATE: int = int(os.environ.get("LOCUST_SPAWN_RATE", "2"))
RUN_TIME_SECONDS: int = int(os.environ.get("LOCUST_RUN_TIME", "30"))
class APIUser(HttpUser):
"""
Simulates an API client executing realistic workload patterns against the host.
"""
host = TARGET_HOST
wait_time = between(1.0, 3.0) # Pause between tasks (in seconds)
def on_start(self) -> None:
"""
Executes when a virtual user spawns. Sets session authorization headers.
"""
self.client.headers.update({
"Authorization": f"Bearer {AUTH_TOKEN}",
"Content-Type": "application/json",
"User-Agent": "LocustLoadTester/1.0",
})
logger.debug("Virtual User initialized with authorization headers.")
@task(5)
def check_health(self) -> None:
"""
High-frequency task: Lightweight endpoint health check.
Weight: 5 (Executes ~50% of the time).
"""
with self.client.get(
"/health",
timeout=REQUEST_TIMEOUT,
catch_response=True,
name="[GET] /health"
) as response:
if response.status_code == 200:
response.success()
else:
response.failure(f"Health check failed with HTTP {response.status_code}")
@task(3)
def list_items(self) -> None:
"""
Medium-frequency task: Query items and perform structural payload validation.
Weight: 3 (Executes ~30% of the time).
"""
with self.client.get(
"/api/v1/items",
timeout=REQUEST_TIMEOUT,
catch_response=True,
name="[GET] /api/v1/items"
) as response:
if response.status_code != 200:
response.failure(f"Failed listing items: HTTP {response.status_code}")
return
try:
payload: dict[str, Any] = response.json()
items: list[dict[str, Any]] = payload.get("data", [])
# Structural assertion on API response schema
if not isinstance(items, list):
response.failure("Malformed JSON: 'data' key must contain a list")
else:
response.success()
except ValueError as err:
response.failure(f"Failed to parse JSON response: {err}")
@task(1)
def create_and_delete_item(self) -> None:
"""
Low-frequency stateful task: Creates a new resource, verifies creation, and cleans it up.
Weight: 1 (Executes ~10% of the time).
"""
item_payload = {
"name": f"test_item_{int(time.time() * 1000)}",
"price": 29.99,
"category": "synthetic-test"
}
# 1. Create Resource
with self.client.post(
"/api/v1/items",
json=item_payload,
timeout=REQUEST_TIMEOUT,
catch_response=True,
name="[POST] /api/v1/items"
) as create_resp:
if create_resp.status_code not in (200, 201):
create_resp.failure(f"Creation failed with status: {create_resp.status_code}")
return
try:
data = create_resp.json()
item_id = data.get("id")
create_resp.success()
except ValueError:
create_resp.failure("Creation response did not return valid JSON")
return
# 2. Clean Up Resource if ID retrieved
if item_id:
with self.client.delete(
f"/api/v1/items/{item_id}",
timeout=REQUEST_TIMEOUT,
catch_response=True,
name="[DELETE] /api/v1/items/{id}"
) as delete_resp:
if delete_resp.status_code in (200, 204):
delete_resp.success()
else:
delete_resp.failure(f"Cleanup failed for ID {item_id}: HTTP {delete_resp.status_code}")
@events.test_start.connect
def on_test_start(environment: Environment, **kwargs: Any) -> None:
"""
Event listener triggered before load testing begins.
"""
logger.info("==========================================")
logger.info("Initializing Locust Load Test")
logger.info(f"Target Host: {environment.host}")
logger.info("==========================================")
@events.test_stop.connect
def on_test_stop(environment: Environment, **kwargs: Any) -> None:
"""
Event listener triggered after load testing finishes. Evaluates threshold metrics.
"""
stats = environment.stats.total
logger.info("==========================================")
logger.info("Load Test Summary")
logger.info(f"Total Requests: {stats.num_requests}")
logger.info(f"Total Failures: {stats.num_failures}")
logger.info(f"Failure Rate: {stats.fail_ratio:.2%}")
logger.info(f"Average RPS: {stats.total_rps:.2f}")
logger.info("==========================================")
# Threshold Check: Fail CI if total failure ratio exceeds 5%
if stats.fail_ratio > 0.05:
logger.error("TEST FAILED: Failure ratio exceeded tolerance (> 5%).")
else:
logger.info("TEST PASSED: Failure ratio within acceptable tolerance.")
def run_programmatically() -> None:
"""
Executes the Locust test programmatically without requiring the CLI runner.
Useful for standalone script execution or custom CI pipelines.
"""
env = Environment(user_classes=[APIUser], host=TARGET_HOST)
runner = env.create_local_runner()
# Start live terminal stats output
env.stats.start_stats_printer()
logger.info(f"Starting test programmatically: {SPAWN_USERS} users @ {SPAWN_RATE} user/s")
runner.start(user_count=SPAWN_USERS, spawn_rate=SPAWN_RATE)
# Let the