Harness 详解
约 1259 字大约 4 分钟
2026-03-19
TestHarness 与 PluginTestHarness 的完整使用指南
目录
1. TestHarness 生命周期
TestHarness 在后台启动一个完整的 BotClient（使用 MockAdapter），无需连接 NapCat。
async with（推荐）
from ncatbot.testing import TestHarness
from ncatbot.testing.factories.qq import group_message
async with TestHarness() as h:
# h.bot, h.mock_api, h.dispatcher 可用
await h.inject(group_message("hi"))
await h.settle()
# 自动 stop手动管理
h = TestHarness()
await h.start() # 启动 BotClient
try:
# ... 测试逻辑 ...
finally:
await h.stop() # 停止 BotClient内部做了什么
start()→ 调用BotClient.run_async()，启动 MockAdapter + Dispatcher + HandlerDispatcherstop()→ 调用BotClient.shutdown()，停止所有后台任务- MockAdapter 替代了真实的 NapCat 连接，所有 API 调用被
MockAPIBase子类记录
2. 事件注入
注入单个事件
from ncatbot.testing.factories.qq import group_message
await h.inject(group_message("hello"))事件会根据 event_data.platform 自动路由到对应平台的 MockAdapter。
注入多个事件
from ncatbot.testing.factories.qq import group_message, private_message
await h.inject_many([
group_message("a"),
private_message("b"),
group_message("c"),
])settle — 等待处理
settle() 给 handler 一点时间执行（默认 50ms）：
await h.settle() # 默认 0.05 秒
await h.settle(0.2) # 复杂 handler 可增大何时增大 settle？ handler 中有
asyncio.sleep()、self.wait_event()或多步对话时。
wait_event — 等待特定事件
event = await h.wait_event(
predicate=lambda e: e.type == "message.group",
timeout=2.0,
)3. Fluent 断言
所有 API 调用都被 Mock 记录为 APICall(action, params: dict)，通过 fluent API 进行语义化断言。
基础断言
# 检查是否被调用
h.assert_api("send_group_msg").called()
# 检查调用次数
h.assert_api("send_group_msg").times(1)
# 检查未被调用
h.assert_api("set_group_kick").not_called()参数匹配
# params 子集匹配
h.assert_api("send_group_msg").with_params(group_id="100")
# 链式组合
h.assert_api("send_group_msg").called().with_params(group_id="100").with_text("hello")文本匹配
with_text() 是跨平台感知的：
- QQ: 从
messagesegments 提取type=text的文本 - Bilibili: 取
text/content字段 - GitHub: 取
body字段
h.assert_api("send_group_msg").with_text("hello")
h.assert_api("send_danmu").with_text("弹幕内容")
h.assert_api("create_issue_comment").with_text("LGTM")自定义条件
h.assert_api("send_group_msg").where(lambda call: len(call.params["message"]) > 1)取值
# 获取最近一次调用
call = h.assert_api("send_group_msg").last
print(call.action, call.params)
# 获取所有匹配调用
calls = h.assert_api("send_group_msg").calls直接访问 MockAPI
fluent API 之外，仍可直接访问 MockAPI 方法：
h.mock_api.called("send_group_msg") # bool
h.mock_api.call_count("send_group_msg") # int
h.mock_api.get_calls("send_group_msg") # List[APICall]
h.mock_api.last_call("send_group_msg") # APICall | None重置调用记录
多步测试中，用 reset_api() 隔离每步的断言：
await h.inject(group_message("step1"))
await h.settle()
h.assert_api("send_group_msg").called()
h.reset_api() # 清空记录
await h.inject(group_message("step2"))
await h.settle()
h.assert_api("send_group_msg").times(1) # 只计 step24. Mock 响应配置
如果 handler 依赖 API 返回值，可预配置 Mock 响应：
h.mock_api.set_response("get_group_member_info", {
"user_id": "99",
"nickname": "测试用户",
"role": "member",
})未配置的 API 调用返回空 {}。
5. 多平台测试
构造多平台 Harness
from ncatbot.testing import TestHarness
from ncatbot.testing.factories import qq, github
async with TestHarness(platforms=["qq", "github"]) as h:
await h.inject(github.issue_opened("Bug report"))
await h.settle()
# 平台作用域断言
h.on("qq").assert_api("send_group_msg").called()
h.on("github").assert_api("create_issue_comment").not_called()平台作用域
h.on(platform) 返回 PlatformScope，限定后续断言到该平台：
scope = h.on("qq")
scope.assert_api("send_group_msg").called()
scope.mock_api.call_count("send_group_msg") # 只统计 QQ 的调用
scope.reset() # 只清空 QQ 的记录指定平台访问 Mock API
qq_mock = h.mock_api_for("qq") # MockBotAPI
gh_mock = h.mock_api_for("github") # MockGitHubAPI6. PluginTestHarness
PluginTestHarness 继承 TestHarness，增加了插件选择性加载和查询能力。
构造参数
async with PluginTestHarness(
plugin_names=["hello_world"], # 要加载的插件名
plugins_dir=Path("examples/common/01_hello_world"), # 插件根目录
platforms=("qq",), # 平台列表（可选）
skip_builtin=True, # 不加载内置插件（默认）
skip_pip=True, # 不安装 pip 依赖（默认）
) as h:
...plugins_dir 是包含插件文件夹的父目录。例如插件在
examples/common/01_hello_world/hello_world/下，则plugins_dir应为examples/common/01_hello_world。
查询已加载的插件
print(h.loaded_plugins) # ["hello_world"]
plugin = h.get_plugin("hello_world")
config = h.plugin_config("hello_world")
data = h.plugin_data("hello_world")热重载
success = await h.reload_plugin("hello_world")
assert success传递依赖
如果目标插件在 manifest.toml 中声明了 [dependencies]，PluginTestHarness 会自动解析并加载传递依赖。
7. 对比表
| 能力 | TestHarness | PluginTestHarness |
|---|---|---|
| 事件注入 | ✓ | ✓ |
| Fluent 断言 | ✓ | ✓ |
| Mock 响应 | ✓ | ✓ |
| 多平台支持 | ✓ | ✓ |
| 选择性加载插件 | ✗ | ✓ |
| 插件状态查询 | ✗ | ✓ |
| 热重载 | ✗ | ✓ |
| skip_builtin / skip_pip | — | ✓ |
插件开发者请始终使用
PluginTestHarness。TestHarness主要用于框架内部测试。
8. 常见模式与陷阱
✅ 多步对话测试
async with PluginTestHarness(...) as h:
await h.inject(group_message("注册", group_id="100", user_id="99"))
await h.settle(0.1)
h.assert_api("send_group_msg").called()
h.reset_api()
await h.inject(group_message("张三", group_id="100", user_id="99"))
await h.settle(0.1)
h.assert_api("send_group_msg").called().with_text("张三")⚠️ settle 时间不足
默认 settle(0.05) 对简单 handler 足够。如果断言失败，先尝试增大 settle：
await h.settle(0.2) # 复杂 handler
await h.settle(0.5) # 含 wait_event 的多步对话⚠️ plugins_dir 路径错误
# ✗ 错误：指向插件本身
PluginTestHarness(plugin_names=["hello"], plugins_dir=Path("plugins/hello/"))
# ✓ 正确：指向插件的父目录
PluginTestHarness(plugin_names=["hello"], plugins_dir=Path("plugins/"))⚠️ 同一测试中测试多个命令
每个命令测试前用 reset_api() 清空记录，避免断言受之前调用的干扰。
⚠️ 未注册平台
注入的事件 platform 必须与 Harness 构造时的 platforms 参数匹配，否则会抛出 ValueError：
# ✗ 错误：Harness 只注册了 qq，但注入 github 事件
async with TestHarness() as h: # 默认 platforms=("qq",)
await h.inject(github.issue_opened("Bug")) # ValueError
# ✓ 正确
async with TestHarness(platforms=["qq", "github"]) as h:
await h.inject(github.issue_opened("Bug"))版权所有
版权归属：MI