Designing a Locust test plan that accurately reflects real-world user behavior is the hardest part of performance testing.
Let’s see Locust in action. Imagine we’re testing a simple API endpoint that fetches user data.
from locust import HttpUser, task, between
class UserApiUser(HttpUser):
wait_time = between(1, 5) # Users wait 1-5 seconds between tasks
@task
def get_user_data(self):
user_id = 1001 # Example user ID
self.client.get(f"/users/{user_id}")
@task(2) # This task is twice as likely to be run
def get_user_posts(self):
user_id = 1001
self.client.get(f"/users/{user_id}/posts")
In this example, UserApiUser is our simulated user. wait_time defines the think time between user actions. The @task decorator marks methods as user tasks. The number in parentheses, (2), is a weight; get_user_posts will be executed roughly twice as often as get_user_data. This is a rudimentary way to model different user activities, but it’s a start.
The core problem Locust solves is simulating many users interacting with your application concurrently. It uses Python, making it flexible and familiar. A "user" in Locust is a Python class that inherits from HttpUser (for HTTP services) or User (for other protocols). Each instance of this class represents a single simulated user.
The HttpUser class has a client attribute, which is an instance of HttpSession (a wrapper around requests.Session). This client is what your simulated users use to make requests to your application. Tasks are defined as methods within the HttpUser class, decorated with @task. Locust picks tasks to run based on their weights.
Here’s how you’d run this:
- Save the code above as
locustfile.py. - Install Locust:
pip install locust. - Run Locust from your terminal:
locust -f locustfile.py --host http://localhost:8080. - Open your browser to
http://localhost:8080.
You’ll see a web UI where you can enter the number of users to simulate and the spawn rate. As users start, they’ll execute the tasks defined in your locustfile.py.
To design realistic load scenarios, we need to go beyond simple task weights. Think about user journeys. A user doesn’t just hit one API endpoint; they navigate through a site, perform a sequence of actions.
Consider a typical e-commerce checkout flow:
- Browse products.
- Add to cart.
- View cart.
- Proceed to checkout.
- Enter shipping details.
- Enter payment details.
- Place order.
We can model this using Locust’s SequentialTaskSet.
from locust import HttpUser, task, between, SequentialTaskSet
class CheckoutFlow(SequentialTaskSet):
@task
def browse_products(self):
self.client.get("/products")
@task
def add_to_cart(self):
self.client.post("/cart/add", json={"product_id": 123})
@task
def view_cart(self):
self.client.get("/cart")
@task
def proceed_to_checkout(self):
self.client.post("/checkout/start")
class WebsiteUser(HttpUser):
wait_time = between(2, 10)
tasks = [CheckoutFlow]
Here, CheckoutFlow ensures tasks are executed in the defined order. WebsiteUser then uses this CheckoutFlow as its sole task set. This means every WebsiteUser will go through the entire checkout process sequentially.
What if users don’t always complete the checkout? What if some abandon their carts? We can introduce conditional logic or other task sets.
from locust import HttpUser, task, between, SequentialTaskSet, TaskSet, events
import random
class BrowseFlow(TaskSet):
@task
def view_product_details(self):
self.client.get("/products/random")
class AddToCartFlow(TaskSet):
@task
def add_item(self):
self.client.post("/cart/add", json={"product_id": random.randint(1, 1000)})
class CheckoutFlow(SequentialTaskSet):
# ... (previous checkout tasks) ...
@task
def place_order(self):
self.client.post("/order/place")
class WebsiteUser(HttpUser):
wait_time = between(1, 3)
# Randomly choose between browsing, adding to cart, or attempting checkout
tasks = [BrowseFlow, AddToCartFlow, CheckoutFlow]
def on_start(self):
# Simulate new user session, maybe clear cart
self.client.cookies.clear()
print("New user starting...")
def on_stop(self):
print("User stopping...")
In this modified WebsiteUser, tasks = [BrowseFlow, AddToCartFlow, CheckoutFlow] means Locust will pick one of these TaskSets for each user and execute its tasks. If a user is assigned CheckoutFlow, they will run through the sequential checkout. If they get BrowseFlow, they’ll just browse. This random selection of task sets for each user instance helps model a more diverse user base.
The on_start and on_stop methods are hooks that run once when a user instance is created and when it’s stopped, respectively. They are invaluable for setting up initial conditions (like logging in or clearing cookies) or performing cleanup.
The most powerful way to model realistic behavior is to use data. What if user IDs aren’t always 1001? What if they vary? You can read from CSV files, databases, or external APIs to fetch dynamic data for your requests.
For instance, to simulate fetching data for different users:
from locust import HttpUser, task, between, constant
import csv
import os
class WebsiteUser(HttpUser):
wait_time = between(1, 5)
host = "http://localhost:8080"
user_ids = []
user_file = "user_ids.csv"
def on_start(self):
if not self.user_ids:
try:
with open(self.user_file, mode='r') as file:
reader = csv.reader(file)
# Skip header if it exists
header = next(reader, None)
for row in reader:
self.user_ids.append(row[0]) # Assuming user IDs are in the first column
if not self.user_ids:
print(f"Warning: No user IDs found in {self.user_file}. Using default.")
self.user_ids = ["default_user_1", "default_user_2"]
except FileNotFoundError:
print(f"Error: {self.user_file} not found. Using default user IDs.")
self.user_ids = ["default_user_1", "default_user_2"]
except Exception as e:
print(f"An error occurred reading {self.user_file}: {e}. Using default user IDs.")
self.user_ids = ["default_user_1", "default_user_2"]
if not self.user_ids:
raise Exception("Could not load any user IDs.")
print(f"Loaded {len(self.user_ids)} user IDs for simulation.")
@task
def get_user_profile(self):
if not self.user_ids:
user_id = "fallback_user"
else:
user_id = random.choice(self.user_ids)
self.client.get(f"/users/{user_id}")
Create a user_ids.csv file with content like:
UserID
user_abc
user_def
user_ghi
When this locustfile.py runs, each WebsiteUser instance will load user IDs from the CSV on startup. Then, in the get_user_profile task, a random user_id from the loaded list is chosen for each request. This is critical for realistic testing because it avoids hitting your application with the same few IDs repeatedly, which might not expose caching or database contention issues that arise with a wide variety of data.
The random.choice here is a simple distribution. For more complex distributions (e.g., some users are "VIP" and hit endpoints more often, or some user IDs are more common than others), you’d implement more sophisticated logic, perhaps using random.choices with weights or custom probability functions.
The host attribute can also be set to None and specified via the command line (--host) or the UI, which is useful for testing against different environments.
A common pitfall when designing realistic scenarios is assuming a uniform distribution of user actions or user types. In reality, user behavior is often skewed. A small percentage of users might perform very complex or frequent actions, while the majority perform simpler, less frequent ones. Using task weights and carefully selecting task sets is how you approximate this skew.
The true power of Locust lies in its Pythonic nature. You can integrate with external data sources, use Python’s standard libraries, and even call other services to fetch realistic test data or determine user behavior dynamically during the test. This means your load tests can evolve with your application’s complexity, mirroring production traffic patterns far more closely than static, predefined scripts.
When defining wait_time, remember that between(min_seconds, max_seconds) creates a uniform distribution. If your users have a more specific "think time" pattern (e.g., most wait 5 seconds, but some wait 30), you might need to implement custom wait time logic using a method and random.randint or other distribution functions.
The next step in creating sophisticated load tests is to simulate user sessions that have state, like a logged-in user or items in a shopping cart, across multiple requests within a SequentialTaskSet or by passing state between TaskSets.