kite/backend/load_tests/locustfile.py

"""Locust load test for DocuVault — D-04, D-05, D-06.

Strategy: pre-authenticated users (Option B) — a test_start listener creates
and authenticates all 50 virtual users before spawning begins.  This avoids
the IP-based auth rate limiter (10 req/min) hitting during the ramp-up phase,
while still exercising per-user isolated document endpoints under real load.

Run (headless, matches Phase 6 SLA gate):
    locust --headless --users 50 --spawn-rate 10 --run-time 12m \\
           --host http://localhost:8000 \\
           --csv backend/load_tests/results \\
           -f backend/load_tests/locustfile.py

    NOTE: --run-time must be 12m+ because Locust counts from process start.
    Setup takes ~6 min (6 batches × 65 s to respect 10 req/min auth rate limit).
    On re-runs users already exist (409 on register) — still need login pacing.

Prerequisites:
    pip install -r backend/requirements-dev.txt
    docker compose up  # backend, postgres, minio must be running

Exit codes (enforced by check_sla listener):
    0 — all SLA thresholds met
    1 — p95 > 200 ms, OR p99 > 500 ms, OR fail_ratio > 1%

Cleanup load-test users:
    docker compose exec postgres psql -U postgres -d docuvault -c \\
        "DELETE FROM users WHERE email LIKE 'loadtest%@example.com';"
"""
import os
import threading
import time
from io import BytesIO

import requests as _req
from locust import HttpUser, between, events, task

_BASE_URL = os.environ.get("LOAD_TEST_HOST", "http://localhost:8000")
_TEST_PASSWORD = os.environ.get("LOAD_TEST_PASSWORD", "Loadtest123!@#")
_NUM_USERS = int(os.environ.get("LOAD_TEST_USERS", "50"))

# Populated by the test_start listener; each slot is an access_token string.
_TOKEN_POOL: list[str] = []
_TOKEN_LOCK = threading.Lock()
_TOKEN_IDX = 0

_FAKE_PDF = b"%PDF-1.4\n1 0 obj<</Type/Catalog>>endobj\nxref\n0 2\ntrailer<</Size 2>>\n%%EOF"


@events.test_start.add_listener
def pre_authenticate(environment, **kwargs) -> None:
    """Create and authenticate all virtual users before spawning begins.

    Rate limit on /login is 10 req/min per IP (fixed-window).  We use a
    batch of 9 logins then a 65-second pause so each batch fits safely inside
    one window before the counter resets.  Setup runs before the test timer,
    so this does not count against the 5-minute SLA window.

    50 users ÷ 9 per batch = 6 batches × 65 s ≈ 6 minutes setup.
    """
    print(f"[setup] Pre-creating {_NUM_USERS} load-test accounts …")
    session = _req.Session()
    batch_size = 9  # /register and /login each have 10 req/min; 9 per window is safe

    for batch_start in range(0, _NUM_USERS, batch_size):
        batch_end = min(batch_start + batch_size, _NUM_USERS)
        batch_num = batch_start // batch_size + 1
        total_batches = (_NUM_USERS + batch_size - 1) // batch_size

        for i in range(batch_start, batch_end):
            email = f"loadtest{i}@example.com"
            handle = f"loadtest{i}"

            # Register — 409 means already exists, both are fine
            session.post(
                f"{_BASE_URL}/api/auth/register",
                json={"handle": handle, "email": email, "password": _TEST_PASSWORD},
                timeout=10,
            )

            # Login
            r = session.post(
                f"{_BASE_URL}/api/auth/login",
                json={"email": email, "password": _TEST_PASSWORD},
                timeout=10,
            )
            if r.status_code == 200:
                _TOKEN_POOL.append(r.json().get("access_token", ""))
            else:
                print(f"[setup] WARNING: login failed for loadtest{i} (status={r.status_code})")
                _TOKEN_POOL.append("")  # placeholder keeps slot indices aligned

        if batch_end < _NUM_USERS:
            print(
                f"[setup] Batch {batch_num}/{total_batches} done ({batch_end}/{_NUM_USERS}). "
                f"Waiting 65 s for rate-limit window reset …"
            )
            time.sleep(65)

    ready = sum(1 for t in _TOKEN_POOL if t)
    print(f"[setup] {ready}/{_NUM_USERS} accounts authenticated and ready.")


class DocuVaultUser(HttpUser):
    """Simulated DocuVault end-user exercising all authenticated endpoints.

    Task weights mirror a realistic read-heavy session per D-05:
      5 — list documents  (most common action)
      3 — get single document
      2 — upload a document
      1 — refresh access token
    """

    wait_time = between(0.5, 2.0)
    access_token: str = ""

    def on_start(self) -> None:
        global _TOKEN_IDX
        with _TOKEN_LOCK:
            idx = _TOKEN_IDX
            _TOKEN_IDX += 1
        if idx < len(_TOKEN_POOL):
            self.access_token = _TOKEN_POOL[idx]
        else:
            print(f"[worker] No token available for slot {idx} — stopping runner")
            self.environment.runner.quit()

    def _auth_headers(self) -> dict:
        return {"Authorization": f"Bearer {self.access_token}"}

    @task(5)
    def list_documents(self) -> None:
        self.client.get("/api/documents/", headers=self._auth_headers(), name="GET /api/documents/")

    @task(3)
    def get_document(self) -> None:
        resp = self.client.get(
            "/api/documents/",
            headers=self._auth_headers(),
            name="GET /api/documents/ (for get_document)",
        )
        if resp.status_code == 200:
            docs = resp.json().get("items", [])
            if docs:
                doc_id = docs[0]["id"]
                self.client.get(
                    f"/api/documents/{doc_id}",
                    headers=self._auth_headers(),
                    name="GET /api/documents/{id}",
                )

    @task(2)
    def upload_document(self) -> None:
        self.client.post(
            "/api/documents/upload",
            headers=self._auth_headers(),
            files={"file": ("load_test.pdf", BytesIO(_FAKE_PDF), "application/pdf")},
            data={"target_backend": "minio"},
            name="POST /api/documents/upload",
        )

    @task(1)
    def refresh_token(self) -> None:
        self.client.post(
            "/api/auth/refresh",
            headers=self._auth_headers(),
            name="POST /api/auth/refresh",
        )


@events.quitting.add_listener
def check_sla(environment, **kwargs) -> None:
    """Gate the Locust exit code on D-06 SLA thresholds."""
    stats = environment.runner.stats.total
    p95 = stats.get_response_time_percentile(0.95)
    p99 = stats.get_response_time_percentile(0.99)
    fail_ratio = stats.fail_ratio

    total_reqs = stats.num_requests
    if total_reqs < 500:
        # Setup consumed the entire run-time window; re-run with --run-time 12m
        print(f"SLA FAIL: only {total_reqs} requests — setup consumed run-time. Use --run-time 12m")
        environment.process_exit_code = 1
    elif fail_ratio > 0.01:
        print(f"SLA FAIL: fail_ratio={fail_ratio:.2%} > 1%")
        environment.process_exit_code = 1
    elif p95 > 200:
        print(f"SLA FAIL: p95={p95:.0f}ms > 200ms")
        environment.process_exit_code = 1
    elif p99 > 500:
        print(f"SLA FAIL: p99={p99:.0f}ms > 500ms")
        environment.process_exit_code = 1
    else:
        print(f"SLA PASS: p95={p95:.0f}ms p99={p99:.0f}ms fail_ratio={fail_ratio:.2%} reqs={total_reqs}")