← Back to DevBytes
Testing July 14, 2026 15 min read

Locust: Complete Testing Guide for Developers

Introduction to Locust

Modern web applications must withstand real-world traffic spikes, unpredictable user behavior, and sustained high concurrency. Traditional load testing tools often require complex XML configurations or proprietary scripting languages. Locust flips this model entirely: you write your test scenarios in plain Python, giving you unlimited flexibility and a remarkably gentle learning curve.

What Is Locust?

🚀 Deploy your AI agent in 10 minutes

Managed Hermes hosting. Zero DevOps. 100M tokens/mo included.

Try it free →

Locust is an open-source, Python-based load testing framework that allows you to define user behavior with standard Python code, then swarm your application with thousands of concurrent simulated users. Unlike tools such as JMeter, which rely on GUI-driven test plans and XML, Locust gives you a programmatic, code-first approach. Each simulated user is a lightweight greenlet (a micro-thread), meaning a single machine can easily handle thousands of concurrent connections without heavy resource consumption.

Core components of Locust:

Why Locust Matters

Load testing is not just about verifying your application can handle 10,000 users. It's about understanding how it degrades under pressure. Locust matters because:

Installing Locust

Locust requires Python 3.7 or later. Install it via pip:

pip install locust

For the latest development version or additional dependencies (like WebSocket testing):

pip install locustio  # older package name (legacy)
pip install locust[websocket]  # with WebSocket support

Verify the installation:

locust --version

Writing Your First Locust Test

Create a file named locustfile.py. This is the default filename Locust looks for. Below is a complete example that simulates a user browsing a website:

from locust import HttpUser, task, between
import random

class WebsiteUser(HttpUser):
    """
    Simulates a typical website visitor:
    - Lands on the homepage
    - Browses a product listing
    - Views a random product detail page
    - Waits 2-5 seconds between actions (think time)
    """
    
    # Wait between 2 and 5 seconds after each task
    wait_time = between(2, 5)
    
    def on_start(self):
        """Called when a simulated user starts. Good for login/setup."""
        self.client.get("/", name="Homepage - initial load")
    
    @task(3)  # Higher weight = executed more frequently
    def view_products_list(self):
        # Simulate browsing the product catalog
        page = random.choice(["electronics", "books", "clothing", "sports"])
        response = self.client.get(f"/products/{page}", name="Product listing")
        if response.status_code == 200:
            # Optionally parse the response to extract links
            pass
    
    @task(1)
    def view_product_detail(self):
        product_id = random.randint(1, 1000)
        with self.client.get(f"/product/{product_id}", 
                             name="Product detail",
                             catch_response=True) as response:
            if response.status_code == 404:
                response.failure("Product not found")
            elif response.elapsed.total_seconds() > 2.0:
                response.failure("Response took too long")
    
    @task(2)
    def search_products(self):
        keywords = ["laptop", "phone", "shoes", "camera"]
        query = random.choice(keywords)
        self.client.get("/search", params={"q": query}, name="Search")
    
    @task(1)
    def view_about_page(self):
        self.client.get("/about", name="About page")

Let's break down what's happening here:

Running Locust

The Web UI Mode

The most common way to run Locust is with its built-in web interface. Navigate to your locustfile.py directory and run:

locust -f locustfile.py --host https://your-app.com

Then open http://localhost:8089 in your browser. You'll see:

Headless / CLI Mode (for CI/CD)

For automated pipelines, run Locust without the web UI:

locust -f locustfile.py --host https://staging.example.com \
    --headless \
    --users 500 \
    --spawn-rate 50 \
    --run-time 10m \
    --csv results \
    --html report.html

Key CLI flags:

Custom Load Shapes

Sometimes a linear ramp isn't realistic. You can define a custom load shape using LoadShape:

from locust import LoadShape, HttpUser, task, between
import math

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)
    
    @task
    def my_task(self):
        self.client.get("/api/endpoint")

class StepLoadShape(LoadShape):
    """
    A stepped load pattern:
    - Minutes 0-5:  100 users
    - Minutes 5-10: 300 users  
    - Minutes 10-15: 600 users
    - Minutes 15-20: 1000 users
    - Then ramp down to 0 over 2 minutes
    """
    
    stages = [
        {"duration": 300, "users": 100, "spawn_rate": 20},
        {"duration": 300, "users": 300, "spawn_rate": 40},
        {"duration": 300, "users": 600, "spawn_rate": 60},
        {"duration": 300, "users": 1000, "spawn_rate": 80},
        {"duration": 120, "users": 0, "spawn_rate": 100},
    ]
    
    def tick(self):
        """Called every second. Must return (user_count, spawn_rate) or None."""
        run_time = self.get_run_time()
        total_duration = 0
        
        for stage in self.stages:
            total_duration += stage["duration"]
            if run_time < total_duration:
                # We're in this stage
                return (stage["users"], stage["spawn_rate"])
        
        # Test complete
        return None

Run with the custom shape:

locust -f locustfile.py --host https://api.example.com --shape StepLoadShape

You can also use built-in shapes like DoubleWave or TimeRamp from locust.shape.

Distributed Load Generation

When a single machine cannot generate enough traffic, run Locust in distributed mode with one master and multiple workers:

Master node (runs the web UI and coordinates workers):

locust -f locustfile.py --host https://target-system.com --master --expect-workers 4

Worker nodes (on separate machines, pointing to the master's IP):

locust -f locustfile.py --worker --master-host 192.168.1.100 --master-port 5557

Important notes for distributed runs:

Advanced Task Patterns

Sequential Tasks

By default, tasks are chosen randomly (weighted). If you need a strict order (like a checkout funnel), use SequentialTaskSet:

from locust import HttpUser, task, between, SequentialTaskSet

class CheckoutFlow(SequentialTaskSet):
    """Executes tasks in the order they are defined, top to bottom."""
    
    @task
    def add_to_cart(self):
        self.client.post("/cart", json={"product_id": 42, "quantity": 1})
    
    @task
    def view_cart(self):
        self.client.get("/cart")
    
    @task
    def initiate_checkout(self):
        self.client.post("/checkout/start")
    
    @task
    def submit_payment(self):
        self.client.post("/checkout/payment", json={
            "card_token": "tok_visa",
            "amount": 2999
        })
    
    @task
    def verify_order(self):
        with self.client.get("/orders/latest", catch_response=True) as resp:
            if "confirmed" not in resp.text:
                resp.failure("Order not confirmed")

class ShoppingUser(HttpUser):
    wait_time = between(3, 8)
    # Attach the sequential task set
    tasks = [CheckoutFlow]

Nesting TaskSets for Complex Workflows

You can nest TaskSets and switch between them at runtime:

from locust import HttpUser, task, between, TaskSet

class BrowseTaskSet(TaskSet):
    @task(3)
    def browse_products(self):
        self.client.get("/products")
    
    @task(1)
    def exit_to_checkout(self):
        """Switch to the checkout TaskSet."""
        self.interrupt()  # Return control to parent, which may switch

class PurchaseTaskSet(TaskSet):
    @task
    def add_to_cart(self):
        self.client.post("/cart/add", json={"item": "widget"})
    
    @task
    def checkout(self):
        self.client.post("/checkout/complete")
        self.interrupt()  # Done purchasing, go back to browsing

class ComplexUser(HttpUser):
    wait_time = between(1, 5)
    
    # Start with browsing
    tasks = [BrowseTaskSet]
    
    @task(1)
    def switch_to_purchase(self):
        """Occasionally switch the entire user to purchase mode."""
        self.client.get("/")  # some trigger
        # Replace current TaskSet with PurchaseTaskSet
        self.tasks = [PurchaseTaskSet]

Custom Wait Time Functions

You can use any callable for wait_time, enabling dynamic think times:

from locust import HttpUser, task
import random

def adaptive_wait(self):
    """
    Longer wait if the last request was slow, 
    shorter if it was fast — simulating user frustration/flow.
    """
    if hasattr(self, 'last_response_time'):
        if self.last_response_time > 1.0:
            return random.uniform(5, 10)  # Frustrated, slow down
        else:
            return random.uniform(1, 3)   # Fast site, keep browsing
    return random.uniform(2, 5)

class AdaptiveUser(HttpUser):
    wait_time = adaptive_wait
    
    @task
    def load_page(self):
        response = self.client.get("/dashboard")
        # Store response time for the wait function to use
        self.last_response_time = response.elapsed.total_seconds()

Testing REST APIs

Here's a comprehensive example for testing a REST API with authentication, CRUD operations, and realistic data:

from locust import HttpUser, task, between, events
import json
import random
import string

def generate_random_payload():
    return {
        "name": ''.join(random.choices(string.ascii_letters, k=10)),
        "email": f"user_{random.randint(1, 100000)}@test.com",
        "status": random.choice(["active", "inactive", "pending"]),
        "metadata": {
            "source": "load-test",
            "timestamp": "2025-01-01T00:00:00Z"
        }
    }

class APITestUser(HttpUser):
    wait_time = between(0.5, 2.0)
    
    # Store created resource IDs for cleanup and read operations
    created_ids = []
    
    def on_start(self):
        """Authenticate and obtain a token."""
        auth_response = self.client.post("/auth/login", json={
            "username": "testuser",
            "password": "testpass123"
        })
        if auth_response.status_code == 200:
            self.token = auth_response.json()["access_token"]
            self.client.headers.update({
                "Authorization": f"Bearer {self.token}",
                "Content-Type": "application/json"
            })
        else:
            self.token = None
    
    @task(2)
    def get_resources(self):
        """List resources with pagination."""
        page = random.randint(1, 10)
        response = self.client.get(f"/api/resources?page={page}&limit=20",
                                    name="GET /api/resources")
        if response.status_code == 200:
            data = response.json()
            # Optionally extract IDs from the response
            for item in data.get("items", [])[:5]:
                self.created_ids.append(item["id"])
    
    @task(3)
    def create_resource(self):
        """Create a new resource."""
        payload = generate_random_payload()
        with self.client.post("/api/resources", 
                              json=payload,
                              name="POST /api/resources",
                              catch_response=True) as resp:
            if resp.status_code == 201:
                resource_id = resp.json().get("id")
                if resource_id:
                    self.created_ids.append(resource_id)
            elif resp.status_code == 409:
                # Conflict is acceptable in some scenarios
                pass
            elif resp.status_code >= 500:
                resp.failure(f"Server error on create: {resp.status_code}")
    
    @task(2)
    def read_resource(self):
        """Read a specific resource by ID."""
        if not self.created_ids:
            return  # Skip if we have no IDs yet
        resource_id = random.choice(self.created_ids)
        with self.client.get(f"/api/resources/{resource_id}",
                             name="GET /api/resources/{id}",
                             catch_response=True) as resp:
            if resp.status_code == 404:
                # Resource may have been deleted by another user
                self.created_ids.remove(resource_id)
            elif resp.status_code == 200:
                pass  # Success
    
    @task(1)
    def update_resource(self):
        """Update an existing resource."""
        if not self.created_ids:
            return
        resource_id = random.choice(self.created_ids)
        update_payload = {"status": "updated", "name": "modified-resource"}
        self.client.put(f"/api/resources/{resource_id}",
                        json=update_payload,
                        name="PUT /api/resources/{id}")
    
    @task(1)
    def delete_resource(self):
        """Delete a resource."""
        if not self.created_ids:
            return
        resource_id = self.created_ids.pop(random.randint(0, len(self.created_ids) - 1))
        with self.client.delete(f"/api/resources/{resource_id}",
                                name="DELETE /api/resources/{id}",
                                catch_response=True) as resp:
            if resp.status_code not in [200, 204]:
                resp.failure(f"Unexpected status on delete: {resp.status_code}")

# Event hook: log failures for later analysis
@events.request.add_listener
def log_failure(request_type, name, response_time, exception, **kwargs):
    if exception:
        print(f"[FAILURE] {request_type} {name}: {exception}")

Event Hooks and Extensibility

Locust provides a rich event system. You can attach listeners to hook into the test lifecycle:

from locust import events
import time
import csv
import os

# Track custom metrics
response_times_by_endpoint = {}

@events.request.add_listener
def collect_response_times(request_type, name, response_time, exception, **kwargs):
    """Collect response times for custom aggregation."""
    if exception:
        return
    if name not in response_times_by_endpoint:
        response_times_by_endpoint[name] = []
    response_times_by_endpoint[name].append(response_time)

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    print(f"Test starting at {time.strftime('%Y-%m-%d %H:%M:%S')}")
    # Initialize external monitoring, clear caches, etc.

@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    """Write custom report when test finishes."""
    print(f"Test completed. Writing custom report...")
    with open("custom_report.csv", "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["Endpoint", "Avg Response Time", "P95", "Count"])
        for endpoint, times in response_times_by_endpoint.items():
            times_sorted = sorted(times)
            avg = sum(times) / len(times)
            p95_index = int(len(times_sorted) * 0.95)
            p95 = times_sorted[p95_index] if times_sorted else 0
            writer.writerow([endpoint, f"{avg:.3f}", f"{p95:.3f}", len(times)])

@events.spawning_complete.add_listener
def on_spawning_complete(user_count, **kwargs):
    print(f"All {user_count} users have been spawned!")

@events.user_error.add_listener
def on_user_error(user_instance, exception, **kwargs):
    print(f"Error in user {user_instance}: {exception}")

Available event hooks:

Best Practices for Effective Load Testing

1. Design Realistic User Behavior

The quality of your test depends entirely on how well it mimics real users. Study your analytics: what pages do users visit, in what order, with what think times? Use between() for variable wait times rather than fixed pauses. Mix read and write operations proportionally to production traffic patterns.

2. Start Small and Ramp Gradually

Never jump directly to 10,000 concurrent users. Start with 10 users to validate your test scripts work correctly. Then run incremental tests: 100, 500, 1000, 5000 users. Use a gradual spawn rate (e.g., 50 users/second) to observe how your system degrades, rather than hammering it instantly. This reveals the "knee" where performance breaks down.

3. Use Custom Names for Dynamic URLs

If your test hits URLs with variable IDs (like /users/12345), use the name parameter to group them in statistics:

self.client.get(f"/users/{user_id}", name="/users/{id}")

Without this, you'll get thousands of separate entries in the statistics table, making analysis impossible.

4. Validate Responses, Not Just Status Codes

A 200 OK doesn't mean the response is correct. Use catch_response=True to validate response bodies, check for expected JSON structure, or ensure response times meet your SLA:

with self.client.get("/api/data", catch_response=True) as resp:
    if resp.status_code == 200:
        data = resp.json()
        if "results" not in data:
            resp.failure("Missing 'results' key in response")
        if len(data["results"]) == 0:
            resp.failure("Empty results returned")
    if resp.elapsed.total_seconds() > 1.5:
        resp.failure("SLA violation: response > 1.5s")

5. Manage Test Data Carefully

Avoid hardcoded IDs that might not exist in the target environment. Generate dynamic data, use factories, or seed a test database before the run. Clean up created resources after the test to avoid polluting your environment.

6. Run in a Production-Like Environment

Test an environment that mirrors production in infrastructure, data volume, and configuration. Load testing a tiny staging database with 100 rows tells you nothing about production with 10 million rows. Consider running tests against a production replica or a scaled-down clone with proportional data.

7. Monitor the Target System, Not Just Locust

Locust shows you client-side metrics. Correlate these with server-side metrics: CPU, memory, database connections, queue lengths, garbage collection pauses. Set up dashboards (Grafana, Datadog) to watch both simultaneously during a test.

8. Use Distributed Mode for High Concurrency

A single Locust node typically handles 500-2000 concurrent users (depending on task complexity). For larger tests, distribute across multiple worker machines. Ensure network egress from workers doesn't become the bottleneck.

9. Incorporate Load Testing into CI/CD

Run a short smoke test (e.g., 50 users for 2 minutes) in your CI pipeline on every deployment to staging. Use the --headless flag and fail the build if error rate exceeds a threshold. For example:

locust -f locustfile.py --host $STAGING_URL --headless \
    --users 50 --spawn-rate 10 --run-time 2m \
    --csv ci_results \
    && python check_results.py ci_results_stats.csv

Where check_results.py parses the CSV and exits non-zero if failures exceed 1%.

10. Avoid Testing from Behind a Single NAT or VPN

If all your load generator workers share the same outbound IP, you may hit rate limits, load balancer affinity, or firewall rules that don't reflect real-world distributed traffic. Use workers on different networks or cloud regions when possible.

Common Patterns and Troubleshooting

Handling Authentication Across Users

For token-based auth, generate a fresh token per user in on_start rather than sharing one globally. This avoids unrealistic contention on a single session:

class AuthenticatedUser(HttpUser):
    wait_time = between(1, 3)
    
    def on_start(self):
        # Each simulated user logs in independently
        creds = self.get_credentials()  # fetch from a pool or generate
        resp = self.client.post("/auth/login", json=creds)
        if resp.status_code == 200:
            self.token = resp.json()["token"]
            self.client.headers["Authorization"] = f"Bearer {self.token}"
        else:
            print(f"Login failed for user: {creds}")
            self.token = None
    
    def get_credentials(self):
        # Rotate through a pool of test accounts
        # or generate unique credentials per user
        user_num = self.environment.runner.user_count if self.environment.runner else 1
        return {"username": f"loadtest_{user_num}", "password": "testpass"}

Using Test Data from CSV Files

For data-driven tests, load a CSV once and distribute rows across users:

import csv
from locust import HttpUser, task, between

# Load data once at module level (shared across all users)
def load_test_data(filepath):
    with open(filepath, "r") as f:
        reader = csv.DictReader(f)
        return list(reader)

TEST_USERS = load_test_data("test_users.csv")  # columns: username, password, role

class DataDrivenUser(HttpUser):
    wait_time = between(2, 5)
    
    def on_start(self):
        # Pick a unique row based on user index
        index = self.environment.runner.user_count % len(TEST_USERS)
        user_data = TEST_USERS[index]
        self.username = user_data["username"]
        self.role = user_data["role"]
        # Login with these credentials
        self.client.post("/login", json={
            "username": user_data["username"],
            "password": user_data["password"]
        })
    
    @task
    def access_dashboard(self):
        self.client.get(f"/dashboard?role={self.role}")

Testing WebSocket Endpoints

Install locust[websocket] and use the WebSocket user:

from locust import User, task, between
from locust.contrib.websocket import WebSocketClient

class WebSocketUser(User):
    wait_time = between(1, 5)
    
    def __init__(self, environment):
        super().__init__(environment)
        self.client = WebSocketClient("wss://echo.example.com/ws", environment)
    
    def on_start(self):
        self.client.connect()
    
    @task
    def send_message(self):
        self.client.send("Hello, server!")
    
    @task
    def receive_message(self):
        # WebSocketClient handles receiving in a non-blocking way
        self.client.recv()

Conclusion

Locust transforms load testing from a cumbersome, GUI-driven chore into a natural extension of your development workflow. By expressing user behavior as Python code, you gain unlimited flexibility to model real-world traffic, validate responses programmatically, and integrate load testing into CI/CD pipelines. Its lightweight greenlet-based architecture lets you simulate massive concurrency from modest hardware, while the master/worker design scales horizontally when needed. The real-time web dashboard gives immediate visibility into how your application performs under stress, and the rich event system allows custom reporting, alerting, and integration with your observability stack. Whether you're testing a simple REST API, a complex multi-step checkout flow, or a WebSocket-based real-time service, Locust provides the tools to answer the critical question: will my application hold up when real users arrive?

🚀 Need a reliable AI agent for your project?

Deploy Hermes Agent in 10 minutes. Managed hosting, zero DevOps.

Get Started — $23.99/mo
← Back to all articles