locust-best-practices

Locust 场景构建器

为 Python 中的 Locust 负载测试脚本规范一致、可用于生产环境的模式，涵盖 HTTP/REST API、逼真的用户行为、自定义负载形状和恰当的响应验证。

输出格式

生成或修复 locustfile 时，始终提供以下三项：

1. 完整、可运行的 locustfile——绝不只提供部分代码片段。
2. 准确的运行命令，包含所有必需的标志（--headless、-u、-r、-t、--host）。
3. 一行说明所选 wait_time 策略及其为何适合该负载目标。

---

步骤 1 — 收集上下文

只询问尚不明确的信息：

• 目标： 新建 locustfile · 修复现有 locustfile · 特定 API 问题
• 协议： HTTP/REST（默认）· WebSocket · gRPC
• 工作负载： 固定并发用户数还是固定到达速率（RPS）？
• 身份验证： 无 · 启动时登录 · 从环境变量获取令牌？
• 数据： 静态 · CSV 文件 · 为每位用户生成？

仅当用户询问自定义负载配置、多阶段递增、尖峰模拟或 LoadTestShape 时，才加载 references/SHAPES.md。

**仅当用户提到 WebSocket、gRPC 或测试非 HTTP 协议时，才加载 references/PROTOCOLS.md。**其中包含自定义客户端包装器、events.request.fire() 模式和协议兼容性说明。

仅当用户询问项目结构、多个用户类、TaskSet、共享身份验证辅助函数、大规模 CSV 数据加载、模块化 locustfile 或多文件组织时，才加载 references/DESIGN-PATTERNS.md。

仅当用户询问无头模式、CI/CD 集成、退出代码、基于 SLA 的构建失败、分布式测试、主从配置、多核、记录日志到文件或自定义日志级别时，才加载 references/RUNNING.md。

---

步骤 2 — 应用 4 块模式

每个 locustfile 都必须遵循此结构。先生成完整框架，再填入详细内容。

Block 1 → Imports + config    locust imports, constants, env vars
Block 2 → User class          HttpUser with host, wait_time, on_start, @task methods
Block 3 → Shape (optional)    LoadTestShape for custom ramp profiles
Block 4 → Run command         locust CLI with all flags as a comment at the bottom

---

步骤 3 — HttpUser 规则

3.1 验证时始终使用 catch_response=True

不要只依赖 HTTP 状态码。始终验证响应正文：

# Wrong — only checks HTTP status, misses business logic errors
self.client.get("/api/users")

# Correct — validates response content
with self.client.get("/api/users", catch_response=True) as response:
    if response.status_code != 200:
        response.failure(f"Expected 200, got {response.status_code}")
    elif "users" not in response.json():
        response.failure("Missing 'users' key in response")

3.2 使用 name= 对动态 URL 分组

不进行分组时，每个唯一 URL 都会生成单独的统计条目，导致报告内容泛滥：

# Wrong — creates N separate stats rows
for user_id in range(100):
    self.client.get(f"/user/{user_id}")

# Correct — groups all under one stats entry
for user_id in range(100):
    self.client.get(f"/user/{user_id}", name="/user/[id]")

3.3 使用 on_start 进行身份验证——绝不在用户之间共享令牌

每位虚拟用户都必须独立进行身份验证：

# Wrong — shared token is a race condition and not realistic
shared_token = None

class MyUser(HttpUser):
    def on_start(self):
        global shared_token  # ❌ shared state across users

# Correct — each user gets its own token
class MyUser(HttpUser):
    def on_start(self):
        response = self.client.post("/auth/login", json={
            "username": os.environ["TEST_USER"],
            "password": os.environ["TEST_PASSWORD"],
        })
        self.token = response.json()["access_token"]

    @task
    def get_profile(self):
        self.client.get("/profile", headers={"Authorization": f"Bearer {self.token}"})

3.4 根据负载目标选择 wait_time

目标	wait_time	原因
模拟带思考时间的真实用户	between(1, 5)	节奏逼真
无论延迟如何都维持固定 RPS	constant_throughput(2)	开放模型——每位用户每秒执行 2 个任务
重现精确的用户节奏	constant(1)	固定间隔

绝不要使用 wait_time = constant(0)——零思考时间会产生比真实用户高 10–100 倍的负载，结果也不真实。

3.5 使用 @task 权重反映真实流量分布

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

    @task(10)   # 10× more common — most users just browse
    def browse_products(self):
        self.client.get("/products")

    @task(3)    # 3× — some users search
    def search(self):
        self.client.get("/search?q=shoes", name="/search")

    @task(1)    # 1× — few users actually purchase
    def checkout(self):
        self.client.post("/checkout", json={"cart_id": self.cart_id})

---

步骤 4 — 标准 Locustfile 模板

import os
import json
from locust import HttpUser, task, between, events

# ---------------------------------------------------------------------------
# Block 1 — Config
# ---------------------------------------------------------------------------
BASE_URL = os.environ.get("TARGET_HOST", "http://localhost:8080")

# ---------------------------------------------------------------------------
# Block 2 — User class
# ---------------------------------------------------------------------------
class ApiUser(HttpUser):
    host = BASE_URL
    wait_time = between(1, 3)

    def on_start(self):
        """Runs once per virtual user before any task executes."""
        with self.client.post(
            "/auth/login",
            json={
                "username": os.environ["TEST_USER"],
                "password": os.environ["TEST_PASSWORD"],
            },
            catch_response=True,
        ) as response:
            if response.status_code == 200:
                self.token = response.json()["access_token"]
            else:
                response.failure(f"Login failed: {response.status_code}")
                self.token = None

    def _auth_headers(self):
        return {"Authorization": f"Bearer {self.token}"}

    @task(3)
    def get_items(self):
        with self.client.get(
            "/api/items",
            headers=self._auth_headers(),
            catch_response=True,
        ) as response:
            if response.status_code != 200:
                response.failure(f"GET /api/items failed: {response.status_code}")

    @task(1)
    def create_item(self):
        with self.client.post(
            "/api/items",
            json={"name": "test-item", "value": 42},
            headers=self._auth_headers(),
            catch_response=True,
        ) as response:
            if response.status_code not in (200, 201):
                response.failure(f"POST /api/items failed: {response.status_code}")

# ---------------------------------------------------------------------------
# Block 3 — Events (optional)
# ---------------------------------------------------------------------------
@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    print("Test started — verifying environment variables...")
    for var in ["TEST_USER", "TEST_PASSWORD"]:
        if not os.environ.get(var):
            raise ValueError(f"Missing required env var: {var}")

# ---------------------------------------------------------------------------
# Run command (headless):
# TEST_USER=alice TEST_PASSWORD=secret locust -f locustfile.py \
#   --headless -u 50 -r 5 -t 5m --host http://your-api.com \
#   --html report.html --csv results
# ---------------------------------------------------------------------------

---

步骤 5 — 常见错误

1. 未使用 catch_response=True——静默失败

若不使用它，即使 HTTP 连接有效，500 响应在 Locust 统计中仍会计为成功。始终使用 catch_response=True 并显式调用 response.failure()。

2. 动态 URL 缺少 name=

每个唯一 URL 路径都会在统计表中新增一行。像 /user/123、/user/456 这样的参数化路径会让报告内容暴增。请使用 name="/user/[id]"。

3. 用户之间共享可变状态

Locust 用户在 greenlet（gevent）中运行，而不是在线程中运行，但共享状态仍会导致逻辑中的竞态条件。始终将每位用户的状态（self.token、self.cart_id）存储为实例属性。

从 CSV 加载凭据时，在 on_start 中将所有字段存储为实例属性，而不仅仅是其中一个：

# Wrong — only stores username, password is lost
self.username = row["username"]

# Correct — store every field the user will need
self.username = row["username"]
self.password = row["password"]

4. wait_time = constant(0) 或不设置 wait_time

省略思考时间会产生不现实的负载，并导致测试机器先于目标系统达到饱和。务必设置有意义的 wait_time。

5. 将主机和凭据硬编码

切勿将 host、用户名或密码硬编码。务必使用环境变量：

host = os.environ.get("TARGET_HOST", "http://localhost:8080")

6. 只断言状态码，不检查响应正文

返回 {"error": "token expired"} 的 200 响应也会通过仅检查状态码的验证。务必验证响应正文，确保业务逻辑正确。

7. 未了解取舍便使用 FastHttpUser

FastHttpUser 使用 geventhttpclient 而非 requests。它速度更快，但不支持 requests 的所有功能（Cookie jar、某些身份验证辅助工具）。只有在确认瓶颈位于测试运行器本身、而非目标系统时，才使用它。

---

第 6 步 — 运行命令参考

# Basic headless run — 50 users, 5/sec ramp, 5 minutes
locust -f locustfile.py --headless -u 50 -r 5 -t 5m --host https://api.example.com

# With HTML report and CSV output
locust -f locustfile.py --headless -u 100 -r 10 -t 10m \
  --host https://api.example.com \
  --html report.html --csv results

# Run specific user class only
locust -f locustfile.py --headless -u 50 -r 5 -t 5m \
  --host https://api.example.com ApiUser

# With environment variables
TARGET_HOST=https://api.example.com \
TEST_USER=alice \
TEST_PASSWORD=secret \
locust -f locustfile.py --headless -u 50 -r 5 -t 5m

# Open web UI (no --headless) — useful for interactive exploration
locust -f locustfile.py --host https://api.example.com

标志参考：

标志	含义
-u	并发用户峰值数量
-r	用户启动速率 — 每秒新增的用户数
-t	总运行时间（例如 30s、5m、2h）
--headless	不显示 Web UI — 运行后自动退出
--host	目标基础 URL
--html	在指定路径生成 HTML 报告
--csv	在指定前缀生成 CSV 统计文件
--only-summary	不输出每个时间间隔的统计信息