Locust for Load Testing: Python-Based Performance Testing
Learn how to use Locust for load testing with Python, from basic scripts to distributed testing across multiple machines.
Mark
Performance Testing Expert
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 statisticsresults_failures.csv- Failed requestsresults_stats_history.csv- Time-series datareport.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.
Tags: