TL;DR: Locust is a Python load testing framework. Write user scenarios as Python classes, run tests via web UI or CLI, scale with distributed mode. Perfect for Python teams wanting code-based performance tests with CI/CD integration.

Locust is a Python-based open-source load testing framework that enables engineers to define complex user behavior scenarios as Python code rather than XML configurations or GUI workflows. With over 24,000 GitHub stars and active maintenance, it has become the preferred load testing tool for Python-centric teams. According to the JetBrains Developer Ecosystem Survey 2024, Python is used by 51% of developers for test automation — making Locust a natural fit for teams already invested in the Python ecosystem. Unlike JMeter’s thread-based model, Locust uses greenlets (lightweight coroutines) enabling a single machine to simulate thousands of concurrent users with minimal memory. This comprehensive guide covers Locust from first test to distributed production load testing.

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 provides a code-first approach with powerful distributed testing capabilities and real-time web-based monitoring.

Locust fits naturally into continuous testing in DevOps workflows and complements API performance testing strategies. When building a comprehensive test automation strategy, Locust’s Python-based approach integrates seamlessly with CI/CD pipeline optimization tooling.

Basic Load Test 

# locustfile.py
from locust 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

FeatureLocustJMeter
LanguagePythonGUI/XML
Learning CurveLow (if Python known)Medium-High
Test as CodeYes (native)Limited (requires plugins)
Distributed TestingBuilt-inBuilt-in
Real-time UIWeb-based, modernJava Swing, dated
Version ControlExcellent (Python files)Poor (XML files)
Resource UsageLowerHigher (Java)
Protocol SupportHTTP/WebSocket (extensible)Extensive built-in
CI/CD IntegrationExcellentGood
Scripting FlexibilityExcellent (Python)Limited (Groovy/BeanShell)
CommunityGrowingLarge, established
LicenseMITApache 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.

Official Resources

“What I love about Locust is that the test IS the documentation. A well-written locustfile shows you exactly what user workflows the system was tested against. That clarity is invaluable when a performance regression appears six months later.” — Yuri Kan, Senior QA Lead

FAQ

What is Locust and how does it work?

Locust is a Python load testing framework where user behavior is defined as Python classes with task methods executed concurrently via greenlets.

Locust uses Python’s gevent library (greenlets) for concurrency rather than threads, allowing thousands of virtual users per process with minimal overhead. You define a User class with @task-decorated methods representing user actions. Locust randomly selects and executes tasks based on their weight, simulating realistic user behavior patterns. The web UI shows real-time metrics; CLI mode enables headless execution in CI/CD.

How does Locust compare to JMeter?

Locust uses Python code for readable, versionable tests; JMeter uses XML with GUI. Locust wins on developer experience; JMeter on protocol variety.

Locust tests are Python classes — readable in PRs, easy to parameterize with Python logic, and straightforward to maintain. JMeter tests are XML (hard to review) with a GUI (not CI/CD friendly). Locust excels for HTTP API testing with complex user simulation logic. JMeter supports more protocols (JDBC, JMS, FTP, SMTP) and has a larger enterprise ecosystem. Performance-wise, both handle similar loads per machine.

How do I run distributed Locust tests?

Start one master process and multiple worker processes. Workers connect to master and receive user simulation tasks automatically.

Distributed Locust setup: start master with locust --master -f locustfile.py, start workers on other machines with locust --worker --master-host=MASTER_IP -f locustfile.py. The master distributes users evenly across workers. For cloud-based distributed testing, use Docker with a master service and scaled worker services, or Kubernetes with a master deployment and worker deployment with replicas: N.

What metrics does Locust report?

Requests/s, response time percentiles (median, 95th), failure rate, and user count. Export to CSV or integrate with Grafana for dashboards.

Locust reports per-endpoint and aggregate metrics: request count, failure count, response time (median, 90th, 95th, 99th percentiles), requests/s, and failures/s. The web UI shows real-time charts. Use --csv=results to export summary and per-request CSVs. For production monitoring dashboards, the locust-grafana plugin sends metrics to InfluxDB for Grafana visualization.

See Also