đź““
Loadium Wiki
  • Welcome to Loadium
  • Getting Started with Loadium
  • Quick Guides
    • Quick Start
    • JMeter Test
    • Locust Test
    • Script Builder
      • Simple HTTP Builder
      • Record&Play
    • Gatling Test
    • WebDriver Test
    • HLS Test
    • Understanding Test Reports
    • Test Execution Fail Reasons
    • File Output
  • ABOUT LOADIUM
    • Features
    • Pricing
    • VUH - Virtual User Hour
  • ACCOUNT
    • Profile & Account
    • Subscription
  • Features
    • Test Report
    • Compare Test
    • Private Location
    • Report PDF Export
  • Test Settings
    • Split CSV
    • Sandbox
    • Multi-Step
    • Geolocation
    • Customize Setup
    • Dedicated IP
    • Send Email
    • DNS Override
    • TPS
    • Network Type
    • Test Scheduling
    • Test Failure Settings
    • JMeter Settings
    • Failure Criteria
    • Flexible Thread Scheduling
  • CONTINUOUS INTEGRATION SUPPORT
    • Azure DevOps Pipeline
    • Jenkins Load Testing with Loadium
  • Integrations
    • Microsoft Teams Integration
    • New Relic Integration
    • AppDynamics Integration
    • TeamCity Integration
  • Jmeter Corner
    • Built-in JMeter Functions
    • How to Install JMeter Plugins
    • Record and Play Load Testing With Jmeter in 5 Steps
    • Websocket with Apache JMeter
    • JMeter Timers
    • Handling Error Messages of JMeter Tests in Loadium
    • Details of Throughput Controller in JMeter (Splitting Threads to Different Requests)
    • How to Add a Varying Number of Parameters in Apache JMeter
    • Local Network Simulation On JMeter
    • Running Load Test with FTP Server
  • Gatling Corner
    • Introduction to Gatling
    • Gatling Recorder
    • Gatling Pros&Cons
  • PUBLIC APIs
    • Get Test Status Service
    • Authentication Service
    • Get Performance Test List Service
    • Start Performance Test Service
Powered by GitBook
On this page
  • What is Locust?
  • How does Locust work in Loadium?
  • Creating a test in Loadium
  • Basic Example
  • CSV Usage Example
  • Logging Example
  • Logs in Loadium Report
  • Running the test

Was this helpful?

  1. Quick Guides

Locust Test

PreviousJMeter TestNextScript Builder

Last updated 3 months ago

Was this helpful?

What is Locust?

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.

For more information, you can visit and repository.

How does Locust work in Loadium?

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.

Loadium's current Python version is 3.12.x.

Creating a test in Loadium

To create a test,

  • Go to page

  • Choose "Locust" test type

  • Upload your Locust project files as a ZIP

  • 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.

Locust’s website
GitHub
Loadium New Test