2.2Locust (Python-Based)

Locust uses Python classes to define user behavior. It is the preferred choice for teams with strong Python skills, complex scenario logic, or when you need to share code with your existing Python test infrastructure.

Why Choose Locust

• Pure Python. Leverage existing libraries (requests, database clients, custom SDKs).
• Distributed by default. Built-in master/worker architecture for horizontal scaling.
• Real-time web UI. Monitor test progress and adjust parameters live.
• Event hooks. Deep customization of request lifecycle, error handling, and reporting.

Complete Locust Example

# locustfile.py -- Multi-persona e-commerce load test
from locust import HttpUser, task, between, tag, events
import random
import logging

logger = logging.getLogger(__name__)


class BrowsingUser(HttpUser):
    """Simulates casual browsing behavior -- 70% of traffic."""
    wait_time = between(1, 5)
    weight = 7  # 70% of simulated users

    @task(5)
    def view_homepage(self):
        with self.client.get("/", name="Homepage", catch_response=True) as response:
            if response.status_code != 200:
                response.failure(f"Homepage returned {response.status_code}")
            elif "Welcome" not in response.text:
                response.failure("Homepage missing welcome text")

    @task(3)
    def view_product(self):
        product_id = random.randint(1, 1000)
        self.client.get(f"/products/{product_id}", name="/products/[id]")

    @task(1)
    @tag("checkout")
    def add_to_cart(self):
        self.client.post("/cart", json={"product_id": 42, "quantity": 1})

    def on_start(self):
        """Runs once per simulated user at start."""
        logger.info("BrowsingUser session started")


class APIUser(HttpUser):
    """Simulates API consumer traffic -- 30% of traffic."""
    wait_time = between(0.1, 0.5)
    weight = 3  # 30% of simulated users

    @task
    def search(self):
        query = random.choice(["laptop", "phone", "headphones", "keyboard"])
        self.client.get(f"/api/search?q={query}", name="API Search")

    @task
    def get_inventory(self):
        product_id = random.randint(1, 100)
        self.client.get(f"/api/inventory/{product_id}", name="API Inventory")


# Custom event listener for reporting
@events.request.add_listener
def on_request(request_type, name, response_time, response_length, exception, **kwargs):
    if response_time > 5000:
        logger.warning(f"Slow request: {name} took {response_time}ms")

Running Locust

# Local single-process run
locust -f locustfile.py --host=https://staging.example.com

# Headless mode for CI (no web UI)
locust -f locustfile.py --host=https://staging.example.com \
  --headless -u 100 -r 10 --run-time 5m

# Distributed mode: start master
locust -f locustfile.py --master --host=https://staging.example.com

# Distributed mode: start workers (run on multiple machines)
locust -f locustfile.py --worker --master-host=192.168.1.100

# Filter by tag
locust -f locustfile.py --tags checkout --host=https://staging.example.com