跳转至

接口 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:

java -jar mockserver-netty-no-dependencies-5.15.0.jar -serverPort 1080 -logLevel INFO
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
requests_mock.exceptions.NoMockAddress: No mock address: GET https://api.example.com/api/v1/users/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(三个月前写的):

{ "code": "SUCCESS", "paid": true }

消费方解析:

def is_paid(order_no):
    return pay_client.query(order_no)["paid"]      # 单测一直通过

支付服务 2.1 改成统一响应包装,真实返回:

{ "code": 0, "data": { "paid": true } }

上线后 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 带时区 解析报错

防漂移的四道闸:

  1. 每个 stub 配一个 JSON Schema 断言,字段层级和类型变化会立刻失败。
  2. 核心链路保留一条不打 Mock 的集成用例,跑在 sandbox 环境,每晚执行。
  3. 用 Pact 做消费方/提供方双向验证,纳入 CI 门禁。
  4. 录制回放定期刷新(例如每周一次),并在 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 随机数据触发业务校验 用例偶发失败 关键字段用固定值,随机只用于非关键字段

检查清单:

  1. 每个 stub 是否声明了 method + 路径 + 至少一个请求校验?
  2. 是否有超时、连接重置、5xx 三类异常 stub?
  3. 是否有 priority 兜底,且兜底不会遮住精确匹配?
  4. 消费方测试是否配了 JSON Schema 或 Pact?
  5. 核心链路是否保留至少一条真实集成用例?
  6. 测试数据是否带 run_id 前缀,失败后能按前缀清理?
  7. 金额、时间、枚举是否用固定值或格式化断言?

相关阅读