Locust deep dive

pip install locust

pip install -r requirements-load.txt   # contains 'locust>=2.43'

from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)   # 1-3 seconds between tasks per user

    def on_start(self):
        # Runs once per virtual user when they start
        self.client.post("/login", json={"user": "test", "pass": "secret"})

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

    @task(3)   # weighted - runs 3x as often as unweighted tasks
    def view_item(self):
        self.client.get("/items/42")

@task
def view_item(self):
    item_id = random.randint(1, 1000)
    # name= rolls /items/42, /items/43, ... up under one "/items/[id]" row
    self.client.get(f"/items/{item_id}", name="/items/[id]")

locust -f locustfile.py

locust --headless --users 10 --spawn-rate 1 -H http://your-server.com

locust -f locustfile.py \
  --headless \
  --users 100 \
  --spawn-rate 10 \
  --run-time 5m \
  --host https://staging.example.com \
  --csv results \
  --html report.html \
  --exit-code-on-error 1

# On the master node
locust -f locustfile.py --master --headless --users 1000 --spawn-rate 50 --host https://staging.example.com

# On each worker node
locust -f locustfile.py --worker --master-host master.internal

# .github/workflows/locust.yml
name: load-test

on:
  pull_request:
    paths: ['tests/load/**']
  schedule:
    - cron: '0 4 * * *'

jobs:
  locust:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v5

      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'

      - run: pip install locust

      - name: Run Locust headless
        env:
          API_TOKEN: ${{ secrets.STAGING_API_TOKEN }}
        run: |
          locust -f tests/load/locustfile.py \
            --headless \
            --users 50 \
            --spawn-rate 5 \
            --run-time 3m \
            --host https://staging.example.com \
            --csv results \
            --html report.html \
            --exit-code-on-error 1

      - name: Custom threshold gate
        run: |
          # Fail if p95 > 500ms on any endpoint
          python - <<'PY'
          import csv, sys
          with open('results_stats.csv') as f:
              for row in csv.DictReader(f):
                  if row['Name'] == 'Aggregated':
                      continue
                  p95 = float(row['95%'])
                  if p95 > 500:
                      print(f"::error::p95 {p95}ms on {row['Name']} (>500 budget)")
                      sys.exit(1)
          PY

      - name: Upload reports
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: locust-reports
          path: |
            results_stats.csv
            results_stats_history.csv
            results_failures.csv
            report.html
          retention-days: 14