Locust’s event system lets you inject custom logic into your load tests at precisely the right moments, transforming a simple script into a sophisticated testing framework.
Let’s see it in action. Imagine you want to log the response time of every single GET request to a specific endpoint, say /api/users.
from locust import HttpUser, task, events
@events.request.add_listener
def log_user_request_times(request_type, name, response_time, response_length, **kwargs):
if request_type == "GET" and name == "/api/users":
print(f"GET /api/users: Response Time = {response_time} ms, Length = {response_length}")
class UserBehavior(HttpUser):
host = "http://localhost:8089" # Replace with your target host
@task
def get_users(self):
self.client.get("/api/users")
@task
def get_products(self):
self.client.get("/api/products")
When you run this Locust file, you’ll see output like this interleaved with Locust’s standard output:
GET /api/users: Response Time = 123 ms, Length = 543
GET /api/products: Response Time = 87 ms, Length = 1209
GET /api/users: Response Time = 155 ms, Length = 543
...
This is the request event firing. Locust has several such events you can hook into: init, start_user, stop_user, request, report, and quit. Each is a crucial juncture for custom actions.
The init event is perfect for setting up global resources or configurations before any users start. For instance, you might initialize a database connection pool or load a configuration file.
from locust import HttpUser, task, events
import logging
@events.init.add_listener
def on_locust_init(environment, **kwargs):
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logging.info("Locust environment initialized. Starting custom logging.")
class WebsiteUser(HttpUser):
host = "http://localhost:8089"
@task
def index(self):
self.client.get("/")
When Locust starts, you’ll see your custom log message:
2023-10-27 10:30:00,123 - INFO - Locust environment initialized. Starting custom logging.
The start_user and stop_user events are tied to the lifecycle of individual Locust users (greenlets). start_user fires when a user instance begins its execution, and stop_user fires when it’s about to be terminated. This is invaluable for managing user-specific state, like creating a unique session ID for each user or cleaning up user-specific resources.
from locust import HttpUser, task, events
import uuid
@events.start_user.add_listener
def on_start_user(environment, **kwargs):
user_id = uuid.uuid4()
environment.user_data = {"user_id": user_id} # Store custom data on the environment
print(f"User {user_id} started.")
@events.stop_user.add_listener
def on_stop_user(environment, **kwargs):
user_id = getattr(environment, 'user_data', {}).get('user_id', 'N/A')
print(f"User {user_id} stopping.")
class UserSession(HttpUser):
host = "http://localhost:8089"
@task
def my_task(self):
user_id = self.environment.user_data.get("user_id")
print(f"Executing task for user: {user_id}")
self.client.get("/")
Output might look like:
User 123e4567-e89b-12d3-a456-426614174000 started.
Executing task for user: 123e4567-e89b-12d3-a456-426614174000
User abcdef01-2345-6789-abcd-ef0123456789 started.
Executing task for user: abcdef01-2345-6789-abcd-ef0123456789
User 123e4567-e89b-12d3-a456-426614174000 stopping.
The request event, as shown earlier, is triggered for every HTTP request made by the HttpUser client. It provides detailed information about the request and response, making it the go-to for response time analysis, custom assertion logic, or logging specific request details.
The report event fires at the end of each statistics collection interval (typically every few seconds). This is where you can aggregate custom metrics, generate reports, or trigger alerts based on the current performance data.
from locust import HttpUser, task, events
from collections import defaultdict
# Store custom stats in a dictionary attached to the environment
request_counts = defaultdict(int)
@events.report.add_listener
def on_report_interval(stats, **kwargs):
# Access the custom stats and do something with them
if request_counts:
print("\n--- Custom Report ---")
for name, count in request_counts.items():
print(f"Custom Count for {name}: {count}")
print("---------------------\n")
@events.request.add_listener
def custom_request_counter(request_type, name, response_time, response_length, **kwargs):
# Increment custom counter for each request
request_counts[name] += 1
class ReportingUser(HttpUser):
host = "http://localhost:8089"
@task
def get_home(self):
self.client.get("/")
@task
def get_about(self):
self.client.get("/about")
When running, you’ll see your custom report output periodically:
--- Custom Report ---
Custom Count for /: 15
Custom Count for /about: 12
---------------------
Finally, the quit event is called when Locust is shutting down. It’s your last chance to perform any cleanup, save final results, or send notifications.
The real power of hooks comes from their ability to decouple your core load testing logic from cross-cutting concerns like logging, monitoring, and custom validation. You can write a standard Locust file and then add hooks to augment its behavior without altering the tasks themselves.
What most people don’t realize is that the environment object passed to most event listeners is a powerful central hub. You can attach arbitrary attributes to environment (as seen in on_start_user and on_report_interval) to share state across different hooks or even make it accessible within your user classes. This allows for complex, stateful extensions that can react to and influence the entire test run.
The next logical step is to explore how these events can be used to integrate with external systems, such as sending alerts to Slack or pushing metrics to Prometheus.