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:
- Python-Based 🐍: Write tests in familiar Python code
- Distributed Testing 💻: Run tests across multiple machines
- Real-Time Monitoring 📊: Watch performance metrics live
- 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
- 🎯 Start Small: Begin with few users, gradually increase
- 📊 Monitor Everything: Watch CPU, memory, and response times
- 🔄 Test Regularly: Run performance tests in CI/CD
- 🎭 Model Real Behavior: Include think time and varied actions
- 📈 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:
- 💻 Install Locust and run the examples above
- 🏗️ Write load tests for your own applications
- 📚 Explore distributed testing with multiple Locust workers
- 🌟 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! 🎉🚀✨