Locust

A developer-friendly load testing framework that enables you to write test scenarios in plain Python code. Locust supports distributed testing, provides a web-based UI for monitoring, and can test HTTP services and other protocols including databases and WebSockets.

Package Information

• Package Name: locust
• Language: Python
• Installation: pip install locust
• Python Requirement: >=3.10

Core Imports

import locust

Common imports for load testing:

from locust import HttpUser, task, between
from locust import User, TaskSet, events

Basic Usage

from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
    wait_time = between(1, 5)  # Wait 1-5 seconds between tasks
    
    @task
    def index_page(self):
        self.client.get("/")
    
    @task(3)  # This task is 3x more likely to run
    def view_item(self):
        item_id = random.randint(1, 10000)
        self.client.get(f"/item?id={item_id}", name="/item")
    
    def on_start(self):
        """Called when user starts"""
        self.client.post("/login", json={
            "username": "testuser",
            "password": "secret"
        })

Run the test:

locust -f locustfile.py --host=http://mywebsite.com

Architecture

Locust uses an event-driven architecture built on gevent for handling thousands of concurrent users efficiently:

• Users: Define behavior patterns and represent individual virtual users
• TaskSets: Organize related tasks and enable complex user flow modeling
• Tasks: Individual actions that users perform, decorated with @task
• Events System: Extensible hooks for custom metrics, logging, and integrations
• Runners: Orchestrate test execution (local, master/worker distributed setups)
• Web UI: Real-time monitoring dashboard for test progress and results

This design enables flexible test scenarios from simple HTTP load tests to complex multi-protocol distributed testing with custom metrics and reporting.

Capabilities

User Classes

Foundation classes for defining virtual user behavior including HTTP users, generic users, and high-performance FastHTTP users. Users define the core behavior patterns that drive load testing scenarios.

class User:
    def __init__(self, environment): ...

class HttpUser(User):
    client: HttpSession

class FastHttpUser(User):
    client: FastHttpSession

User Classes

TaskSets and Task Management

System for organizing and controlling task execution including sequential, random, and Markov chain-based task selection. TaskSets enable complex user behavior modeling and task flow control.

class TaskSet:
    def __init__(self, parent): ...

class SequentialTaskSet(TaskSet): ...
class MarkovTaskSet(TaskSet): ...

def task(weight=1): ...
def tag(*tags): ...

TaskSets and Tasks

Wait Time Functions

Functions for controlling timing between task executions including random intervals, constant delays, pacing, and throughput control. These functions enable realistic user behavior simulation.

def between(min_wait, max_wait): ...
def constant(wait_time): ...
def constant_pacing(wait_time): ...
def constant_throughput(task_runs_per_second): ...

Wait Time Functions

Events System

Comprehensive event system for extending Locust with custom functionality including request hooks, lifecycle events, and distributed testing coordination.

events: Events  # Global events instance

class Events:
    request: EventHook
    user_error: EventHook
    test_start: EventHook
    test_stop: EventHook
    # ... 15+ additional events

Events System

Protocol Extensions

Built-in support for testing non-HTTP protocols including MongoDB, PostgreSQL, WebSocket/SocketIO, and other systems through the contrib module ecosystem.

class MongoDBUser(User): ...
class PostgresUser(User): ...
class SocketIOUser(User): ...

Protocol Extensions

Load Test Shaping

Custom load test patterns and shapes for advanced testing scenarios including ramp-up patterns, step functions, and complex load profiles over time.

class LoadTestShape:
    def tick(self): ...  # Returns (user_count, spawn_rate) or None

Load Test Shaping

Exception Handling

Comprehensive exception system for controlling test flow including task interruption, user stopping, response validation, and RPC communication errors.

class LocustError(Exception): ...
class InterruptTaskSet(Exception): ...
class StopUser(Exception): ...
class ResponseError(Exception): ...

Exception Handling

Debugging and Utilities

Tools for debugging test scenarios including single-user execution, task ratio inspection, and development-time testing utilities.

def run_single_user(user_class, **options): ...
def print_task_ratio(user_classes): ...

Debugging and Utilities

Types

Core Types

from typing import Callable, Dict, List, Optional, Union, Any
from gevent.greenlet import Greenlet

# User and TaskSet configuration
TaskFunction = Callable[[], None]
TaskDict = Dict[TaskFunction, int]
TaskList = List[Union[TaskFunction, tuple]]
WaitTimeFunction = Callable[[], float]

# Event system types  
EventHandler = Callable[..., None]
ResponseContextManager = Any  # Context manager for response validation