tessl/pypi-locust
Developer-friendly load testing framework for HTTP and other protocols with distributed testing capabilities.
—
Pending
user-classes.mddocs/
User Classes
User classes define the behavior patterns for virtual users in load tests. Locust provides three main user classes: the base User class, HttpUser for HTTP testing, and FastHttpUser for high-performance HTTP testing.
Capabilities
Base User Class
The abstract base class that all user classes inherit from. Defines the core user lifecycle and task execution framework.
class User:
"""
Base class for all user classes.
Attributes:
host (str): Target host for requests
weight (int): Relative weight for user selection in load tests
fixed_count (int): Fixed number of users (overrides weight-based selection)
tasks (list | dict): Tasks to execute - list of functions or dict with weights
wait_time (callable): Function returning wait time between tasks
abstract (bool): Mark class as abstract (won't be executed)
"""
def __init__(self, environment):
"""
Initialize user with environment.
Args:
environment: Locust environment instance
"""
def on_start(self):
"""Called when user starts executing."""
def on_stop(self):
"""Called when user stops executing."""
def run(self):
"""Main execution loop - runs tasks with wait times."""
def wait(self):
"""Wait between tasks using wait_time function."""
def start(self, group):
"""Start user execution in a greenlet group."""
def stop(self, force=False):
"""
Stop user execution.
Args:
force (bool): Force immediate stop without cleanup
"""
def context(self, **kwargs):
"""
Context manager for temporary user configuration.
Returns:
Context manager for user settings
"""
@property
def group(self):
"""Group this user belongs to."""
@property
def greenlet(self):
"""Greenlet instance running this user."""HttpUser Class
HTTP-specific user class that provides an HTTP client for making web requests. Built on the requests library with Locust-specific enhancements.
class HttpUser(User):
"""
User class for HTTP load testing using requests library.
Attributes:
client (HttpSession): HTTP session for making requests
pool_manager: Connection pool manager
"""
# Inherits all User methods and attributes
# client is automatically created as HttpSession instanceHttpSession
The HTTP client used by HttpUser, extending requests.Session with load testing features.
class HttpSession:
"""
Extended requests.Session with Locust event integration.
Attributes:
base_url (str): Base URL for all requests
request_event: Event hook for request statistics
user: Associated user instance
request_name (str): Current request name for statistics
"""
def __init__(self, base_url, request_event, user, *args, pool_manager=None, **kwargs):
"""
Initialize HTTP session.
Args:
base_url (str): Base URL for requests
request_event: Event hook for statistics
user: User instance
pool_manager: Optional connection pool manager
"""
def get(self, url, **kwargs):
"""
Send GET request.
Args:
url (str): Request URL
**kwargs: Additional requests parameters
Returns:
LocustResponse: Enhanced response object
"""
def post(self, url, data=None, json=None, **kwargs):
"""
Send POST request.
Args:
url (str): Request URL
data: Request body data
json: JSON data to send
**kwargs: Additional requests parameters
Returns:
LocustResponse: Enhanced response object
"""
def put(self, url, data=None, **kwargs):
"""Send PUT request."""
def patch(self, url, data=None, **kwargs):
"""Send PATCH request."""
def delete(self, url, **kwargs):
"""Send DELETE request."""
def head(self, url, **kwargs):
"""Send HEAD request."""
def options(self, url, **kwargs):
"""Send OPTIONS request."""
def request(self, method, url, **kwargs):
"""
Send HTTP request with specified method.
Args:
method (str): HTTP method
url (str): Request URL
**kwargs: Additional requests parameters
Returns:
LocustResponse: Enhanced response object
"""
def rename_request(self, name):
"""
Context manager to rename request for statistics.
Args:
name (str): Name to use in statistics
Returns:
Context manager
"""FastHttpUser Class
High-performance HTTP user class using geventhttpclient instead of requests for better performance with high concurrency.
class FastHttpUser(User):
"""
High-performance HTTP user using geventhttpclient.
Configuration Attributes:
network_timeout (float): Network timeout in seconds
connection_timeout (float): Connection timeout in seconds
max_redirects (int): Maximum redirects to follow
max_retries (int): Maximum retry attempts
insecure (bool): Skip SSL certificate verification
default_headers (dict): Default headers for all requests
concurrency (int): Connection concurrency level
proxy_host (str): Proxy hostname
proxy_port (int): Proxy port
client_pool: Connection pool configuration
ssl_context_factory: Custom SSL context factory
Attributes:
client (FastHttpSession): Fast HTTP session for requests
"""
# Inherits all User methods and attributes
# client is automatically created as FastHttpSession instance
def rest(self, method, url, **kwargs):
"""
Make REST API request with automatic JSON handling.
Args:
method (str): HTTP method
url (str): Request URL
**kwargs: Request parameters
Returns:
Parsed JSON response
"""
def rest_(self, method, url, **kwargs):
"""
Make REST API request returning response context manager.
Args:
method (str): HTTP method
url (str): Request URL
**kwargs: Request parameters
Returns:
RestResponseContextManager: Context for response validation
"""Usage Examples
Basic HttpUser Example
from locust import HttpUser, task, between
class WebsiteUser(HttpUser):
wait_time = between(1, 3)
def on_start(self):
# Login when user starts
response = self.client.post("/login", json={
"username": "testuser",
"password": "secret"
})
# Store auth token if needed
self.auth_token = response.json()["token"]
@task(3)
def browse_products(self):
# Get product list
self.client.get("/products")
# View random product
product_id = random.randint(1, 100)
self.client.get(f"/products/{product_id}")
@task(1)
def add_to_cart(self):
self.client.post("/cart", json={
"product_id": random.randint(1, 100),
"quantity": random.randint(1, 5)
})
def on_stop(self):
# Cleanup when user stops
self.client.post("/logout")FastHttpUser Example
from locust.contrib.fasthttp import FastHttpUser
from locust import task, between
class HighPerformanceUser(FastHttpUser):
wait_time = between(0.1, 0.5) # Very fast requests
# Configure for high performance
connection_timeout = 60.0
network_timeout = 60.0
concurrency = 10
@task
def api_endpoint(self):
# Use REST helper for JSON APIs
data = self.rest("GET", "/api/data")
# Traditional request
response = self.client.get("/api/status")
# Validate with context manager
with self.rest_("POST", "/api/process", json={"data": data}) as response:
if response.js["status"] != "success":
response.failure("Processing failed")Custom User Base Class
from locust import User, task, events
import time
class CustomProtocolUser(User):
"""User for testing custom protocols."""
abstract = True # Don't run this class directly
def __init__(self, environment):
super().__init__(environment)
# Initialize custom client
self.client = CustomClient(host=self.host)
def on_start(self):
# Connect to custom service
start_time = time.time()
try:
self.client.connect()
# Fire request event for statistics
events.request.fire(
request_type="CONNECT",
name="connect",
response_time=(time.time() - start_time) * 1000,
response_length=0
)
except Exception as e:
events.request.fire(
request_type="CONNECT",
name="connect",
response_time=(time.time() - start_time) * 1000,
response_length=0,
exception=e
)
def on_stop(self):
self.client.disconnect()
class MyCustomUser(CustomProtocolUser):
wait_time = between(1, 2)
@task
def custom_operation(self):
start_time = time.time()
try:
result = self.client.send_command("GET_DATA")
events.request.fire(
request_type="CUSTOM",
name="get_data",
response_time=(time.time() - start_time) * 1000,
response_length=len(result)
)
except Exception as e:
events.request.fire(
request_type="CUSTOM",
name="get_data",
response_time=(time.time() - start_time) * 1000,
response_length=0,
exception=e
)Types
from typing import Optional, Dict, Any, Callable
from requests import Response
import gevent
# HTTP Session types
HttpMethod = str # "GET", "POST", "PUT", etc.
RequestData = Optional[Dict[str, Any]]
RequestHeaders = Optional[Dict[str, str]]
# Response types
class LocustResponse(Response):
"""Extended requests.Response with Locust features."""
class FastResponse:
"""Fast HTTP response from geventhttpclient."""
status_code: int
headers: Dict[str, str]
content: bytes
text: str
def json(self) -> Any:
"""Parse response as JSON."""
# Context managers
class ResponseContextManager:
"""Context manager for response validation."""
def __enter__(self) -> LocustResponse: ...
def __exit__(self, exc_type, exc_val, exc_tb): ...
class RestResponseContextManager:
"""REST-specific response context manager."""
def __enter__(self) -> FastResponse: ...
def __exit__(self, exc_type, exc_val, exc_tb): ...Install with Tessl CLI
npx tessl i tessl/pypi-locust