第 6 章 · 断言与失败检测

本章目标：掌握 Locust 中使用 catch_response=True 进行请求级断言、resp.failure() 手动标记失败、通过 name= 参数聚合统计，以及构建可靠的成功条件判断逻辑。

6.1 为什么需要断言

默认情况下 Locust 只根据 HTTP 状态码判断请求成败：2xx 为成功，4xx/5xx 为失败。但生产环境中"返回 200 却是错误数据"的情况非常常见——比如接口返回 {"code": 500, "msg": "内部错误", "data": null} 但 HTTP 状态码仍是 200。

没有断言的压测就像没有温度计的体温检测：你看到请求都"成功"了，但实际上系统可能已经出了问题。

from locust import HttpUser, task, between

class ApiUser(HttpUser):
    wait_time = between(1, 3)
    host = "https://api.example.com"

    @task
    def get_user(self):
        # 默认行为：只检查 HTTP 状态码
        self.client.get("/api/users/1")
        # 状态码 200 → 标记成功（即使 body 是错误信息）

6.2 catch_response=True

在 self.client.get() / post() 等方法中传入 catch_response=True，Locust 会暂停自动标记，让你手动决定这次请求算成功还是失败：

from locust import HttpUser, task, between
import json

class ApiUser(HttpUser):
    wait_time = between(1, 2)
    host = "https://api.example.com"

    @task
    def get_user_with_assertion(self):
        with self.client.get("/api/users/1", catch_response=True) as resp:
            # 断言 1：HTTP 状态码必须是 200
            if resp.status_code != 200:
                resp.failure(f"状态码异常: {resp.status_code}")
                return

            # 断言 2：响应体必须是合法 JSON
            try:
                data = resp.json()
            except json.JSONDecodeError:
                resp.failure("响应不是有效 JSON")
                return

            # 断言 3：业务 code 必须为 0
            if data.get("code") != 0:
                resp.failure(f"业务错误: code={data.get('code')}, msg={data.get('msg')}")

            # 断言 4：关键字段存在
            if "data" not in data or not data["data"]:
                resp.failure("data 字段缺失或为空")

with ... as resp 是 Python 的上下文管理器语法。进入 with 块时 Locust 暂停统计；退出块时根据你是否调用了 resp.failure() 来标记成功/失败。

success() 方法

你也可以显式调用 resp.success() 标记成功，但这不是必须的——只要没调用 failure()，退出 with 块后 Locust 会自动标记为成功。

6.3 自定义统计名（name= 参数）

当 URL 包含动态路径参数时（如 /api/users/12345），每个不同 ID 都会生成一条独立的统计记录，导致统计表爆炸。使用 name= 参数可以将它们聚合：

import random
from locust import HttpUser, task, between

class SearchUser(HttpUser):
    wait_time = between(1, 3)
    host = "https://api.example.com"

    @task
    def search_products(self):
        product_id = random.randint(1, 10000)

        # 不用 name= → 统计表中出现 10000 条不同 URL 的记录
        # 用 name= → 所有请求聚合到 "/api/products/[id]" 一条记录
        with self.client.get(
            f"/api/products/{product_id}",
            name="/api/products/[id]",
            catch_response=True
        ) as resp:
            if resp.status_code == 200:
                data = resp.json()
                if data.get("price") is None:
                    resp.failure("价格字段缺失")
            else:
                resp.failure(f"HTTP {resp.status_code}")

    @task
    def get_order(self):
        order_id = f"ORD-{random.randint(1000, 9999)}"
        self.client.get(
            f"/api/orders/{order_id}",
            name="/api/orders/[order_id]"
        )

聚合后的统计表清晰易读：

Name                    # Requests   Fails   Median   95%ile
----------------------------------------------------------------
GET /api/products/[id]       5000     12      45ms    120ms
GET /api/orders/[order_id]   3000      5      62ms    180ms

6.4 高级断言模式

响应时间断言

除了验证内容，还可以把"响应太慢"视为失败：

import time
from locust import HttpUser, task, between

class LatencySensitiveUser(HttpUser):
    host = "https://api.example.com"
    wait_time = between(1, 2)

    @task
    def latency_check(self):
        start = time.monotonic()
        with self.client.get("/api/critical", catch_response=True) as resp:
            elapsed_ms = (time.monotonic() - start) * 1000

            if resp.status_code != 200:
                resp.failure(f"HTTP {resp.status_code}")
            elif elapsed_ms > 500:
                resp.failure(f"响应过慢: {elapsed_ms:.0f}ms > 500ms")
            else:
                pass  # 自动标记成功

条件性失败策略

某些场景下你可能希望只统计特定比例的错误，或者对非关键接口宽容一些：

class SmartUser(HttpUser):
    host = "https://staging.example.com"
    wait_time = between(1, 3)

    @task
    def flexible_check(self):
        with self.client.get("/api/data", catch_response=True) as resp:
            if resp.status_code >= 500:
                resp.failure(f"Server Error: {resp.status_code}")
            elif resp.status_code == 429:
                resp.failure("Rate limited")
            elif resp.status_code == 404:
                pass  # 测试环境的 404 可能只是测试数据问题，静默处理
            else:
                resp.success()

6.5 全局成功条件

有时你需要在整个压测结束后判断整体是否通过。可以结合事件钩子实现：

from locust import events
import logging

# 收集全局错误率
stats = {"total": 0, "failures": 0}

@events.request.add_listener
def on_request(request_type, name, response_time, response_length, exception, **kwargs):
    stats["total"] += 1
    if exception:
        stats["failures"] += 1

@events.quitting.add_listener
def on_quitting(environment, **kwargs):
    """压测结束时检查全局成功率"""
    if stats["total"] > 0:
        error_rate = stats["failures"] / stats["total"]
        if error_rate > 0.05:
            logging.error(f"压测失败！全局错误率 {error_rate:.1%} 超过阈值 5%")
            environment.process_exit_code = 1
        else:
            logging.info(f"压测通过，错误率 {error_rate:.1%}")

设置 environment.process_exit_code = 1 后，CI/CD 流水线可以通过退出码判断压测是否达标。

本章小结

• 默认按 HTTP 状态码判定成败；catch_response=True 开启手动断言模式；
• resp.failure(msg) 标记失败，resp.success() 显式标记成功；
• name= 聚合动态路径参数，避免统计表爆炸；
• 响应时间也可以作为断言条件；
• 结合 events.quitting 钩子设置进程退出码，让 CI/CD 自动判断压测是否通过。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. Locust 中开启手动断言的参数是什么？

2. 当 URL 包含动态路径参数（如 /users/123）时，应该用什么参数聚合统计？

3. resp.failure("timeout") 调用后，这条请求会被如何处理？

4. 要让 CI/CD 在压测未达标时自动失败，最关键的做法是？

🛠️ 动手实践

1. 编写一个带 catch_response=True 的任务，同时校验 HTTP 状态码、JSON 格式和业务 code 字段。
2. 使用 name= 参数将 /api/items/1 到 /api/items/100 的请求聚合为一条统计记录。
3. 实现一个全局错误率检查器：当总错误率超过 3% 时设置退出码为 1。