Locust Python Load Testing: Complete Performance Testing Guide

Introduction to Locust

Locust is a Python-based, open-source load testing tool that enables developers to write test scenarios in pure Python code. Unlike GUI-heavy tools like JMeter, Locust (as discussed in Gatling: High-Performance Load Testing with Scala DSL) provides a code-first approach with powerful distributed testing capabilities and real-time web-based monitoring.

Basic Load Test

# locustfile.py
from locust (as discussed in [K6: Modern Load Testing with JavaScript for DevOps Teams](/blog/k6-modern-load-testing)) import HttpUser, task, between

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)  # Wait 1-3 seconds between tasks
    host = "https://api.example.com"

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

    @task(1)
    def view_product_detail(self):
        product_id = 123
        self.client.get(f"/api/products/{product_id}")

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

    def on_start(self):
        # Login before starting tasks
        response = self.client.post("/api/auth/login", json={
            "username": "test@example.com",
            "password": "password123"
        })
        self.token = response.json()["token"]
        self.client.headers.update({"Authorization": f"Bearer {self.token}"})

Advanced Scenarios

Sequential User Flow

from locust import HttpUser, task, SequentialTaskSet, between

class UserBehavior(SequentialTaskSet):
    def on_start(self):
        self.product_id = None
        self.cart_id = None

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

    @task
    def search_products(self):
        response = self.client.get("/api/products?category=electronics")
        products = response.json()
        if products:
            self.product_id = products[0]["id"]

    @task
    def view_product(self):
        if self.product_id:
            self.client.get(f"/api/products/{self.product_id}")

    @task
    def add_to_cart(self):
        if self.product_id:
            response = self.client.post("/api/cart", json={
                "product_id": self.product_id,
                "quantity": 1
            })
            self.cart_id = response.json()["cart_id"]

    @task
    def checkout(self):
        if self.cart_id:
            self.client.post(f"/api/checkout/{self.cart_id}", json={
                "payment_method": "credit_card",
                "shipping_address": "123 Main St"
            })

    @task
    def stop(self):
        self.interrupt()  # Stop task sequence

class EcommerceUser(HttpUser):
    wait_time = between(2, 5)
    tasks = [UserBehavior]

Data-Driven Testing

import csv
from locust import HttpUser, task, between
from itertools import cycle

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

    def on_start(self):
        # Load test data
        with open('test_users.csv', 'r') as f:
            reader = csv.DictReader(f)
            self.user_data = cycle(list(reader))

    @task
    def login_with_test_data(self):
        user = next(self.user_data)
        response = self.client.post("/api/login", json={
            "username": user['username'],
            "password": user['password']
        })

        if response.status_code == 200:
            self.token = response.json()["token"]

Distributed Load Testing

Master Configuration

# Run master node
# locust -f locustfile.py --master --expect-workers=4

# Run worker nodes
# locust -f locustfile.py --worker --master-host=localhost

Docker Compose Setup

# docker-compose.yml
version: '3'

services:
  master:
    image: locustio/locust
    ports:
      - "8089:8089"
    volumes:
      - ./:/mnt/locust
    command: -f /mnt/locust/locustfile.py --master --expect-workers=4

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

Kubernetes Deployment

# locust-master.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: locust-master
spec:
  replicas: 1
  template:
    spec:
      containers:
      - name: locust
        image: locustio/locust
        args: ["-f", "/locust/locustfile.py", "--master"]
        ports:
        - containerPort: 8089
        - containerPort: 5557
        volumeMounts:
        - name: locust-scripts
          mountPath: /locust

---
# locust-worker.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: locust-worker
spec:
  replicas: 10
  template:
    spec:
      containers:
      - name: locust
        image: locustio/locust
        args: ["-f", "/locust/locustfile.py", "--worker", "--master-host=locust-master"]
        volumeMounts:
        - name: locust-scripts
          mountPath: /locust

Custom Metrics and Reporting

from locust import HttpUser, task, events
import time

# Custom metric tracking
request_times = []

@events.request.add_listener
def on_request(request_type, name, response_time, response_length, exception, **kwargs):
    request_times.append(response_time)

    if response_time > 1000:  # Alert on slow requests
        print(f"Slow request detected: {name} took {response_time}ms")

@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    # Calculate percentiles
    if request_times:
        sorted_times = sorted(request_times)
        p50 = sorted_times[len(sorted_times) // 2]
        p95 = sorted_times[int(len(sorted_times) * 0.95)]
        p99 = sorted_times[int(len(sorted_times) * 0.99)]

        print(f"\nPerformance Summary:")
        print(f"P50: {p50}ms")
        print(f"P95: {p95}ms")
        print(f"P99: {p99}ms")

class MonitoredUser(HttpUser):
    @task
    def monitored_request(self):
        start_time = time.time()
        response = self.client.get("/api/data")
        elapsed = (time.time() - start_time) * 1000

        # Custom validation
        if response.status_code == 200:
            data = response.json()
            if len(data) < 10:
                events.request.fire(
                    request_type="GET",
                    name="/api/data",
                    response_time=elapsed,
                    response_length=len(response.content),
                    exception=Exception("Insufficient data returned")
                )

Locust vs JMeter Comparison

Feature	Locust	JMeter
Language	Python	GUI/XML
Learning Curve	Low (if Python known)	Medium-High
Test as Code	Yes (native)	Limited (requires plugins)
Distributed Testing	Built-in	Built-in
Real-time UI	Web-based, modern	Java Swing, dated
Version Control	Excellent (Python files)	Poor (XML files)
Resource Usage	Lower	Higher (Java)
Protocol Support	HTTP/WebSocket (extensible)	Extensive built-in
CI/CD Integration	Excellent	Good
Scripting Flexibility	Excellent (Python)	Limited (Groovy/BeanShell)
Community	Growing	Large, established
License	MIT	Apache 2.0

CI/CD Integration

GitHub Actions

# .github/workflows/load-test.yml
name: Load Test

on:
  schedule:
    - cron: '0 2 * * *'  # Daily at 2 AM
  workflow_dispatch:

jobs:
  load-test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install locust

      - name: Run Load Test
        run: |
          locust -f locustfile.py \
            --headless \
            --users 100 \
            --spawn-rate 10 \
            --run-time 5m \
            --host https://api.example.com \
            --html report.html \
            --csv results

      - name: Upload Results
        uses: actions/upload-artifact@v3
        with:
          name: load-test-results
          path: |
            report.html
            results_*.csv

      - name: Check Performance Thresholds
        run: |
          python check_thresholds.py results_stats.csv

Threshold Validation Script

# check_thresholds.py
import csv
import sys

def check_thresholds(stats_file):
    thresholds = {
        'avg_response_time': 200,  # ms
        'max_response_time': 2000,  # ms
        'failure_rate': 0.01  # 1%
    }

    with open(stats_file, 'r') as f:
        reader = csv.DictReader(f)
        for row in reader:
            if row['Name'] == 'Aggregated':
                avg_response = float(row['Average Response Time'])
                max_response = float(row['Max Response Time'])
                failure_rate = float(row['Failure Count']) / float(row['Request Count'])

                if avg_response > thresholds['avg_response_time']:
                    print(f"FAIL: Average response time {avg_response}ms exceeds threshold")
                    sys.exit(1)

                if max_response > thresholds['max_response_time']:
                    print(f"FAIL: Max response time {max_response}ms exceeds threshold")
                    sys.exit(1)

                if failure_rate > thresholds['failure_rate']:
                    print(f"FAIL: Failure rate {failure_rate:.2%} exceeds threshold")
                    sys.exit(1)

    print("PASS: All thresholds met")
    sys.exit(0)

if __name__ == '__main__':
    check_thresholds(sys.argv[1])

Best Practices

1. Realistic User Simulation

from locust import HttpUser, task, between
import random

class RealisticUser(HttpUser):
    wait_time = between(1, 5)

    @task(10)
    def browse(self):
        # Simulate browsing behavior
        pages = ['/products', '/about', '/contact']
        self.client.get(random.choice(pages))

    @task(3)
    def search(self):
        keywords = ['laptop', 'phone', 'tablet', 'watch']
        self.client.get(f"/search?q={random.choice(keywords)}")

    @task(1)
    def purchase(self):
        # Only 10% of users purchase
        if random.random() < 0.1:
            self.client.post("/api/orders", json={
                "items": [{"id": random.randint(1, 100), "qty": 1}]
            })

2. Proper Error Handling

from locust import HttpUser, task
from locust.exception import RescheduleTask

class ResilientUser(HttpUser):
    @task
    def api_call(self):
        with self.client.get("/api/data", catch_response=True) as response:
            if response.status_code == 429:  # Rate limited
                response.failure("Rate limited")
                raise RescheduleTask()  # Retry later

            elif response.status_code == 500:
                response.failure("Server error")

            elif response.elapsed.total_seconds() > 2:
                response.failure("Request too slow")

            else:
                response.success()

3. Environment-Specific Configuration

import os
from locust import HttpUser, task

class ConfigurableUser(HttpUser):
    host = os.getenv("TARGET_HOST", "http://localhost:3000")
    wait_time = between(
        int(os.getenv("MIN_WAIT", 1)),
        int(os.getenv("MAX_WAIT", 3))
    )

    @task
    def make_request(self):
        self.client.get(os.getenv("ENDPOINT", "/api/data"))

Conclusion

Locust provides a modern, Python-based approach to load testing that excels in code-first workflows, version control integration, and distributed testing. Its simplicity and flexibility make it ideal for developers familiar with Python who want to define performance tests as code.

Choose Locust when:

Team is comfortable with Python
Tests as code is preferred
Modern, web-based UI desired
Easy version control needed
Distributed testing required

Choose JMeter when:

Extensive protocol support needed
Team prefers GUI-based test creation
Large existing JMeter infrastructure
Non-HTTP protocols heavily used

For modern development teams practicing DevOps and infrastructure as code, Locust offers the perfect blend of simplicity, power, and maintainability for load testing (as discussed in Artillery Performance Testing: Modern Load Testing with YAML Scenarios).