Locust Cheatsheet: HttpUser, Tasks and Custom Shapes Reference

# Install
pip install "locust>=2.30"
# Optional: gevent-friendly faster HTTP (already bundled, but you may pin)
pip install "geventhttpclient>=2"

# Project layout
#   locustfile.py            # auto-discovered when running `locust`
#   tests/load_*.py          # alternative — pass with -f
#   shapes.py                # custom LoadTestShape (optional)

# Hello world (locustfile.py)
cat > locustfile.py <<'EOF'
from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)

    @task
    def index(self):
        self.client.get("/")
EOF

# Run — web UI on http://localhost:8089
locust -f locustfile.py --host https://stg.example.com

# Headless (CI mode)
locust -f locustfile.py --host https://stg.example.com \
       --users 200 --spawn-rate 50 --run-time 2m \
       --headless --print-stats --html report.html --csv results

# Distributed (separate terminals / hosts)
locust -f locustfile.py --master                  # control plane
locust -f locustfile.py --worker --master-host=10.0.0.1
# Or scale workers with --processes (single-host multi-core)
locust -f locustfile.py --processes 4 --headless --users 1000 --spawn-rate 100 --run-time 5m

from locust import HttpUser, FastHttpUser, SequentialTaskSet, TaskSet, task, between, tag

class Browser(HttpUser):
    """Slow, realistic — full requests session, cookies, redirects."""
    wait_time = between(1, 3)
    weight = 3                                  # 3:1 vs Buyer

    @task(5)                                    # weighted within this user
    def browse(self):
        self.client.get("/items", name="GET /items")  # group ad-hoc URLs

    @task
    @tag("smoke", "search")
    def search(self):
        self.client.get("/search?q=widget", name="GET /search")

    def on_start(self):                         # login once per VU
        self.client.post("/login", json={"u": "x", "p": "y"})


class Buyer(FastHttpUser):
    """3-10x faster, no requests session — use when scaling matters."""
    wait_time = between(0.5, 2)
    weight = 1

    @task
    class Checkout(SequentialTaskSet):          # tasks run in order, top-to-bottom
        @task
        def add_to_cart(self):
            self.client.post("/cart", json={"sku": "abc"})

        @task
        def pay(self):
            with self.client.post("/checkout", catch_response=True, name="POST /checkout") as r:
                if r.status_code != 201:
                    r.failure(f"unexpected {r.status_code}")
                elif r.elapsed.total_seconds() > 1:
                    r.failure("too slow")

        @task
        def stop(self):
            self.interrupt()                    # return to outer pool

# shapes.py — define load over time programmatically
from locust import LoadTestShape

class StagesShape(LoadTestShape):
    """Linear ramp-up, hold, spike, ramp-down."""
    stages = [
        # (run_time_seconds, target_users, spawn_rate)
        ( 30,  10, 5),    # warm-up
        ( 90,  50, 10),   # baseline
        (120, 200, 50),   # peak
        (150,  50, 20),   # cool-down
    ]

    def tick(self):
        rt = self.get_run_time()
        for end, users, rate in self.stages:
            if rt < end:
                return users, rate
        return None                              # stop the test


# Use it: locust -f locustfile.py shapes.py --headless
# (Locust picks up the LoadTestShape subclass automatically.)

# Other built-in shapes:
#   DoubleWaveShape, StepLoadShape (from locust-plugins)
#   or roll your own using __ENV / time.time() for dynamic patterns.

# locustfile.py — login, browse, checkout with events & SLO gating
import os, json, logging
from locust import FastHttpUser, task, between, events
from locust.runners import MasterRunner

BASE = os.getenv("API_URL", "https://stg.example.com")

with open("users.json") as f:                   # loaded once per worker process
    USERS = json.load(f)

class Shopper(FastHttpUser):
    host = BASE
    wait_time = between(0.5, 2)

    def on_start(self):
        u = USERS[self.environment.runner.user_count % len(USERS)]
        r = self.client.post("/api/login", json=u, name="POST /api/login")
        self.token = r.json()["token"]
        self.client.headers["Authorization"] = f"Bearer {self.token}"

    @task(4)
    def browse(self):
        self.client.get("/api/items", name="GET /api/items")

    @task(1)
    def checkout(self):
        with self.client.post("/api/orders", json={}, catch_response=True,
                              name="POST /api/orders") as r:
            if r.status_code != 201:
                r.failure(f"status={r.status_code}")

# Fail the run when SLOs miss
@events.quitting.add_listener
def _check_slo(environment, **kw):
    s = environment.stats.total
    if s.fail_ratio > 0.01:
        environment.process_exit_code = 1
        logging.error(f"fail ratio {s.fail_ratio:.2%} > 1%")
    if s.get_response_time_percentile(0.95) > 500:
        environment.process_exit_code = 1
        logging.error(f"p95 {s.get_response_time_percentile(0.95)}ms > 500ms")

@events.test_start.add_listener
def _on_start(environment, **kw):
    if isinstance(environment.runner, MasterRunner):
        logging.info(f"Starting against {BASE}")