结论先行:限流/熔断是系统容量行为,不是用例行为。你的核心问题是把「系统正在限流/熔断」当成了「用例断言失败」。解法是把两者解耦——额度按用例隔离、熔断态感知并标记为 xfail、限流断言独立成专项用例。下面 4 步,每步带代码和判据。
第 1 步:额度隔离——让每条用例有独立的限流 key
Sentinel 网关层的限流 key 通常是 resource name(路径 + method)或 param 维度。3 人共享 staging,同一个 resource 的 token 桶是共享的,你跑 500 条、他跑 500 条,峰值 QPS 轻松打穿阈值。
方案 A(推荐,改网关配置):让网关支持按 X-Tenant-Id 或 X-Test-Id header 做限流 key 的维度拆分。每条用例带独立 header,Sentinel 按这个维度分桶:
# conftest.py
import pytest, uuid, os
@pytest.fixture(autouse=True)
def isolated_tenant():
"""为每条用例分配独立租户,Sentinel 限流按 tenant 分桶"""
tenant_id = f"auto-{uuid.uuid4().hex[:8]}"
# 调 admin 接口创建临时租户(staging 专用)
yield tenant_id
# teardown: 清理临时租户
@pytest.fixture
def client(isolated_tenant):
return httpx.Client(
base_url=os.environ["STAGING_URL"],
headers={"X-Tenant-Id": isolated_tenant,
"X-Test-Id": os.environ.get("PYTEST_NODEID", "manual")},
timeout=10.0,
)
方案 B(不改网关):3 人各用独立 merchant_id(不同租户),Sentinel 限流规则天然按 tenant 分桶。CI 里写死各自的 tenant,互不干扰。
判据:改完后跑 pytest-xdist -n 4 并发回归,429 出现率应从「~5%」降到「<0.1%」。如果还高,去 Sentinel 规则里查 resource 定义,确认限流 key 真的包含了 tenant 维度。
第 2 步:熔断态感知——httpx 中间件区分「限流」和「业务错」
Sentinel 限流返回 429,熔断返回 503,通常 response header 里带 X-Sentinel-Blocked: true(或你们网关自定义的 header)。在 httpx transport 层加中间件:
import httpx
class SentinelAwareTransport(httpx.BaseTransport):
SENTINEL_HEADERS = ("x-sentinel-blocked", "x-rate-limit-blocked")
def __init__(self, inner: httpx.HTTPTransport):
self._inner = inner
def handle_request(self, request: httpx.Request) -> httpx.Response:
response = self._inner.handle_request(request)
if response.status_code in (429, 503):
blocked = any(h.lower() in self.SENTINEL_HEADERS
for h in response.headers.keys())
request.extensions["sentinel_blocked"] = blocked
request.extensions["block_type"] = (
"rate-limit" if response.status_code == 429
else "circuit-breaker"
)
return response
# conftest.py
@pytest.fixture
def client(isolated_tenant):
inner = httpx.HTTPTransport()
transport = SentinelAwareTransport(inner)
with httpx.Client(transport=transport,
base_url=os.environ["STAGING_URL"],
headers={"X-Tenant-Id": isolated_tenant}) as c:
yield c
断言层:
def test_order_create(client, tenant):
resp = client.post("/api/orders", json={...})
if resp.extensions.get("sentinel_blocked"):
pytest.xfail(f"环境限流/熔断: {resp.extensions['block_type']}")
assert resp.status_code == 201 # 真正的业务断言
效果:限流/熔断导致的失败变成 xfail(黄色),不再和真正的断言失败(红色)混在一起。跑完一轮 pytest 报告里直接看到「本轮 N 条被环境限流/熔断影响」,不用在 500 条里逐条翻。
第 3 步:熔断恢复——测试前后重置 Half-Open 状态
Sentinel 熔断是 per-resource 的。一条用例触发了熔断,Half-Open 窗口(默认 5s)内如果探测失败回到 Open,后面十几条全挂——这就是你说的「多米诺骨牌」。
a) 有 Sentinel Dashboard(默认 8080):测试前直接调 API 重置:
import requests, os
def reset_sentinel_circuit(resource_name: str):
try:
requests.post(
f"{os.environ['SENTINEL_DASH']}/api/circuit-breaker/reset",
json={"resource": resource_name}, timeout=5)
except Exception:
pass
@pytest.fixture(autouse=True)
def reset_breaker():
reset_sentinel_circuit("api:order:create")
reset_sentinel_circuit("api:export")
yield
b) 没有 Dashboard:测试启动时对该接口发 1 个轻量探测,如果 503 说明还在熔断态,等 Half-Open 窗口 × 2 后再跑:
import time
def wait_for_circuit_recovery(client, url, max_wait=30):
deadline = time.time() + max_wait
while time.time() < deadline:
resp = client.get(url)
if resp.status_code != 503:
return
time.sleep(3)
pytest.skip(f"熔断未恢复,跳过 {url}")
c) 根因:如果熔断频繁触发,大概率是并发度超过限流阈值。staging 阈值 50 QPS,3 人 × 500 条并发跑,峰值 150+ QPS。建议:
- CI 里写死
pytest-xdist -n 2(不是 -n 4),让峰值 QPS 在阈值内 - 或加
threading.Semaphore(10)限制同时在飞请求数 - 和后端确认 staging 阈值,写进 CI 配置文档
第 4 步:限流断言独立成专项用例
「验证限流生效」和「验证业务逻辑」是两件事。把限流验证抽成独立文件,标记 order_last,放在所有业务用例之后跑:
# test_rate_limit.py
@pytest.mark.slow
@pytest.mark.order_last # 放最后跑,避免污染前面的用例
def test_order_rate_limit_triggers_429(client):
"""快速发 100 个请求,期望前 N 个 201,后续 429"""
results = [client.post("/api/orders", json={...}).status_code
for _ in range(100)]
first_429 = next((i for i, c in enumerate(results) if c == 429), None)
assert first_429 is not None, "限流未触发,检查阈值配置"
assert all(c == 429 for c in results[first_429:]), "限流后应持续 429"
assert all(c == 201 for c in results[:first_429]), "限流前应全部成功"
这样它触发限流/熔断不会影响前面的业务用例,而且「限流本身」成为可被单独断言的合法场景。
可执行 checklist
| 步骤 | 动作 | 判据 | |------|------|------| | 1 | 每条用例独立 tenant / test-id | 并发回归 429 率 < 0.1% | | 2 | httpx 中间件识别 Sentinel 响应 → xfail | 限流失败不再混入断言失败 | | 3 | 测试前重置熔断态 / 等待恢复 | 多米诺骨牌效应消失 | | 4 | 限流断言独立成专项用例,order_last | 业务用例 0 条 429 | | 5 | 并发度控制在限流阈值内 | 整轮通过率 > 99% |
一句话:限流/熔断是系统容量行为,测试框架的职责是隔离它、感知它、独立验证它,而不是和它赛跑。把「额度按用例分桶」和「熔断态标记为 xfail」这两件事做到位,500 条用例的稳定性会立刻上一个台阶。
补一个可观测性视角:额度隔离和 xfail 解决的是「不污染」,但题干最烦的「谁吃掉了额度」其实是归因问题。建议再加一个限流预算 ledger——httpx 中间件记录每个请求 (ts, resource, tenant, X-RateLimit-Remaining),429 时把该 resource 最近 20 条按 tenant 聚合打印,谁打空了桶一眼可见,比在 500 条里逐条翻快得多。
补一个零配置成本的一招,前面几位都没提:不改网关,直接用 pytest-xdist 的 --dist loadgroup + @pytest.mark.xdist_group("api:order"),把同一路径(同 resource)的用例强制落到同一 worker 串行、组间并行。Sentinel 限流本来就按 resource 分桶,只要保证「同 resource 的用例不同时发」就不会互相吃额度——429 出现率能从 ~5% 压到 <0.5%。建议作为第一刀先上,比改网关 header 分桶更快见效。
补充一个 SRE 视角的踩坑:Sentinel 熔断的 Half-Open 态自带「探测请求」机制——第一个请求会被放行作为探测,成功则关闭熔断。测试里可以主动利用这个:启动时先发 1 个轻量 GET,200 说明已关闭,503 说明还在 Open 态,等 statIntervalMs(默认 1s)× minRequestAmount(默认 5)后再重试,比「等 Half-Open 窗口 × 2」更精确,因为你是按 Sentinel 实际参数算的。另外,3 人共享 staging 跑 1500 条用例,峰值 QPS 打穿阈值是必然的,隔离额度只是缓解,根本解法还是给自动化测试一个独立环境。
补充一个网关侧的视角:方案 A 的"每条用例造独立 tenant"在 3 人 × 500 条的场景下会瞬间产生 1500 个临时租户,staging DB 和缓存的清理压力不小。我在 Envoy + Sentinel 网关上都验证过一个更轻的做法——加一个 X-Request-Source: staging-auto 的 header,网关侧按这个 header 把测试流量路由到独立的限流桶(阈值放大 50~100x),测试框架只改一个 header,0.5 天落地,429 出现率从 ~5% 降到 <0.1%。如果 3 人共用一个 CI runner(同出口 IP),这个方案比 tenant 隔离更干净,因为限流 key 里的 IP 维度被绕开了。具体代码片段和成本对比放在我的回答里了。
补充一个前端侧容易踩的坑:如果你的自动化里还有 Playwright E2E 用例走浏览器调同一批接口,Service Worker 的 cache-first 策略会把 429 吞掉返回缓存的 200,导致断言假绿。我们在一个 B 端项目里就遇到过——用户反馈「按钮点了 3 分钟没反应」,实际是 429 被浏览器缓存了 60 秒(网关在 429 响应里加了 Cache-Control: max-age=60),后续点击根本没发请求。E2E 里记得 context.serviceWorkers = 'deny' 或测试启动时 reg.delete(),另外检查 429 的 Response Headers 里有没有被网关塞了 Cache-Control。