Load tests development for website (Locust)

Load Testing Development for Websites (Locust)

Locust is a Python load testing tool. Scenarios are written as regular Python code, providing maximum flexibility. Built-in web interface for managing tests in real-time. Easily scales horizontally for high loads.

Installation

pip install locust

Basic Scenario

# locustfile.py
from locust import HttpUser, task, between, events
import json
import random

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)  # wait between requests 1-3 seconds

    def on_start(self):
        """Executed when each user starts"""
        self.login()

    def login(self):
        response = self.client.post("/api/auth/login", json={
            "email": f"user{random.randint(1, 1000)}@test.com",
            "password": "testpassword",
        })
        if response.status_code == 200:
            self.token = response.json()["access_token"]
            self.client.headers.update({"Authorization": f"Bearer {self.token}"})
        else:
            self.token = None

    @task(3)  # weight 3 — executed 3x more often
    def browse_products(self):
        page = random.randint(1, 10)
        with self.client.get(f"/api/products?page={page}", catch_response=True) as response:
            if response.status_code == 200:
                data = response.json()
                if "data" not in data:
                    response.failure("Missing 'data' key in response")
            else:
                response.failure(f"Got {response.status_code}")

    @task(2)
    def view_product(self):
        product_id = random.randint(1, 500)
        self.client.get(f"/api/products/{product_id}")

    @task(1)
    def create_order(self):
        if not self.token:
            return
        self.client.post("/api/orders", json={
            "product_id": random.randint(1, 100),
            "quantity": random.randint(1, 3),
        })

Custom Metrics

from locust import events
from locust.runners import MasterRunner

@events.request.add_listener
def on_request(request_type, name, response_time, response_length, response,
               context, exception, start_time, url, **kwargs):
    if exception:
        print(f"Request failed: {name} - {exception}")
    elif response_time > 2000:
        print(f"Slow request: {name} - {response_time}ms")


class FastApiUser(HttpUser):
    wait_time = between(0.5, 2)

    @task
    def search(self):
        start = time.time()
        with self.client.get("/api/search?q=laptop", catch_response=True) as response:
            duration = (time.time() - start) * 1000

            if response.status_code != 200:
                response.failure(f"Wrong status: {response.status_code}")
            elif duration > 1000:
                response.failure(f"Too slow: {duration:.0f}ms")

Multiple User Types

class BrowsingUser(HttpUser):
    weight = 7  # 70% of users
    wait_time = between(2, 5)

    @task
    def browse(self):
        self.client.get("/api/products")

class PowerUser(HttpUser):
    weight = 3  # 30% of users
    wait_time = between(0.5, 1)

    @task
    def create_content(self):
        self.client.post("/api/products", json={
            "name": f"Product {random.randint(1, 9999)}",
            "price": random.uniform(100, 10000),
        })

Running Tests

# With web interface
locust -f locustfile.py
# Open http://localhost:8089

# Headless mode
locust -f locustfile.py \
    --headless \
    --users 100 \
    --spawn-rate 10 \
    --run-time 5m \
    --host https://staging.example.com \
    --html report.html \
    --csv results

# Distributed mode (multiple machines)
# Master
locust -f locustfile.py --master --expect-workers=3

# Workers (on other servers)
locust -f locustfile.py --worker --master-host=192.168.1.100

Thresholds (Assertions)

# locustfile.py
from locust import events

@events.quitting.add_listener
def assert_stats(environment, **kwargs):
    stats = environment.runner.stats
    total = stats.total

    if total.fail_ratio > 0.01:
        print(f"FAIL: Error rate {total.fail_ratio:.2%} > 1%")
        environment.process_exit_code = 1

    if total.avg_response_time > 500:
        print(f"FAIL: Avg response time {total.avg_response_time:.0f}ms > 500ms")
        environment.process_exit_code = 1

    p99 = total.get_response_time_percentile(0.99)
    if p99 > 2000:
        print(f"FAIL: p99 {p99:.0f}ms > 2000ms")
        environment.process_exit_code = 1

CI/CD Integration

# GitHub Actions
- name: Run Locust Load Test
  run: |
    locust -f locustfile.py \
      --headless \
      --users 50 \
      --spawn-rate 5 \
      --run-time 3m \
      --host ${{ vars.STAGING_URL }} \
      --html load-report.html
  continue-on-error: false

- name: Upload Report
  uses: actions/upload-artifact@v3
  with:
    name: load-test-report
    path: load-report.html

Implementation Timeline

Writing 3–5 Locust scenarios with user types and assertions: 2–4 days.