WeClaw_83|Mock、Fixture 与录制回放:三种测试替身术的工程实战
系列文章第 83 篇 - 从 WeClaw v7.0.0 工作流队列 282 项测试中,拆解 Mock / Fixture / 录制回放三种测试隔离策略的选型逻辑与实战代码
📚 专栏信息
《从零到一构建跨平台 AI 助手:WeClaw 实战指南》专栏
本文是模块九【工程实践与质量保障】第 1 篇。 WeClaw v7.0.0 是一次架构级重大迭代——新增工作流队列核心域(7 个模块)、Agent 工具包、UI 组件包, 横跨持久化调度、安全模型、模板分享四个层面,累计 282 项自动化测试全部通过。 本篇以这次迭代的真实测试代码为案例,讲解三种"测试替身术"的本质、选型与陷阱。
👨💻 作者与项目
作者简介:翁勇刚 WENG YONGGANG 新概念龙虾-WeClaw 开发团队负责人,一群专注于跨平台 AI 应用的实践者 理念:"再复杂的技术,也能用代码讲清楚"
- 💻 项目地址:https://github.com/wyg5208/weclaw.git
- 🌐 官网地址:https://weclaw.link
- 📝 作者 CSDN:https://blog.csdn.net/yweng18
- ⭐ 欢迎 Star⭐、Fork🍴、贡献代码🤝
📝 摘要
核心问题:单元测试要"快、稳、独立",但真实系统依赖数据库、网络、LLM、GUI 框架——如何在不调用任何外部服务的前提下验证业务逻辑?
三种替身术:
- Mock(模拟对象):手写一个"假的依赖",你完全控制它的行为
- Fixture(测试夹具):预先搭好"舞台"——临时数据库、配置对象、初始数据
- 录制回放(Record & Replay):把一次真实交互录下来,之后离线反复播放
关键成果:
- v7.0.0 迭代 282 项测试在无 PySide6/litellm 环境下全部通过
- 连续复跑 3-15 次确认无 flaky
- 过程中发现并修复 2 处测试自身竞态 + 2 处生产缺陷
适合读者:想系统理解测试隔离策略、正在为异步/IO 密集系统编写测试的 Python 开发者
阅读时长:约 18 分钟
关键词:Mock、Fixture、录制回放、pytest、asyncio测试、测试隔离、工作流队列
一、为什么需要"替身"?——从一个真实困境说起
1.1 v7.0.0 的测试困境
WeClaw v7.0.0 工作流队列的核心调度器 SerialQueueDispatcher 需要:
| 真实依赖 | 为什么不能直接用 |
|---|---|
| GuiAgent(LLM 对话) | 调用一次 3-15 秒,282 项测试跑完要 1 小时+ |
| PySide6 GUI 框架 | CI 环境没有显示器,QApplication 直接崩溃 |
| litellm(模型调用) | 需要 API Key,有速率限制,结果不确定 |
| 真实 SQLite 数据库文件 | 测试间互相污染,并发跑测试时文件锁冲突 |
如果直接对接真实依赖,测试会变成:慢、不稳定、不可重复、需要外部环境。 这违反了单元测试的四条基本原则——快(Fast)、独立(Isolated)、可重复(Repeatable)、自验证(Self-validating)。
1.2 三种替身术的定位
┌─────────────────────────────────────────────────────────┐
│ 测试隔离策略光谱 │
│ │
│ 完全虚构 ◄─────────────────────────────► 完全真实 │
│ │
│ Mock Fixture 录制回放 真实调用 │
│ (我编的) (我搭的舞台) (录下来的) (现场直播) │
│ │
│ 控制力:高 ────────────────────────────── 低 │
│ 真实度:低 ────────────────────────────── 高 │
│ 维护成本:接口变 → 手动改 ──────── 接口变 → 重新录制 │
└─────────────────────────────────────────────────────────┘
三者不是互斥的——一个成熟的测试套件往往同时使用三种。下面逐一拆解。
二、Mock:我假装是你
2.1 本质
Mock 的核心思想:用一个完全受控的假对象替代真实依赖,让测试只验证目标逻辑。
你不需要真的调用 LLM,只需要一个"被调用时返回预设结果"的替身。
2.2 WeClaw 实战:FakeExecutor
v7.0.0 调度器测试中最核心的 Mock 是 FakeExecutor——它替代了真实的 GuiAgent.chat():
# test_scripts/test_workflow_queue_dispatcher.py
class FakeExecutor:
"""可编程的假执行器:按 instruction 内容决定行为,记录执行顺序与并发峰值。"""
def __init__(self, delay: float = 0.02):
self.delay = delay
self.order: list[str] = [] # 记录执行顺序
self.concurrent = 0 # 当前并发数
self.max_concurrent = 0 # 并发峰值(验证串行性)
async def __call__(self, item: WorkflowItem) -> ExecutionOutcome:
self.concurrent += 1
self.max_concurrent = max(self.max_concurrent, self.concurrent)
self.order.append(item.instruction)
try:
if item.instruction.startswith("SLOW_CANCEL"):
await asyncio.sleep(5) # 模拟长任务,会被 request_stop() 取消
return ExecutionOutcome(status=ItemStatus.SUCCEEDED)
await asyncio.sleep(self.delay)
if item.instruction.startswith("FAIL"):
return ExecutionOutcome(status=ItemStatus.FAILED, error="模拟失败")
return ExecutionOutcome(status=ItemStatus.SUCCEEDED, result_summary="ok")
finally:
self.concurrent -= 1
设计精髓:
- 可编程行为:通过指令前缀(
FAIL、SLOW_CANCEL)控制返回成功/失败/超时,一个类覆盖所有分支 - 可观测状态:
order记录执行顺序,max_concurrent验证"严格串行"——如果调度器有并发 Bug,max_concurrent > 1立刻暴露 - 零外部依赖:不需要 LLM、不需要网络、不需要 GUI,20ms 就能模拟一次"执行"
2.3 验证串行性的断言
async def _test_strict_serial_and_advance():
with tempfile.TemporaryDirectory(prefix="weclaw_wfq_disp_") as tmp:
storage = WorkflowQueueStorage(db_path=Path(tmp) / "wfq.db")
service = WorkflowQueueService(storage)
executor = FakeExecutor(delay=0.02)
dispatcher = SerialQueueDispatcher(service, executor)
# 提交 5 条指令
queue = await service.create_queue("测试", items=["A", "B", "C", "D", "E"])
await dispatcher.start()
await _wait_for_queue_status(service, queue.queue_id, {QueueStatus.COMPLETED})
# 核心断言:执行顺序 = 提交顺序,且从未并发
assert executor.order == ["A", "B", "C", "D", "E"]
assert executor.max_concurrent == 1, f"检测到并发: {executor.max_concurrent}"
如果不用 Mock 而用真实 GuiAgent:
- 每次调用 3-15 秒 → 5 条指令至少 15 秒
- LLM 输出不确定 → 无法断言精确顺序
- 需要 API Key → CI 跑不了
2.4 unittest.mock 的标准用法
在 tests/test_binding_and_routing.py 中,使用了 Python 标准库的 patch:
from unittest.mock import MagicMock, AsyncMock, patch
@patch("remote_server.auth.user_manager.send_websocket_message")
def test_binding_notifies_only_target_device(mock_send):
"""绑定成功后只通知目标设备,不广播"""
mock_send.return_value = None
# ... 执行绑定逻辑 ...
# 验证:只调用了一次,且目标是正确的 device_id
mock_send.assert_called_once()
call_args = mock_send.call_args
assert call_args[0][0] == "target_device_id"
2.5 Mock 的陷阱:过度 Mock 导致"测试通过但系统是坏的"
v7.0.0 迭代记录中有一条深刻的教训:
P0 的
reorder_item()测试通过、接口调用成功,但因为claim_next_item()从未读取它修改的字段,实际执行顺序完全不受影响——这种"测试通过但业务逻辑是假的"情况只能靠端到端验证才能发现。
这就是过度 Mock 的典型症状:你 Mock 了太多东西,测试验证的是"Mock 的行为"而不是"系统的行为"。
WeClaw 的对策:调度器测试用 FakeExecutor(Mock 执行层),但 Storage/Service 层用真实 SQLite(不 Mock)——在"可控"与"真实"之间找到平衡点。
三、Fixture:我先把舞台搭好
3.1 本质
Fixture 的核心思想:在每个测试运行前,自动准备好它需要的环境和数据;测试结束后,自动清理干净。
它不是"假的"——临时数据库是真实的数据库,只是用完就扔。
3.2 WeClaw 实战:工厂函数式 Fixture
v7.0.0 存储层测试没有使用 pytest 的 @fixture 装饰器(因为要支持无 pytest 直接运行),而是用工厂函数实现等价语义:
# test_scripts/test_workflow_queue_storage.py
def _new_storage(tmp: Path) -> WorkflowQueueStorage:
"""每个测试拿到一个全新的、隔离的存储实例"""
return WorkflowQueueStorage(db_path=tmp / "workflow_queue.db")
def _make_queue(**overrides) -> WorkflowQueue:
"""队列工厂:默认值 + 按需覆盖"""
defaults = dict(
queue_id=new_id("q"),
name="测试队列",
source=QueueSource.DESKTOP_INTERACTIVE,
session_id="desktop_main",
)
defaults.update(overrides)
return WorkflowQueue(**defaults)
def _make_item(queue_id: str, sequence: int, instruction: str = "指令", **overrides) -> WorkflowItem:
"""项目工厂:最小必填参数 + 按需扩展"""
defaults = dict(
item_id=new_id("i"),
queue_id=queue_id,
sequence=sequence,
instruction=instruction,
)
defaults.update(overrides)
return WorkflowItem(**defaults)
每个测试函数内部用 tempfile.TemporaryDirectory 确保隔离:
def test_schema_and_crud():
with tempfile.TemporaryDirectory(prefix="weclaw_wfq_") as tmp:
storage = _new_storage(Path(tmp)) # 全新数据库
queue = _make_queue(name="我的队列") # 全新队列对象
storage.create_queue(queue)
fetched = storage.get_queue(queue.queue_id)
assert fetched.name == "我的队列"
# with 块结束 → 临时目录自动删除 → 零残留
3.3 pytest 原生 Fixture 对比
在 tests/test_binding_and_routing.py 中,使用了 pytest 内置的 tmp_path fixture:
class TestDeviceFingerprintUniqueConstraint:
def test_same_fingerprint_cannot_bind_two_users(self, tmp_path):
"""同一设备指纹不能同时绑定两个不同用户"""
um = make_user_manager(tmp_path / "test.db") # pytest 自动提供临时路径
user1 = um.create_user("user1", "password123")
user2 = um.create_user("user2", "password456")
# ...
tmp_path 是 pytest 内置 fixture——每个测试函数自动获得一个唯一的临时目录,测试结束后由 pytest 统一清理。
3.4 全局 Fixture:conftest.py
# conftest.py(项目根目录)
"""pytest 全局配置 - 设置测试路径"""
import sys
from pathlib import Path
WECLAW_SERVER_DIR = str(Path(__file__).parent / "weclaw_server")
if WECLAW_SERVER_DIR not in sys.path:
sys.path.insert(0, WECLAW_SERVER_DIR)
这个"隐形 Fixture"确保所有测试文件都能 from remote_server.auth.user_manager import UserManager,无需每个文件重复写路径注入。
3.5 Fixture 设计原则(WeClaw 实践总结)
| 原则 | 说明 | WeClaw 示例 |
|---|---|---|
| 最小化 | 只提供测试真正需要的 | _make_item() 只有 4 个必填字段 |
| 可覆盖 | 默认值 + **overrides 模式 | _make_queue(status=QueueStatus.PAUSED) |
| 自清理 | 不留残留,测试间零耦合 | TemporaryDirectory 上下文管理器 |
| 层次化 | 全局 → 模块 → 函数三级 | conftest.py → 文件头 helper → 测试内 |
四、录制回放:我把你的表演录下来
4.1 本质
录制回放的核心思想:第一次运行时,把真实交互(请求-响应对)保存到文件;之后每次运行,直接从文件读取,不再发起真实调用。
与 Mock 的区别:Mock 的返回值是你编的;录制回放的返回值是真实系统曾经返回的。
4.2 WeClaw 的"准录制回放"实践
v7.0.0 安全策略测试采用了一种"准录制回放"策略——直接使用仓库中的真实配置文件作为测试输入:
# test_scripts/test_workflow_queue_security.py
"""
不依赖 PySide6/litellm;`config/tools.json` 使用仓库真实文件,验证与实际配置
的一致性(而不是构造假配置绕过真实判定逻辑)。
"""
def test_is_action_safe_for_unattended_real_config():
# 直接读取仓库中真实的 config/tools.json
check(
"低风险工具(datetime_tool)无需预授权",
wq_security.is_action_safe_for_unattended("datetime_tool", "get_current_time"),
)
check(
"高风险 shell.run 不在 safe_actions 中,需要预授权",
not wq_security.is_action_safe_for_unattended("shell", "run"),
)
check(
"未知工具保守返回 False(不视为安全)",
not wq_security.is_action_safe_for_unattended("__not_a_real_tool__", "anything"),
)
为什么不用 Mock 构造一个假的 tools.json?
迭代文档给出了明确理由:
验证与实际配置的一致性(而不是构造假配置绕过真实判定逻辑)。
如果你 Mock 了 tools.json,测试验证的是"你假设的配置长什么样";用真实文件,验证的是"系统在当前真实配置下是否安全"。配置改了,测试立刻能发现。
4.3 标准录制回放工具(VCR.py 示例)
对于 HTTP API 测试,Python 生态有成熟的录制回放库:
# 以 vcrpy 为例(WeClaw 远程桥接测试的推荐方案)
import vcr
import requests
@vcr.use_cassette("tests/cassettes/weather_api.yaml")
def test_weather_query():
# 第一次运行:真实请求 → 录制到 yaml
# 之后运行:直接从 yaml 读取响应,零网络
resp = requests.get("https://api.weather.com/v1/current?city=beijing")
assert resp.status_code == 200
assert "temperature" in resp.json()
录制后的 weather_api.yaml 长这样:
interactions:
- request:
body: null
headers: {}
method: GET
uri: https://api.weather.com/v1/current?city=beijing
response:
body:
string: '{"temperature": 28, "humidity": 65, "city": "beijing"}'
headers:
content-type: application/json
status:
code: 200
message: OK
version: 1
4.4 录制回放的适用场景
| 场景 | 为什么适合录制回放 |
|---|---|
| 第三方 API(天气、股票、翻译) | 有速率限制、收费、结果随时间变化 |
| LLM 响应 | 不确定性高、成本高、速度慢 |
| OAuth 认证流程 | 不可能每次测试都真的走一遍授权 |
| WebSocket 长连接交互 | 建连成本高、状态复杂 |
4.5 录制回放的陷阱
| 陷阱 | 说明 | 对策 |
|---|---|---|
| 数据过期 | 录制的响应可能已不符合新 API 版本 | 定期重新录制 + CI 中加"新鲜度检查" |
| 敏感信息泄漏 | 响应中可能包含 Token/用户数据 | 录制后脱敏(vcrpy 支持 filter_headers) |
| 过度依赖快照 | 只验证"和上次一样",不验证"是否正确" | 录制回放 + 关键断言结合使用 |
五、三者的协同:v7.0.0 测试架构全景
5.1 分层策略
┌─────────────────────────────────────────────────────────────┐
│ test_workflow_queue_dispatcher.py (73 项) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Mock 层:FakeExecutor 替代 GuiAgent │ │
│ │ Fixture 层:TemporaryDirectory + Service 工厂 │ │
│ │ 真实层:WorkflowQueueStorage(真实 SQLite) │ │
│ └─────────────────────────────────────────────────────┘ │
├─────────────────────────────────────────────────────────────┤
│ test_workflow_queue_storage.py (87 项) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Fixture 层:_new_storage / _make_queue / _make_item │ │
│ │ 真实层:100% 真实 SQLite 操作(零 Mock) │ │
│ └─────────────────────────────────────────────────────┘ │
├─────────────────────────────────────────────────────────────┤
│ test_workflow_queue_security.py (35 项) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 录制回放层:真实 config/tools.json 作为测试输入 │ │
│ │ 真实层:security.py 的完整判定逻辑(零 Mock) │ │
│ └─────────────────────────────────────────────────────┘ │
├─────────────────────────────────────────────────────────────┤
│ test_workflow_queue_tool.py (33 项) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Fixture 层:_make_tool_with_service 工厂 │ │
│ │ 真实层:ToolRegistry + Service + Storage 全真实 │ │
│ │ 明确注释:"直接使用真实 WorkflowQueueStorage(不 mock)"│ │
│ └─────────────────────────────────────────────────────┘ │
├─────────────────────────────────────────────────────────────┤
│ test_workflow_queue_templates.py (54 项) │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 录制回放层:3 个内置示例模板 JSON/CSV 作为基准数据 │ │
│ │ Fixture 层:临时目录 + 构造的模板字符串 │ │
│ └─────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
5.2 选型决策树
需要测试的依赖是什么?
│
├─ 纯计算逻辑(无 IO)
│ └─ 不需要替身,直接测
│
├─ 数据库 / 文件系统
│ └─ Fixture:临时目录 + 真实引擎(不 Mock SQLite)
│ 理由:SQLite 本身够快(内存级),Mock 它反而丢失真实性
│
├─ 外部服务(LLM / HTTP API / WebSocket)
│ ├─ 只关心"是否被调用、参数对不对" → Mock
│ ├─ 关心"返回值处理逻辑" → 录制回放
│ └─ 关心"端到端正确性" → 集成测试(少量,慢速,手动触发)
│
├─ GUI 框架(PySide6)
│ └─ Mock 或跳过(`@pytest.mark.skipif(not HAS_PYSIDE6)`)
│
└─ 配置文件(tools.json / default.toml)
└─ 录制回放思路:直接用仓库真实文件,不构造假配置
六、异步测试中的竞态陷阱——来自 v7.0.0 的血泪教训
6.1 问题:固定 sleep 断言
v7.0.0 迭代中发现的两处测试竞态,本质是Fixture 的时间维度设计失误:
# ❌ 错误写法:固定延时后断言瞬时状态
await service.create_workflow("test", items=["A"])
await asyncio.sleep(0.05) # "猜"调度器已经开始了
assert executor.order == ["A"] # 偶发失败!
create_workflow 内部会唤醒 dispatcher 协程,但"唤醒"到"实际执行"之间有调度延迟。
0.05 秒在大多数时候够用,但在 CI 高负载时不够——这就是 flaky test 的根源。
6.2 对策:条件轮询等待
# ✅ 正确写法:轮询等待目标状态
async def _wait_until(predicate, timeout: float = 5.0, interval: float = 0.02) -> bool:
loop = asyncio.get_event_loop()
deadline = loop.time() + timeout
while loop.time() < deadline:
if predicate():
return True
await asyncio.sleep(interval)
return predicate()
# 使用
await service.create_workflow("test", items=["A"])
await _wait_until(lambda: "A" in executor.order, timeout=5.0)
assert executor.order == ["A"]
原则:给异步系统写测试时,永远不要"猜时机",要"等状态"。
6.3 更可靠的等待:轮询队列状态而非执行器记录
async def _wait_for_queue_status(service, queue_id, statuses, timeout=5.0):
"""轮询队列真实状态(比轮询 executor.order 更可靠:
order 在执行器调用*开始*时即写入,不代表 finalize_item 已完成)"""
loop = asyncio.get_event_loop()
deadline = loop.time() + timeout
while loop.time() < deadline:
q = await service.get_queue(queue_id)
if q and q.status in statuses:
return q.status
await asyncio.sleep(0.03)
return None
这段注释揭示了一个微妙问题:executor.order.append() 发生在执行开始时,但队列状态迁移发生在执行结束后。如果你要断言"队列已完成",必须等队列状态,而不是执行器记录。
七、三者的对比总结
| 维度 | Mock | Fixture | 录制回放 |
|---|---|---|---|
| 本质 | 手写假对象 | 预置环境/数据 | 保存真实交互 |
| 真实度 | 低(你编的) | 中(数据真、环境临时) | 高(真实响应) |
| 速度 | 极快(ns 级) | 快(ms 级,SQLite 建表) | 快(读文件) |
| 维护成本 | 接口变 → 手动改 | 低(自动清理) | 接口变 → 重新录制 |
| 典型工具 | unittest.mock / FakeXxx | pytest.fixture / tmp_path | vcrpy / responses |
| WeClaw 用例 | FakeExecutor 替代 GuiAgent | TemporaryDirectory + 工厂函数 | 真实 tools.json 作为输入 |
| 最大风险 | 过度 Mock → 测了个寂寞 | 搭建成本过高 → 测试变慢 | 数据过期 → 假绿 |
八、实战建议:何时用什么?
8.1 优先用 Fixture 的情况
- 依赖本身够快、够轻(SQLite、内存缓存、本地文件)
- 你需要验证真实的 IO 行为(SQL 语法、事务隔离、文件编码)
- WeClaw 的选择:Storage 层 87 项测试零 Mock,全用真实 SQLite
8.2 必须用 Mock 的情况
- 依赖慢、贵、不确定(LLM 调用、第三方 API)
- 你需要验证交互行为(是否调用了、参数对不对、调用了几次)
- 依赖不可用(CI 没有显示器 → Mock GUI)
- WeClaw 的选择:FakeExecutor 替代 3-15 秒的 LLM 调用
8.3 适合录制回放的情况
- 外部 API 有速率限制/收费
- 响应内容复杂但相对稳定
- 你需要离线可重复的集成测试
- WeClaw 的选择:安全测试直接读仓库真实配置("穷人版录制回放")
8.4 黄金法则
能真实就真实,必须隔离才替身,替身时只替"不可控"的那一层。
v7.0.0 的 282 项测试之所以能在无 PySide6/litellm 环境下全部通过且无 flaky, 核心原因不是"Mock 用得多",而是精确选择了在哪一层切断真实依赖:
- 执行层(GuiAgent/LLM)→ Mock(FakeExecutor)
- 存储层(SQLite)→ 真实(Fixture 提供临时实例)
- 配置层(tools.json)→ 真实(录制回放思路)
九、结语
测试替身术不是"造假",而是控制变量。
科学家做实验时,不会把整个自然界搬进实验室——他们控制温度、压力、浓度,只让一个变量变化。 Mock 控制"谁被调用",Fixture 控制"环境长什么样",录制回放控制"外部世界返回什么"。 三者协同,让你在 0.2 秒内跑完 87 项存储测试,在 20ms 内模拟一次 LLM 对话, 在离线环境中验证安全策略与真实配置的一致性。
v7.0.0 迭代记录的最后一条经验教训说得好:
"编码完成"与"经过验证"之间的差距,只能靠认真测试来弥补。
而认真测试的前提,是选对替身。
下一篇预告:WeClaw_84|故障注入测试:如何在实验室里制造"崩溃"