Locust is an open-source load testing and performance testing framework designed to measure how systems behave under load using scenarios primarily written in Python. It allows you to simulate user behavior to evaluate the scalability of web applications, APIs, and other systems.
You can upload your Locust test files as a ZIP through the Loadium interface. Running on Loadium will enable your Locust tests to run tests across multiple regions and handle high load levels.
Loadium uses Locust version 2.29.1. If you require a different version, you can forward your request to Loadium's support.
Enter your configurations (test name, virtual user amount, load regions etc.)
The locustfile.py file in the uploaded ZIP will be launched on the server. Therefore, the "main" file (where the test execution steps are placed) of your test must be named "locustfile.py".
If your test depends on other files (such as reading data from CSV files), you should import them using the path format:
/locust/tests-files/{file path}
Example: Use import(/locust/tests-files/abc.csv) instead of import(abc.csv)
Basic Example
# locustfile.py
from locust import HttpUser, TaskSet, task, between
import json
class UserBehavior(TaskSet):
"""
Defines the tasks that each simulated user will perform.
"""
@task(2) # This task has a weight of 2
def get_items(self):
"""
Perform a GET request to retrieve items.
"""
self.client.get("/api/items", name="GET /api/items")
@task(1) # This task has a weight of 1
def create_item(self):
"""
Perform a POST request to create a new item.
"""
payload = {
"name": "New Item",
"description": "This is a new item.",
"price": 19.99
}
headers = {'Content-Type': 'application/json'}
self.client.post("/api/items", data=json.dumps(payload), headers=headers, name="POST /api/items")
def on_start(self):
"""
Code to run when a simulated user starts executing tasks.
Useful for setup tasks like authentication.
"""
# Example: Authenticate and store a token
# response = self.client.post("/api/login", json={"username": "testuser", "password": "testpass"})
# if response.status_code == 200:
# self.token = response.json()["token"]
# self.client.headers.update({"Authorization": f"Bearer {self.token}"})
pass
def on_stop(self):
"""
Code to run when a simulated user stops executing tasks.
Useful for teardown tasks like logging out.
"""
pass
class WebsiteUser(HttpUser):
"""
Defines the user class that will run the tasks.
"""
tasks = [UserBehavior]
wait_time = between(1, 5) # Simulate a wait time between 1 and 5 seconds between tasks
host = "http://your-api-domain.com" # Replace with your target host
CSV Usage Example
import csv
import logging
from locust import HttpUser, SequentialTaskSet, task, between
logger = logging.getLogger(__name__)
class UserBehaviour(SequentialTaskSet):
def __init__(self, parent):
super().__init__(parent)
self.credentials = [] # Store user credentials from CSV
self.current_index = 0 # Track the next user in a round-robin manner
def load_credentials_from_csv(self):
""" Load user credentials from CSV file """
with open('/locust/test-files/testUser.csv', mode='r') as file: #This filepath is important
reader = csv.DictReader(file, delimiter=';')
self.credentials = list(reader)
logger.info(f"Loaded {len(self.credentials)} users from CSV.")
def get_next_user(self):
""" Retrieve the next user from the credential list in a cyclic manner """
user = self.credentials[self.current_index]
self.current_index = (self.current_index + 1) % len(self.credentials) # Loop back if at the end
return user
@task
def login(self):
""" Perform a login request with user credentials """
user = self.get_next_user() # Get next user credentials
response = self.client.post(
f"/index.php?route=account/login.login&language=en-gb&login_token={self.login_token}",
name="Login",
verify=False,
data={
"email": user["mail"],
"password": user["password"]
}
)
Verify.check_response(response)
logger.info(f"Login request with Username: {user['mail']}, Password: {user['password']}")
class MyUser(HttpUser):
""" Define user behavior and test configuration """
wait_time = between(1, 2)
host = "https://demo-opencart.com"
tasks = [UserBehaviour]
Logging Example
import logging
from locust import HttpUser, task, between
# Configure logger
logging.basicConfig(
level=logging.INFO, # Set logging level
format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__) # Create logger instance
class UserBehavior(HttpUser):
wait_time = between(1, 2) # Random wait time between requests
host = "https://example.com" # Target website
@task
def example_task(self):
""" Send a simple GET request """
response = self.client.get("/") # Perform GET request
if response.status_code == 200:
logger.info("Request successful") # Log success
else:
logger.error(f"Request failed with status {response.status_code}") # Log failure
# Example log messages
logger.debug("This is a debug message") # Debug message
logger.info("This is an info message") # Info message
logger.warning("This is a warning message") # Warning message
logger.error("This is an error message") # Error message
logger.critical("This is a critical message") # Critical error message
Logs in Loadium Report
Running the test
To start the test, click the play button on the relevant screen. After a final review on the summarization window, you can initiate the execution.
After approximately 30 seconds, load generators will be ready and you’ll be taken to the report screen and the test will start.
In the report screen, you can click the red “X” button at the top at any time to stop the test. If you do not stop the test manually, it will run for the duration you specified and then stop automatically.
On the report screen, the data coming from your selected regions is displayed in 1-second intervals and visualized in charts. If you wish, you can export these reports. Your reports are stored and protected by Loadium according to your subscription plan.
