Locust for Load Testing: Python-Based Performance Testing

Locust is a Python-based load testing framework that uses plain Python code to define user behaviour. For teams already working in Python, it’s a natural fit that eliminates the need to learn a new domain-specific language or GUI tool.

Why Locust?

Locust differentiates itself in several ways:

Feature	Locust	JMeter	k6
Language	Python	XML/GUI	JavaScript
Web UI	Built-in	Separate	Cloud/Grafana
Distributed testing	Native	Master/slave	Cloud/Operator
Resource usage	Moderate	High	Low
Protocol support	HTTP-focused	Extensive	HTTP-focused

The main advantage is Python itself. If you can write Python, you can write complex test scenarios without learning new tools.

Installation

Install Locust using pip:

pip install locust

Verify the installation:

locust --version

Your First Locustfile

Create a file called locustfile.py:

from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)  # Random wait between requests

    @task
    def view_homepage(self):
        self.client.get("/")

    @task(3)  # 3x more likely than other tasks
    def view_products(self):
        self.client.get("/products")

    @task
    def view_about(self):
        self.client.get("/about")

Run the test:

locust -f locustfile.py --host=https://example.com

Open http://localhost:8089 to access the web UI, set your user count and spawn rate, and start the test.

Command-Line Mode

For CI/CD integration, run without the web UI:

locust -f locustfile.py \
  --host=https://example.com \
  --users 100 \
  --spawn-rate 10 \
  --run-time 5m \
  --headless \
  --csv=results

This generates CSV files with request statistics.

Handling Authentication

Locust’s HttpUser maintains session cookies automatically. For login flows:

from locust import HttpUser, task, between

class AuthenticatedUser(HttpUser):
    wait_time = between(1, 2)

    def on_start(self):
        """Called when a user starts - login here"""
        response = self.client.post("/login", json={
            "username": "testuser",
            "password": "testpass"
        })
        self.token = response.json().get("access_token")

    @task
    def access_protected_resource(self):
        headers = {"Authorization": f"Bearer {self.token}"}
        self.client.get("/api/protected", headers=headers)

Sequential Task Flows

For ordered user journeys, use SequentialTaskSet:

from locust import HttpUser, SequentialTaskSet, task, between

class CheckoutFlow(SequentialTaskSet):
    @task
    def browse_products(self):
        self.client.get("/products")

    @task
    def add_to_cart(self):
        self.client.post("/cart", json={"product_id": 123, "quantity": 1})

    @task
    def view_cart(self):
        self.client.get("/cart")

    @task
    def checkout(self):
        self.client.post("/checkout", json={"payment_method": "card"})

    @task
    def stop(self):
        self.interrupt()  # Exit the flow

class ShoppingUser(HttpUser):
    wait_time = between(2, 5)
    tasks = [CheckoutFlow]

Parameterised Data

Load test data from files or generate it dynamically:

import csv
from locust import HttpUser, task, between

class DataDrivenUser(HttpUser):
    wait_time = between(1, 2)

    def on_start(self):
        # Load test data
        with open('users.csv', 'r') as f:
            reader = csv.DictReader(f)
            self.users = list(reader)
        self.current_user = self.users.pop()

    @task
    def login_user(self):
        self.client.post("/login", json={
            "username": self.current_user["username"],
            "password": self.current_user["password"]
        })

Custom Metrics and Validation

Add response validation and custom metrics:

from locust import HttpUser, task, between, events
import time

class ValidatingUser(HttpUser):
    wait_time = between(1, 2)

    @task
    def search_products(self):
        start_time = time.time()

        with self.client.get("/api/search?q=laptop", catch_response=True) as response:
            if response.status_code == 200:
                data = response.json()
                if len(data.get("results", [])) > 0:
                    response.success()
                else:
                    response.failure("No search results returned")
            else:
                response.failure(f"Got status code {response.status_code}")

        # Custom timing for specific operations
        duration = time.time() - start_time
        events.request.fire(
            request_type="SEARCH",
            name="Product Search",
            response_time=duration * 1000,
            response_length=len(response.content),
            exception=None,
        )

Distributed Testing

Locust supports distributed testing with master and worker nodes.

Master node:

locust -f locustfile.py --master --host=https://example.com

Worker nodes (run on multiple machines):

locust -f locustfile.py --worker --master-host=192.168.1.100

For Docker-based distributed testing:

# docker-compose.yml
version: '3'
services:
  master:
    image: locustio/locust
    ports:
      - "8089:8089"
    volumes:
      - ./:/mnt/locust
    command: -f /mnt/locust/locustfile.py --master -H https://example.com

  worker:
    image: locustio/locust
    volumes:
      - ./:/mnt/locust
    command: -f /mnt/locust/locustfile.py --worker --master-host master
    deploy:
      replicas: 4

Run with:

docker-compose up --scale worker=4

Custom Load Shapes

Define complex load patterns with LoadTestShape:

from locust import HttpUser, task, between, LoadTestShape

class StagesShape(LoadTestShape):
    stages = [
        {"duration": 60, "users": 10, "spawn_rate": 1},
        {"duration": 120, "users": 50, "spawn_rate": 5},
        {"duration": 180, "users": 100, "spawn_rate": 10},
        {"duration": 240, "users": 50, "spawn_rate": 5},
        {"duration": 300, "users": 0, "spawn_rate": 5},
    ]

    def tick(self):
        run_time = self.get_run_time()

        for stage in self.stages:
            if run_time < stage["duration"]:
                return (stage["users"], stage["spawn_rate"])

        return None  # Stop the test

class WebUser(HttpUser):
    wait_time = between(1, 3)

    @task
    def index(self):
        self.client.get("/")

Exporting Results

Export results for analysis:

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

This generates:

• results_stats.csv - Request statistics
• results_failures.csv - Failed requests
• results_stats_history.csv - Time-series data
• report.html - Visual report

Integration with pytest

Run Locust tests as part of your test suite:

# test_performance.py
import subprocess

def test_api_performance():
    result = subprocess.run([
        "locust",
        "-f", "locustfile.py",
        "--host", "https://staging.example.com",
        "--users", "10",
        "--spawn-rate", "2",
        "--run-time", "1m",
        "--headless",
        "--only-summary"
    ], capture_output=True, text=True)

    assert result.returncode == 0
    assert "0 failures" in result.stdout or "0(0.00%)" in result.stdout

Locust’s Python foundation makes it accessible to development teams and integrates naturally with existing Python tooling. For HTTP-based load testing with complex business logic, it’s an excellent choice that balances power with simplicity.