+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
Home / Tutorials / 🐍 python
Part 208 of 365

📘 Performance Testing: Load Testing with Locust

Master performance testing: load testing with locust in Python with practical examples, best practices, and real-world applications 🚀

🚀Intermediate
30 min read

Prerequisites

  • Basic understanding of programming concepts 📝
  • Python installation (3.8+) 🐍
  • VS Code or preferred IDE 💻

What you'll learn

  • Understand the concept fundamentals 🎯
  • Apply the concept in real projects 🏗️
  • Debug common issues 🐛
  • Write clean, Pythonic code ✨

🎯 Introduction

Welcome to the exciting world of performance testing with Locust! 🎉 In this guide, we’ll explore how to ensure your Python applications can handle real-world traffic without breaking a sweat.

You’ll discover how Locust can help you simulate thousands of users hammering your application simultaneously. Whether you’re building web APIs 🌐, microservices 🖥️, or full-stack applications 📚, understanding performance testing is essential for delivering reliable, scalable software.

By the end of this tutorial, you’ll feel confident writing and running performance tests that reveal how your application behaves under stress! Let’s dive in! 🏊‍♂️

📚 Understanding Performance Testing with Locust

🤔 What is Locust?

Locust is like a swarm of virtual users attacking your application 🐝. Think of it as a stress test for your code - like seeing how many people can fit in an elevator before it starts groaning!

In Python terms, Locust lets you write test scenarios in pure Python code that simulate real user behavior. This means you can:

  • ✨ Test thousands of concurrent users
  • 🚀 Find performance bottlenecks
  • 🛡️ Ensure your app won’t crash on launch day

💡 Why Use Locust?

Here’s why developers love Locust for performance testing:

  1. Python-Based 🐍: Write tests in familiar Python code
  2. Distributed Testing 💻: Run tests across multiple machines
  3. Real-Time Monitoring 📊: Watch performance metrics live
  4. Flexible Scenarios 🎯: Model complex user behaviors

Real-world example: Imagine launching an online store 🛒. With Locust, you can simulate Black Friday traffic before the actual day to ensure your servers won’t melt!

🔧 Basic Syntax and Usage

📝 Installing Locust

First, let’s get Locust installed:

# 👋 Hello, Locust!
pip install locust

🎯 Your First Locust Test

Let’s start with a simple example:

# 🐝 locustfile.py - Your first swarm!
from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
    # ⏱️ Wait 1-3 seconds between tasks
    wait_time = between(1, 3)
    
    @task
    def index_page(self):
        # 🏠 Visit the homepage
        self.client.get("/")
    
    @task(3)  # 🎯 3x more likely to run
    def view_products(self):
        # 🛍️ Browse products
        self.client.get("/products")
    
    @task(2)
    def view_product_details(self):
        # 👀 Check product details
        product_id = 42
        self.client.get(f"/products/{product_id}")

💡 Explanation: Each @task represents something a user might do. The numbers indicate relative frequency - users browse products more than they view the homepage!

🚀 Running Your Test

Launch your swarm with:

# 🎮 Start Locust with web UI
locust -f locustfile.py --host=http://localhost:8000

# 🤖 Or run headless
locust -f locustfile.py --host=http://localhost:8000 --headless -u 100 -r 10

💡 Practical Examples

🛒 Example 1: E-Commerce Load Test

Let’s build a realistic e-commerce test:

# 🛍️ E-commerce user behavior
from locust import HttpUser, task, between
import random

class ShoppingUser(HttpUser):
    wait_time = between(1, 5)
    
    def on_start(self):
        # 🚪 User logs in when starting
        self.login()
    
    def login(self):
        # 🔐 Authenticate the user
        response = self.client.post("/api/login", json={
            "username": f"user{random.randint(1, 1000)}",
            "password": "testpass123"
        })
        
        if response.status_code == 200:
            # 🎉 Save auth token
            self.auth_token = response.json()["token"]
            self.client.headers.update({"Authorization": f"Bearer {self.auth_token}"})
    
    @task(10)
    def browse_catalog(self):
        # 📚 Browse product catalog
        self.client.get("/api/products", name="Browse Catalog")
    
    @task(5)
    def search_products(self):
        # 🔍 Search for specific items
        search_terms = ["laptop", "phone", "headphones", "tablet"]
        term = random.choice(search_terms)
        self.client.get(f"/api/search?q={term}", name="Search Products")
    
    @task(3)
    def add_to_cart(self):
        # 🛒 Add random product to cart
        product_id = random.randint(1, 100)
        self.client.post(f"/api/cart/add", json={
            "product_id": product_id,
            "quantity": random.randint(1, 3)
        }, name="Add to Cart")
    
    @task(1)
    def checkout(self):
        # 💳 Complete purchase
        with self.client.post("/api/checkout", 
                            json={"payment_method": "credit_card"},
                            name="Checkout",
                            catch_response=True) as response:
            if response.status_code == 200:
                # ✅ Purchase successful!
                response.success()
            else:
                # ❌ Something went wrong
                response.failure(f"Checkout failed: {response.text}")

🎯 Try it yourself: Add a task for viewing order history and leaving product reviews!

🎮 Example 2: API Performance Test

Let’s test a REST API thoroughly:

# 🚀 API performance testing
from locust import HttpUser, task, between, events
import json
import time

class APIUser(HttpUser):
    wait_time = between(0.5, 2)
    
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.user_id = None
        self.response_times = []
    
    @task(20)
    def get_users(self):
        # 👥 Fetch user list
        start_time = time.time()
        with self.client.get("/api/users", 
                           catch_response=True) as response:
            response_time = time.time() - start_time
            self.response_times.append(response_time)
            
            if response.elapsed.total_seconds() > 2:
                # ⚠️ Slow response!
                response.failure(f"Too slow: {response.elapsed.total_seconds()}s")
            elif response.status_code == 200:
                # ✅ All good!
                response.success()
    
    @task(10)
    def create_user(self):
        # 🆕 Create new user
        user_data = {
            "name": f"Test User {time.time()}",
            "email": f"test{time.time()}@example.com",
            "role": random.choice(["user", "admin", "moderator"])
        }
        
        with self.client.post("/api/users", 
                            json=user_data,
                            catch_response=True) as response:
            if response.status_code == 201:
                # 🎉 User created!
                self.user_id = response.json()["id"]
                response.success()
            else:
                response.failure(f"Failed to create user: {response.status_code}")
    
    @task(5)
    def update_user(self):
        # ✏️ Update existing user
        if self.user_id:
            update_data = {"status": "active"}
            self.client.patch(f"/api/users/{self.user_id}", 
                            json=update_data,
                            name="Update User")
    
    @task(15)
    def complex_operation(self):
        # 🏗️ Test complex endpoint
        with self.client.post("/api/analytics/report",
                            json={"start_date": "2024-01-01", 
                                  "end_date": "2024-12-31"},
                            timeout=30,
                            catch_response=True) as response:
            if response.status_code == 200:
                # 📊 Check response size
                data = response.json()
                if len(data["results"]) > 0:
                    response.success()
                else:
                    response.failure("No data returned")

# 📊 Custom event handlers
@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    # 📈 Print final statistics
    print("🏁 Test completed!")
    print(f"📊 Total requests: {environment.stats.total.num_requests}")
    print(f"❌ Total failures: {environment.stats.total.num_failures}")

🏆 Example 3: Advanced Load Patterns

Create realistic traffic patterns:

# 🌊 Wave-like traffic patterns
from locust import HttpUser, task, between, LoadTestShape
import math

class StressTestUser(HttpUser):
    wait_time = between(0.5, 1.5)
    
    @task
    def health_check(self):
        # 🏥 Simple health check
        self.client.get("/health")
    
    @task(5)
    def main_endpoint(self):
        # 🎯 Main application endpoint
        headers = {"X-Request-ID": f"locust-{time.time()}"}
        self.client.get("/api/data", headers=headers)

class SteppedLoadShape(LoadTestShape):
    """
    📈 Gradually increase load in steps
    """
    
    step_time = 60  # 🕐 Each step lasts 60 seconds
    step_load = 50  # 👥 Add 50 users per step
    max_users = 500  # 🎯 Maximum users
    
    def tick(self):
        run_time = self.get_run_time()
        
        if run_time > 600:  # 🛑 Stop after 10 minutes
            return None
        
        current_step = math.floor(run_time / self.step_time)
        target_users = min(self.step_load * current_step, self.max_users)
        
        # 🚀 Spawn rate increases with users
        spawn_rate = max(1, target_users / 10)
        
        return (target_users, spawn_rate)

class SpikeLoadShape(LoadTestShape):
    """
    ⚡ Simulate sudden traffic spikes
    """
    
    stages = [
        {"duration": 60, "users": 10, "spawn_rate": 1},    # 🐌 Warm up
        {"duration": 30, "users": 500, "spawn_rate": 50},  # 🚀 Spike!
        {"duration": 120, "users": 500, "spawn_rate": 5},  # 📊 Sustained load
        {"duration": 30, "users": 10, "spawn_rate": 10},   # 📉 Cool down
    ]
    
    def tick(self):
        run_time = self.get_run_time()
        
        for stage in self.stages:
            if run_time < sum([s["duration"] for s in self.stages[:self.stages.index(stage)+1]]):
                return (stage["users"], stage["spawn_rate"])
        
        return None

🚀 Advanced Concepts

🧙‍♂️ Custom Clients and Protocols

Test any protocol, not just HTTP:

# 🔌 Custom protocol testing
from locust import User, task, between
import socket
import time

class TCPUser(User):
    wait_time = between(1, 2)
    
    def on_start(self):
        # 🔗 Establish connection
        self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.client.connect(("localhost", 9999))
    
    def on_stop(self):
        # 👋 Clean up
        self.client.close()
    
    @task
    def send_message(self):
        # 📤 Send custom protocol message
        start_time = time.time()
        
        try:
            message = b"PING\n"
            self.client.send(message)
            
            response = self.client.recv(1024)
            response_time = time.time() - start_time
            
            # 📊 Report to Locust
            self.environment.events.request.fire(
                request_type="TCP",
                name="ping",
                response_time=response_time * 1000,  # Convert to ms
                response_length=len(response),
                exception=None,
                context={}
            )
        except Exception as e:
            # ❌ Report failure
            self.environment.events.request.fire(
                request_type="TCP",
                name="ping",
                response_time=0,
                response_length=0,
                exception=e,
                context={}
            )

🏗️ Correlation and Session Management

Handle complex user flows:

# 🎭 Complex user sessions
from locust import HttpUser, task, between, SequentialTaskSet
import re

class UserJourney(SequentialTaskSet):
    """
    📍 User follows a specific path
    """
    
    def on_start(self):
        # 🏁 Initialize journey
        self.product_ids = []
        self.cart_token = None
    
    @task
    def visit_homepage(self):
        # 🏠 Start at homepage
        response = self.client.get("/")
        
        # 🔍 Extract CSRF token
        csrf_match = re.search(r'csrf-token" content="([^"]+)"', response.text)
        if csrf_match:
            self.csrf_token = csrf_match.group(1)
    
    @task
    def browse_products(self):
        # 🛍️ Get product list
        response = self.client.get("/api/products")
        products = response.json()
        
        # 📝 Save product IDs for later
        self.product_ids = [p["id"] for p in products[:5]]
    
    @task
    def view_product_details(self):
        # 👀 View specific products
        for product_id in self.product_ids[:3]:
            self.client.get(f"/api/products/{product_id}")
            time.sleep(random.uniform(0.5, 2))  # 🤔 Think time
    
    @task
    def add_to_cart(self):
        # 🛒 Add favorite product
        if self.product_ids:
            response = self.client.post("/api/cart/add", 
                                      json={"product_id": self.product_ids[0]},
                                      headers={"X-CSRF-Token": self.csrf_token})
            
            if response.status_code == 200:
                self.cart_token = response.json()["cart_token"]
    
    @task
    def complete_purchase(self):
        # 💳 Checkout
        if self.cart_token:
            self.client.post("/api/checkout",
                           json={"cart_token": self.cart_token,
                                 "payment_method": "test_card"})

class ShoppingUser(HttpUser):
    tasks = [UserJourney]
    wait_time = between(1, 3)

⚠️ Common Pitfalls and Solutions

😱 Pitfall 1: Not Simulating Realistic Behavior

# ❌ Wrong way - unrealistic constant hammering
class BadUser(HttpUser):
    wait_time = constant(0)  # No wait time!
    
    @task
    def hammer_server(self):
        self.client.get("/")  # Just hitting one endpoint

# ✅ Correct way - realistic user behavior
class RealisticUser(HttpUser):
    wait_time = between(1, 5)  # 🤔 Users think between actions
    
    @task(3)
    def browse(self):
        # 📖 Reading content
        self.client.get("/articles")
        time.sleep(random.uniform(2, 8))  # Reading time
    
    @task(1)
    def interact(self):
        # 💬 User interaction
        self.client.post("/api/comment", json={"text": "Great article!"})

🤯 Pitfall 2: Ignoring Error Handling

# ❌ Dangerous - no error handling
class FragileUser(HttpUser):
    @task
    def risky_request(self):
        response = self.client.get("/api/data")
        data = response.json()  # 💥 Might crash if not JSON!

# ✅ Safe - proper error handling
class RobustUser(HttpUser):
    @task
    def safe_request(self):
        with self.client.get("/api/data", catch_response=True) as response:
            try:
                if response.status_code == 200:
                    data = response.json()
                    if "error" in data:
                        response.failure(f"API error: {data['error']}")
                    else:
                        response.success()
                else:
                    response.failure(f"HTTP {response.status_code}")
            except json.JSONDecodeError:
                response.failure("Invalid JSON response")

🎯 Pitfall 3: Resource Leaks

# ❌ Bad - leaking connections
class LeakyUser(HttpUser):
    def on_start(self):
        self.db_conn = create_connection()  # Never closed!

# ✅ Good - proper cleanup
class CleanUser(HttpUser):
    def on_start(self):
        # 🔗 Setup resources
        self.db_conn = create_connection()
    
    def on_stop(self):
        # 🧹 Clean up properly
        if hasattr(self, 'db_conn'):
            self.db_conn.close()

🛠️ Best Practices

  1. 🎯 Start Small: Begin with few users, gradually increase
  2. 📊 Monitor Everything: Watch CPU, memory, and response times
  3. 🔄 Test Regularly: Run performance tests in CI/CD
  4. 🎭 Model Real Behavior: Include think time and varied actions
  5. 📈 Analyze Trends: Look for degradation over time

📊 Performance Testing Strategy

# 🏗️ Comprehensive test strategy
from locust import events
import logging

# 📊 Custom metrics collection
class PerformanceMonitor:
    def __init__(self):
        self.response_times = []
        self.error_count = 0
        
    @events.request.add_listener
    def on_request(self, request_type, name, response_time, response_length, exception, **kwargs):
        if exception:
            self.error_count += 1
            logging.error(f"❌ Request failed: {name} - {exception}")
        else:
            self.response_times.append(response_time)
            
            # ⚠️ Alert on slow responses
            if response_time > 1000:  # > 1 second
                logging.warning(f"🐌 Slow response: {name} took {response_time}ms")
    
    @events.test_stop.add_listener
    def on_test_stop(self, **kwargs):
        # 📈 Print summary statistics
        if self.response_times:
            avg_response = sum(self.response_times) / len(self.response_times)
            p95_response = sorted(self.response_times)[int(len(self.response_times) * 0.95)]
            
            print(f"""
            📊 Performance Summary:
            ✅ Total Requests: {len(self.response_times)}
            ❌ Failed Requests: {self.error_count}
            ⏱️ Average Response: {avg_response:.2f}ms
            📈 95th Percentile: {p95_response:.2f}ms
            """)

# 🚀 Initialize monitoring
monitor = PerformanceMonitor()

🧪 Hands-On Exercise

🎯 Challenge: Build a Social Media Load Test

Create a comprehensive load test for a social media API:

📋 Requirements:

  • ✅ User registration and login flow
  • 🏷️ Create posts with random content
  • 👤 Follow/unfollow other users
  • 📅 Fetch timeline with pagination
  • 🎨 Upload profile pictures (simulate file uploads)

🚀 Bonus Points:

  • Add WebSocket testing for real-time notifications
  • Implement rate limiting detection
  • Create a custom LoadTestShape for viral content simulation

💡 Solution

🔍 Click to see solution
# 🎯 Social media load test solution!
from locust import HttpUser, task, between, events
import random
import string
import base64
from datetime import datetime

class SocialMediaUser(HttpUser):
    wait_time = between(1, 3)
    
    def on_start(self):
        # 🚪 Register and login
        self.username = self.register_user()
        self.login()
        self.following = []
    
    def register_user(self):
        # 📝 Create unique user
        username = f"user_{datetime.now().timestamp()}_{random.randint(1000, 9999)}"
        
        response = self.client.post("/api/register", json={
            "username": username,
            "email": f"{username}@test.com",
            "password": "Test123!",
            "bio": f"🤖 Load test user {username}"
        })
        
        if response.status_code == 201:
            print(f"✅ Registered: {username}")
            return username
        return None
    
    def login(self):
        # 🔐 Authenticate
        if self.username:
            response = self.client.post("/api/login", json={
                "username": self.username,
                "password": "Test123!"
            })
            
            if response.status_code == 200:
                self.token = response.json()["token"]
                self.client.headers.update({
                    "Authorization": f"Bearer {self.token}"
                })
    
    @task(10)
    def create_post(self):
        # ✍️ Write a post
        post_content = {
            "text": f"🎯 Load testing at {datetime.now()} - {random.choice(['Amazing!', 'Testing 123', 'Hello Locust! 🐝'])}",
            "tags": random.sample(["#testing", "#performance", "#locust", "#python"], k=2)
        }
        
        response = self.client.post("/api/posts", json=post_content)
        
        if response.status_code == 201:
            # 💾 Save post ID for later
            self.last_post_id = response.json()["id"]
    
    @task(15)
    def read_timeline(self):
        # 📰 Get timeline with pagination
        page = random.randint(1, 5)
        limit = 20
        
        self.client.get(f"/api/timeline?page={page}&limit={limit}",
                       name="Timeline")
    
    @task(5)
    def follow_user(self):
        # 👥 Follow random user
        target_user = f"user_{random.randint(1, 1000)}"
        
        response = self.client.post(f"/api/follow/{target_user}")
        
        if response.status_code == 200:
            self.following.append(target_user)
            print(f"✅ Now following {target_user}")
    
    @task(3)
    def like_post(self):
        # ❤️ Like random posts
        post_id = random.randint(1, 10000)
        self.client.post(f"/api/posts/{post_id}/like",
                        name="Like Post")
    
    @task(2)
    def upload_profile_picture(self):
        # 📸 Simulate image upload
        fake_image = base64.b64encode(b"fake_image_data_" * 100).decode()
        
        self.client.post("/api/profile/picture",
                        files={"image": ("profile.jpg", fake_image, "image/jpeg")},
                        name="Upload Profile Pic")
    
    @task(8)
    def search_users(self):
        # 🔍 Search functionality
        search_term = random.choice(["test", "user", "python", "locust"])
        self.client.get(f"/api/search/users?q={search_term}",
                       name="Search Users")
    
    @task(1)
    def delete_post(self):
        # 🗑️ Delete own post
        if hasattr(self, 'last_post_id'):
            self.client.delete(f"/api/posts/{self.last_post_id}",
                             name="Delete Post")

# 🌊 Viral content simulation
class ViralContentShape(LoadTestShape):
    """
    📈 Simulate viral content spreading
    """
    
    def tick(self):
        run_time = self.get_run_time()
        
        if run_time < 60:
            # 🐌 Normal activity
            return (50, 5)
        elif run_time < 180:
            # 📈 Content going viral
            viral_users = int(50 * (1.5 ** ((run_time - 60) / 30)))
            return (min(viral_users, 1000), 20)
        elif run_time < 300:
            # 🔥 Peak viral moment
            return (1000, 50)
        elif run_time < 420:
            # 📉 Dying down
            decline_users = int(1000 * (0.8 ** ((run_time - 300) / 30)))
            return (max(decline_users, 50), 10)
        else:
            # 🏁 Back to normal
            return (50, 5)

# 📊 Performance tracking
@events.request.add_listener
def track_performance(request_type, name, response_time, response_length, exception, **kwargs):
    if response_time > 2000:  # > 2 seconds
        print(f"⚠️ Slow request: {name} took {response_time}ms")
    
    if exception:
        print(f"❌ Failed: {name} - {exception}")

🎓 Key Takeaways

You’ve learned so much about performance testing with Locust! Here’s what you can now do:

  • ✅ Write load tests in pure Python code 💪
  • ✅ Simulate realistic user behavior with tasks and wait times 🛡️
  • ✅ Monitor performance metrics in real-time 🎯
  • ✅ Create complex test scenarios with custom shapes 🐛
  • ✅ Find performance bottlenecks before they impact users! 🚀

Remember: Performance testing isn’t about breaking your app - it’s about making it unbreakable! 🤝

🤝 Next Steps

Congratulations! 🎉 You’ve mastered performance testing with Locust!

Here’s what to do next:

  1. 💻 Install Locust and run the examples above
  2. 🏗️ Write load tests for your own applications
  3. 📚 Explore distributed testing with multiple Locust workers
  4. 🌟 Share your performance improvements with your team!

Remember: Every high-performance application started with someone who cared enough to test it properly. Keep testing, keep improving, and most importantly, keep your users happy! 🚀

Happy testing! 🎉🚀✨

Previous Tutorial
📘 API Testing: Testing REST APIs
Next Tutorial
📘 Property-Based Testing: Hypothesis Framework

More python Tutorials

Part 207

📘 API Testing: Testing REST APIs

🚀Intermediate • 30 min read
Part 208 - Currently Reading

📘 Performance Testing: Load Testing with Locust

🚀Intermediate • 30 min read
Part 209

📘 Property-Based Testing: Hypothesis Framework

🚀Intermediate • 30 min read
View all 365 python tutorials →

Tutorial Info

Series Progress
Part 208 of 365 57%
Category
🐍 python
Difficulty
🚀Intermediate
Duration
30 min read
Published
July 4, 2025