接口 Mock 与测试数据构造¶
Mock 用错方向的典型症状:单测全绿,联调一跑就 500。原因通常不是 Mock 写少了,而是 Mock 写的是想象出来的响应,既没校验请求,也没和真实契约对齐。下面按“能跑起来”的顺序给出四层方案和一套数据构造/清理模板。
一、Mock 的四个层次¶
| 层次 | 手段 | 拦截位置 | 适合解决 | 失效方式 |
|---|---|---|---|---|
| 前端拦截 | MSW、axios adapter、mirage | 浏览器/Node 内 | 前端联调、页面异常态 | 只覆盖前端,后端契约完全没验证 |
| Mock Server | WireMock、MockServer、moco | 网络层,独立进程 | 下游未就绪、异常注入、录制回放 | stub 与真实实现漂移 |
| 契约 Mock | Pact、Spring Cloud Contract | 消费方测试 + 提供方验证 | 双方对字段/状态的约定 | 只写消费方测试,不跑 provider 验证 |
| 测试替身 | monkeypatch、Mockito、fakes | 进程内函数/对象 | 单元测试、难触发的分支 | 替身行为与真实实现不一致 |
原则只有一条:Mock 边界,不 Mock 被测逻辑。被测服务内部的金额计算用 Mock 掉,等于没测。
二、WireMock:独立进程 + stub JSON¶
2.1 启动¶
# 方式一:standalone jar(Java 17+ 对应 WireMock 3.x)
java -jar wiremock-standalone-3.9.1.jar \
--port 9999 --root-dir ./wiremock --verbose
# 方式二:容器
docker run --rm -p 9999:8080 \
-v "$PWD/wiremock:/home/wiremock" \
wiremock/wiremock:3.9.1 --verbose
目录约定:wiremock/mappings/*.json 启动时自动加载,wiremock/__files/ 放响应体文件。--verbose 会把未匹配请求的 diff 打到控制台,排错必开。
2.2 一个带请求校验的 stub¶
wiremock/mappings/pay-query.json:
{
"priority": 1,
"scenarioName": "pay-query",
"requiredScenarioState": "Started",
"newScenarioState": "paid",
"request": {
"method": "POST",
"urlPath": "/pay/query",
"headers": {
"Content-Type": { "contains": "application/json" },
"Authorization": { "matches": "Bearer .+" }
},
"bodyPatterns": [
{ "matchesJsonPath": "$.orderNo" },
{ "equalToJson": { "orderNo": "P1001", "channel": "ALIPAY" },
"ignoreExtraElements": true }
]
},
"response": {
"status": 200,
"headers": { "Content-Type": "application/json;charset=UTF-8" },
"jsonBody": { "code": "SUCCESS", "data": { "paid": true, "paidAt": "2026-10-22T11:20:00+08:00" } },
"fixedDelayMilliseconds": 300
}
}
加载与验证:
curl -sS -X POST http://localhost:9999/__admin/mappings \
-H 'Content-Type: application/json' \
-d @wiremock/mappings/pay-query.json
# {"id":"...","request":{...},"response":{...},"priority":1,"scenarioName":"pay-query",...}
curl -sS -X POST http://localhost:9999/pay/query \
-H 'Content-Type: application/json' \
-H 'Authorization: Bearer test-token' \
--data-raw '{"orderNo":"P1001","channel":"ALIPAY"}'
# {"code":"SUCCESS","data":{"paid":true,...}}
匹配规则要点:
urlPath精确匹配路径不含 query;要匹配 query 用queryParameters(equalTo/matches/absent)。equalToJson默认要求对象结构完全一致;ignoreExtraElements: true才允许请求多字段,ignoreArrayOrder: true忽略数组顺序。- 优先级数字越小越优先,默认 5。精确 stub 设 1,兜底 stub 设 10,否则兜底会先命中。
matchesJsonPath只校验“路径存在”,不校验值;要校验值写{"matchesJsonPath": {"expression": "$.orderNo", "equalTo": "P1001"}}。
2.3 延迟、故障注入、兜底¶
{
"priority": 10,
"request": { "method": "POST", "urlPath": "/pay/query" },
"response": {
"status": 200,
"fixedDelayMilliseconds": 6000,
"jsonBody": { "code": "SUCCESS" }
}
}
{
"priority": 1,
"request": { "method": "POST", "urlPath": "/pay/query",
"bodyPatterns": [{ "matchesJsonPath": "$.orderNo[?(@ == 'TIMEOUT')]" }] },
"response": { "fault": "CONNECTION_RESET_BY_PEER" }
}
可用 fault:CONNECTION_RESET_BY_PEER、EMPTY_RESPONSE、MALFORMED_RESPONSE_CHUNK、RANDOM_DATA_THEN_CLOSE。区别要分清:
| fault | 客户端看到 | 对应真实场景 |
|---|---|---|
CONNECTION_RESET_BY_PEER |
Connection reset by peer |
下游进程崩溃、LB 掐连接 |
EMPTY_RESPONSE |
Remote end closed connection without response |
上游未写响应就关连接 |
MALFORMED_RESPONSE_CHUNK |
解析 chunked 失败 | 中间设备截断 |
fixedDelayMilliseconds: 6000 |
读超时 | 下游慢查询 |
“超时”必须用延迟模拟,“连接断开”必须用 fault,两者在客户端走的异常分支不同——只测前者会漏掉重试逻辑里的连接异常处理。
2.4 录制真实流量¶
# 开始录制:所有打到 9999 的请求转发到真实下游并记录
curl -sS -X POST http://localhost:9999/__admin/recordings/start \
-H 'Content-Type: application/json' \
-d '{"targetBaseUrl":"https://sandbox.example.com",
"filters":{"urlPathPattern":"/api/.*","method":"POST"}}'
# 跑一轮真实业务用例……
curl -sS -X POST http://localhost:9999/__admin/recordings/stop
# 生成的 stub 会落在 mappings/ 下,__files/ 下是对应响应体
# 把录制结果固化成快照(用于回归对比),并清空状态
curl -sS -X POST http://localhost:9999/__admin/recordings/snapshot \
-H 'Content-Type: application/json' -d '{"persist":true}'
curl -sS -X POST http://localhost:9999/__admin/reset
录制回放的坑:录到的响应里带真实 token、时间戳、自增 ID,回放时断言会随日期变化。落库前要脱敏并参数化,例如把 "paidAt" 换成固定值,把 "orderNo" 改成请求里回显。
三、MockServer(简要)¶
需要“动态期望 + 断言收到过哪些请求”时用 MockServer:
curl -sS -X PUT http://localhost:1080/mockserver/expectation \
-H 'Content-Type: application/json' \
-d '{
"httpRequest": { "method": "GET", "path": "/inventory/1001" },
"httpResponse": { "statusCode": 200,
"body": {"skuId":1001,"stock":0},
"delay": {"timeUnit":"SECONDS","value":2} },
"times": { "remainingTimes": 1 }
}'
# 验证调用次数(断言被测服务确实调了下游)
curl -sS -X PUT http://localhost:1080/mockserver/verify \
-H 'Content-Type: application/json' \
-d '{"httpRequest":{"method":"GET","path":"/inventory/1001"},"times":{"atLeast":1}}'
WireMock 强在 stub 文件化和录制,MockServer 强在验证调用与动态期望;纯 HTTP 场景两者都能覆盖,选团队已有依赖的那个即可。
四、pytest 内三种替身¶
4.1 responses:拦截 requests¶
# requirements: requests==2.32.3 responses==0.25.3
import pytest, requests, responses
@responses.activate
def test_pay_success():
responses.add(
responses.POST, "https://pay.example.com/pay/query",
json={"code": "SUCCESS", "data": {"paid": True}},
status=200,
match=[responses.matchers.json_params_matcher({"orderNo": "P1001"})],
)
r = requests.post("https://pay.example.com/pay/query",
json={"orderNo": "P1001"}, timeout=3)
assert r.json()["data"]["paid"] is True
assert len(responses.calls) == 1
assert responses.calls[0].request.headers["Content-Type"] == "application/json"
未匹配时的真实报错:
requests.exceptions.ConnectionError: Connection refused by Responses -
the call doesn't match any registered mock.
看到它先检查 URL 是否带尾斜杠、query 是否一致、json_params_matcher 是否多/少字段。responses 会连 query string 一起比较(0.22 之后 match_querystring 参数已废弃)。
4.2 requests-mock:pytest 插件¶
def test_get_user(requests_mock):
requests_mock.get(
"https://api.example.com/api/v1/users/1",
json={"code": 0, "data": {"id": 1, "name": "qa"}},
headers={"Content-Type": "application/json"},
)
r = requests.get("https://api.example.com/api/v1/users/1", timeout=3)
assert r.json()["data"]["id"] == 1
这个报错说明 URL 没注册,常见原因是代码里 base_url 末尾多了一个 / 导致拼成 //api/v1。
4.3 monkeypatch:替换下游客户端¶
被测代码(app/service/refund.py):
from app.client.pay import pay_client
def refund(order_no: str) -> dict:
result = pay_client.query(order_no) # 需要在测试里替换
if not result["paid"]:
return {"code": "NOT_PAID"}
return {"code": "REFUNDED"}
测试:
class FakePayClient:
def __init__(self, paid): self.paid = paid
def query(self, order_no): return {"paid": self.paid, "orderNo": order_no}
def test_refund_when_paid(monkeypatch):
monkeypatch.setattr("app.service.refund.pay_client", FakePayClient(True))
from app.service.refund import refund
assert refund("P1001")["code"] == "REFUNDED"
def test_refund_when_not_paid(monkeypatch):
monkeypatch.setattr("app.service.refund.pay_client", FakePayClient(False))
from app.service.refund import refund
assert refund("P1001")["code"] == "NOT_PAID"
必须 patch 使用处(app.service.refund.pay_client),patch 定义处(app.client.pay.pay_client)在已 import 的情况下不生效。替身只实现被调用的方法,别顺手复制真实客户端的全部逻辑——那就是第二份实现。
五、Pact 契约测试¶
消费方(订单服务)写期望,生成 pact 文件:
# requirements: pact-python==2.0.1
import atexit, requests
from pact import Consumer, Provider
pact = Consumer("order-service").has_pact_with(
Provider("pay-service"), host_name="127.0.0.1", port=1234,
pact_dir="./pacts")
pact.start_service()
atexit.register(pact.stop_service)
def test_query_payment():
(pact
.given("订单 P1001 已支付")
.upon_receiving("查询支付状态")
.with_request("GET", "/pay/P1001")
.will_respond_with(200, body={"code": "SUCCESS", "paid": True}))
with pact:
r = requests.get("http://127.0.0.1:1234/pay/P1001", timeout=3)
assert r.json()["paid"] is True
提供方(支付服务)用同一份 pact 做验证,通常放在 CI 的“提供方流水线”:
# 发布 pact 到 broker(pact_broker-client)
pact-broker publish ./pacts \
--consumer-app-version=$(git rev-parse --short HEAD) \
--branch=main \
--broker-base-url=https://pact.example.com
# 提供方验证前先确认能不能部署
pact-broker can-i-deploy \
--pacticipant pay-service --version=$(git rev-parse --short HEAD) \
--to-environment=prod --broker-base-url=https://pact.example.com
判定标准:只有“消费方测试通过 + 提供方验证通过 + can-i-deploy 为 true”三件事同时成立,契约才算有效。只写消费方测试,等于把 Mock 换了个名字。
六、Mock 漂移:一次真实事故复盘¶
测试代码里的 stub(三个月前写的):
消费方解析:
支付服务 2.1 改成统一响应包装,真实返回:
上线后 is_paid 抛 KeyError: 'paid'。同一批漂移还包括:
| 漂移类型 | Mock 里 | 真实服务 | 后果 |
|---|---|---|---|
| 字段层级 | 顶层 paid |
data.paid |
KeyError |
| 金额类型 | 100.0(number) |
"100.00"(string) |
断言 == 100 失败或精度误差 |
| 状态码 | 一律 200 | 创建成功返回 201,重复返回 409 | 用例误判失败 |
| 枚举值 | PAID |
新增 PARTIAL_REFUND |
前端分支未覆盖 |
| 时间格式 | 2026-10-22 11:20:00 |
ISO8601 带时区 | 解析报错 |
防漂移的四道闸:
- 每个 stub 配一个 JSON Schema 断言,字段层级和类型变化会立刻失败。
- 核心链路保留一条不打 Mock 的集成用例,跑在 sandbox 环境,每晚执行。
- 用 Pact 做消费方/提供方双向验证,纳入 CI 门禁。
- 录制回放定期刷新(例如每周一次),并在 diff 里人工确认字段变化。
七、测试数据构造¶
7.1 工厂 + Faker¶
# requirements: factory_boy==3.3.1 Faker==30.1.0
import factory
from faker import Faker
fake = Faker("zh_CN")
class UserFactory(factory.Factory):
class Meta:
model = dict
id = factory.Sequence(lambda n: 10000 + n)
name = factory.LazyFunction(fake.name)
mobile = factory.LazyFunction(fake.unique.phone_number)
email = factory.LazyAttribute(lambda o: f"qa_{o.id}@example.com")
status = "ACTIVE"
user = UserFactory(status="FROZEN")
factory.Sequence 保证同进程内唯一;fake.unique 只在单进程内唯一,pytest-xdist 并发跑时要在值里加 worker 前缀:
import os, uuid
worker = os.environ.get("PYTEST_XDIST_WORKER", "master")
email = f"qa+{worker}-{uuid.uuid4().hex[:8]}@example.com"
7.2 分层构造¶
| 层 | 内容 | 维护方式 |
|---|---|---|
| 基础数据 | 枚举、币种、仓库编码 | 配置文件,禁止硬编码在用例里 |
| 业务实体 | 用户、商品、商户 | 工厂 + 接口创建,不直插数据库 |
| 场景数据 | 已支付订单、过期券、冻结账户 | 通过接口按业务动作构造,保证状态合法 |
场景数据尽量走接口构造。直接 INSERT 出“已支付订单”而跳过支付动作,常常缺了流水、账务、消息,测退款时才会暴露。
7.3 唯一性与金额¶
import time, uuid
def unique_mobile() -> str:
# 13 + 9 位,用时间戳后 7 位 + 2 位随机,避免同秒并发碰撞
return "13" + str(int(time.time() * 1000))[-7:] + uuid.uuid4().hex[:2]
from decimal import Decimal, ROUND_HALF_UP
def money(v: str) -> str:
return str(Decimal(v).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
金额不要用 float 断言:0.1 + 0.2 != 0.3,且 json 解析大数会丢精度(超过 2^53 的整数在 JS 里直接变形)。断言用字符串或 Decimal。
7.4 清理¶
import uuid, pytest
@pytest.fixture
def run_id():
return uuid.uuid4().hex[:8]
@pytest.fixture
def created_user(api, run_id):
email = f"qa+{run_id}@example.com"
user = api.post("/api/v1/users", json={"email": email, "name": "qa"})
yield user
api.delete(f"/api/v1/test/users/{user['id']}") # 失败也不影响其他用例
兜底清理(测试库专用账号):
DELETE FROM orders
WHERE remark LIKE 'qa-%'
AND created_at < now() - interval '1 day';
DELETE FROM users
WHERE email LIKE 'qa+%@example.com'
AND created_at < now() - interval '7 days';
清理策略对比:
| 策略 | 优点 | 风险 |
|---|---|---|
| 用例 teardown 按 ID 删 | 精确 | 用例中途崩溃会残留,需要 try/finally |
| 事务回滚 | 干净、快 | 仅限单库直连,走 HTTP 的用例用不上 |
| 测试租户/独立库 | 隔离最好 | 成本高,跨服务调用仍可能污染 |
| 定时脚本兜底 | 覆盖残留 | 只能按前缀/时间,误删风险要评估 |
八、坑位表与检查清单¶
| 坑 | 现象 | 处理 |
|---|---|---|
stub 只写 url 不写方法 |
GET 也能命中 POST 的 stub | 必须写 method |
| 兜底优先级过高 | 精确 stub 永远不命中 | 兜底 priority 设 10,精确设 1 |
| 忘记校验请求体 | Mock 永远成功,参数 bug 漏测 | 至少加 matchesJsonPath + 一个 equalToJson |
延迟用 sleep 在测试里 |
用例变慢且不稳定 | 用 fixedDelayMilliseconds 控制下游 |
responses 与 requests.Session |
patch 不生效 | responses 基于 adapter,Session 也生效;自建 HTTPAdapter 需确认 |
| monkeypatch 打错位置 | 改了没反应 | patch 使用处,不 patch 定义处 |
| 录制数据含真实 token | 回放时 401 | 落库前脱敏,token 用占位符 |
| 数据清理顺序 | 外键冲突删不掉 | 按依赖倒序删,或整租户删 |
| Faker 随机数据触发业务校验 | 用例偶发失败 | 关键字段用固定值,随机只用于非关键字段 |
检查清单:
- 每个 stub 是否声明了 method + 路径 + 至少一个请求校验?
- 是否有超时、连接重置、5xx 三类异常 stub?
- 是否有
priority兜底,且兜底不会遮住精确匹配? - 消费方测试是否配了 JSON Schema 或 Pact?
- 核心链路是否保留至少一条真实集成用例?
- 测试数据是否带
run_id前缀,失败后能按前缀清理? - 金额、时间、枚举是否用固定值或格式化断言?