Testing

Locust Load Testing: Python Performance Testing Tutorial

HelpMeTest

HelpMeTest

— 4 min read
Locust Load Testing: Python Performance Testing Tutorial

Locust is a Python-based load testing tool that lets you define user behavior as plain Python code. If your team writes Python, Locust is the most natural load testing choice — no XML, no proprietary scripting language, just standard Python classes.

What Is Locust?

Locust is an open-source load testing framework where you define test scenarios as Python classes. Virtual users ("locusts") execute tasks you define, and the framework tracks response times, error rates, and throughput. It offers both a web UI for interactive testing and a headless mode for CI/CD.

Key features:

  • Pure Python — define behavior with classes and decorators
  • Web UI — real-time dashboard for monitoring tests
  • Headless mode — fully scriptable for CI pipelines
  • Distributed — scale across multiple machines natively
  • Extensible — events, hooks, custom clients

Installing Locust

Requires Python 3.9+.

pip install locust

Verify:

locust --version

Writing Your First Locust Test

Locust tests live in a file (by default locustfile.py). Users extend HttpUser and define tasks:

# locustfile.py
from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)  # wait 1-3 seconds between tasks

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

    @task(3)  # weight: runs 3x more often than weight-1 tasks
    def view_products(self):
        self.client.get("/products")

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

Run with the web UI:

locust --host https://example.com

Open http://localhost:8089 to see the dashboard, enter user count and spawn rate, and start the test.

Understanding Task Weights

The integer argument to @task(N) sets relative frequency. A task with @task(3) runs three times more often than @task(1):

class ShopUser(HttpUser):
    wait_time = between(0.5, 2)

    @task(10)
    def browse(self):
        self.client.get("/products")

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

    @task(1)
    def checkout(self):
        self.client.post("/checkout", json={"payment": "test"})

This models realistic user behavior: most users browse, fewer add to cart, fewer still checkout.

Testing REST APIs

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

class APIUser(HttpUser):
    wait_time = between(0.5, 1.5)
    token = None

    def on_start(self):
        """Called once when each user starts."""
        response = self.client.post("/auth/login", json={
            "username": "testuser",
            "password": "testpass",
        })
        if response.status_code == 200:
            self.token = response.json()["token"]
        else:
            raise RescheduleTask()  # retry on failure

    @task
    def list_items(self):
        self.client.get(
            "/api/items",
            headers={"Authorization": f"Bearer {self.token}"},
            name="/api/items [GET]",
        )

    @task
    def create_item(self):
        self.client.post(
            "/api/items",
            json={"name": f"item-{self.user_id}", "value": 42},
            headers={"Authorization": f"Bearer {self.token}"},
            name="/api/items [POST]",
        )

The name parameter groups similar URLs in the statistics table — useful when URLs contain dynamic IDs.

Handling Dynamic URLs

Without grouping, every /api/items/123, /api/items/456 appears as a separate entry:

# Bad: creates hundreds of separate statistics entries
self.client.get(f"/api/items/{item_id}")

# Good: group them
self.client.get(f"/api/items/{item_id}", name="/api/items/[id]")

Assertions and Error Handling

By default, Locust marks any non-2xx response as a failure. You can customize this:

@task
def search(self):
    with self.client.get("/search?q=test", catch_response=True) as response:
        if response.status_code == 200:
            if "results" not in response.json():
                response.failure("No 'results' key in response")
            else:
                response.success()
        elif response.status_code == 404:
            response.success()  # 404 is expected for missing items
        else:
            response.failure(f"Unexpected status: {response.status_code}")

Running Locust Headless (CI/CD)

locust \
  --headless \
  --host https://api.example.com \
  --users 100 \
  --spawn-rate 10 \
  --run-time 60s \
  --html report.html \
  --csv results

Exit code is 0 on success. To fail on performance thresholds, use custom event hooks (see below) or check the CSV results in your CI script.

Custom Failure Thresholds

Locust doesn't natively support threshold-based pass/fail the way k6 does, but you can implement it with events:

from locust import events

@events.quitting.add_listener
def check_thresholds(environment, **kwargs):
    stats = environment.runner.stats.total
    
    # Fail if error rate > 1%
    if stats.fail_ratio > 0.01:
        print(f"FAIL: error rate {stats.fail_ratio:.1%} exceeds 1%")
        environment.process_exit_code = 1
    
    # Fail if 95th percentile > 500ms
    if stats.get_response_time_percentile(0.95) > 500:
        print(f"FAIL: p95 {stats.get_response_time_percentile(0.95)}ms exceeds 500ms")
        environment.process_exit_code = 1

Sequential Tasks with TaskSet

For multi-step flows, use TaskSet:

from locust import HttpUser, TaskSet, task, between

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

    @task
    def apply_coupon(self):
        self.client.post("/cart/coupon", json={"code": "TEST10"})

    @task
    def place_order(self):
        self.client.post("/checkout")
        self.interrupt()  # exit the TaskSet

class ShopUser(HttpUser):
    wait_time = between(1, 2)
    tasks = [CheckoutFlow]

Distributed Locust Testing

Scale Locust across multiple machines:

Master node:

locust --master --host https://example.com

Worker nodes:

locust --worker --master-host 192.168.1.100

The web UI on the master coordinates all workers. Workers pick up tasks automatically.

GitHub Actions Integration

name: Load Test
on:
  push:
    branches: [main]

jobs:
  locust:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Install Locust
        run: pip install locust
      
      - name: Run load test
        run: |
          locust \
            --headless \
            --host https://staging.example.com \
            --users 50 \
            --spawn-rate 5 \
            --run-time 120s \
            --html load-report.html
      
      - name: Upload report
        uses: actions/upload-artifact@v3
        if: always()
        with:
          name: load-test-report
          path: load-report.html

Key Metrics to Watch

The Locust web UI and CSV output include:

Metric What it means
RPS Requests per second (throughput)
Failures/s Failed requests per second
Median 50th percentile response time
95%ile 95th percentile — what most users experience
99%ile Tail latency — worst-case experience
Max Slowest single request (often an outlier)

Focus on the 95th percentile for SLA definitions, not the average — averages hide outliers.

Pair Load Testing with Functional Testing

Locust tells you how your system handles concurrency. It doesn't verify that responses are semantically correct under load. HelpMeTest handles that layer — AI-powered test automation that validates user journeys and monitors your application 24/7.

Use Locust for throughput and performance validation. Use HelpMeTest for functional correctness and continuous monitoring.

Start free with HelpMeTest — 10 tests, no code required, monitoring every 5 minutes.

Read more