Python Locust Testing: Load Testing Made Simple

A comprehensive guide to performance testing with Locust - from basics to advanced techniques

Locust is a powerful and easy-to-use load testing tool written in Python. In this article, I'll share my experience using Locust to test website and API performance, from basic concepts to advanced techniques.

What is Locust?

Locust is an open-source framework for load testing written in Python. It allows you to write test scenarios using Python code, making it easier to create complex test cases compared to traditional tools like Apache JMeter.

Advantages of Locust:

Write tests in Python - easy to learn and maintain
Intuitive web UI for test monitoring
Distributed testing - can run tests on multiple machines
Real-time metrics and charts
Support for HTTP, WebSocket and other protocols

Installation and Setup

# Install Locust
pip install locust

# Or use requirements.txt
echo "locust>=2.15.0" >> requirements.txt
pip install -r requirements.txt

Basic Structure of a Locust Test

A basic Locust test file has the following structure:

from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)  # Wait time between requests
    
    def on_start(self):
        """Runs when user starts testing"""
        pass
    
    @task
    def index_page(self):
        """Test homepage"""
        self.client.get("/")
    
    @task(2)  # Weight 2 - runs twice as often as other tasks
    def view_item(self):
        """Test viewing product"""
        self.client.get("/item/1")

Real-world Example: Testing API with Authentication

Based on my real-world experience, here's an example of testing an API with authentication:

import random
from locust import HttpUser, task, between

USER_CREDENTIALS = [
    ("user1", "password1"),
    ("user2", "password2")
]

class APIUser(HttpUser):
    wait_time = between(15, 30)  # Simulate user think time
    
    def on_start(self):
        """Login when starting test"""
        self.access_token = None
        self.user_id = None
        self.login()
        self.get_user_info()
    
    def login(self):
        """Login and store access token"""
        username, password = random.choice(USER_CREDENTIALS)
        response = self.client.post("/api/v1/auth/login", json={
            "username": username,
            "password": password
        })
        
        if response.status_code == 200:
            data = response.json()
            self.access_token = data.get("data", {}).get("accessToken")
    
    def get_auth_headers(self):
        """Helper to create Authorization header"""
        return {"Authorization": f"Bearer {self.access_token}"} if self.access_token else {}
    
    @task(1)
    def get_user_info(self):
        """Get user information"""
        response = self.client.get("/api/v1/auth/me", 
                                 headers=self.get_auth_headers())
        if response.ok:
            data = response.json().get("data", {})
            self.user_id = data.get("id")
    
    @task(2)
    def get_dashboard_stats(self):
        """Get dashboard statistics"""
        if self.user_id:
            params = {"userId": self.user_id}
            response = self.client.get("/api/v1/dashboard/stats", 
                                     params=params, 
                                     headers=self.get_auth_headers())

Important Concepts

1. User Classes

User classes define the behavior of virtual users. Each user will run tasks according to the defined order and timing.

2. Tasks

Tasks are actions that users perform. Use the @task decorator to mark a method as a task.

3. Wait Time

wait_time defines the waiting time between tasks:

between(min, max): Random time between min and max seconds
constant(wait_time): Fixed waiting time
constant_throughput(rate): Maintain fixed throughput

4. Task Weighting

Use @task(weight) to define the frequency of each task. Tasks with higher weights will run more often.

Running Locust Tests

Command Line

# Run test with locustfile.py
locust -f locustfile.py

# Run with specific host
locust -f locustfile.py --host=http://localhost:8000

# Run in headless mode (no web UI)
locust -f locustfile.py --host=http://localhost:8000 --headless --users 10 --spawn-rate 2 --run-time 60s

Web UI

After running the command above, open your browser and visit http://localhost:8089 to view Locust's web UI.

Monitoring and Metrics

Locust provides important real-time metrics:

RPS (Requests Per Second): Number of requests per second
Response Time: Average response time, 95th percentile
Failure Rate: Rate of failed requests
Active Users: Number of active users

Distributed Testing

To run tests on multiple machines: Master node:

locust -f locustfile.py --master

Worker nodes:

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

Best Practices

Think Time: Always add wait_time to simulate real user behavior
Data Management: Use random data to avoid cache effects
Error Handling: Handle errors gracefully and log thoroughly
Resource Cleanup: Clean up resources in on_stop()
Monitoring: Monitor system resources (CPU, memory, network) when running tests

Advanced Example: Testing with Custom Data

import csv
import random
from locust import HttpUser, task, between

class AdvancedAPIUser(HttpUser):
    wait_time = between(1, 5)
    
    def on_start(self):
        """Load test data from CSV file"""
        self.test_data = self.load_test_data()
        self.current_data = random.choice(self.test_data)
    
    def load_test_data(self):
        """Load data from CSV file"""
        data = []
        with open('test_data.csv', 'r') as file:
            reader = csv.DictReader(file)
            for row in reader:
                data.append(row)
        return data
    
    @task
    def create_resource(self):
        """Create resource with random data"""
        payload = {
            "name": f"Test_{random.randint(1000, 9999)}",
            "description": "Generated by Locust test",
            "data": self.current_data
        }
        
        response = self.client.post("/api/resources", 
                                  json=payload,
                                  headers={"Content-Type": "application/json"})
        
        if response.status_code == 201:
            # Save ID for use in subsequent requests
            self.resource_id = response.json().get("id")
    
    @task
    def update_resource(self):
        """Update created resource"""
        if hasattr(self, 'resource_id'):
            payload = {"status": "updated"}
            self.client.put(f"/api/resources/{self.resource_id}", 
                           json=payload)

Conclusion

Locust is a powerful tool for load testing, especially suitable for Python developers. With simple syntax, intuitive web UI, and scalability, Locust makes performance testing easier and more effective. When to use Locust:

You want to write tests in Python
Need to test APIs or web applications
Want a web UI to monitor tests
Need distributed testing
Want to customize complex test logic

When not to use Locust:

Testing protocols not supported (like gRPC, custom protocols)
Need to record and replay user actions
Team lacks Python experience

With Locust, you can easily create complex test scenarios and monitor system performance in real-time. This is an essential tool in the toolkit of any DevOps engineer or developer concerned with performance testing.
Ngày đăng: Aug. 24, 2025

8 total views

Comment

Hiện tại chưa có comment nào...

Sấu Gấu Blog