TestHarness + PluginTestHarness
约 732 字大约 2 分钟
2026-03-19
测试编排器完整 API 参考
TestHarness
from ncatbot.testing import TestHarness测试编排器,在后台启动 BotClient(使用 MockAdapter),提供事件注入和断言 API。
构造
TestHarness() -> TestHarness无参数。内部创建 MockAdapter + BotClient。
Properties
| 属性 | 类型 | 说明 |
|---|---|---|
bot | BotClient | Bot 客户端实例 |
adapter | MockAdapter | Mock 适配器实例 |
mock_api | MockBotAPI | Mock API 实例(用于 set_response) |
dispatcher | AsyncEventDispatcher | 事件分发器 |
api_calls | List[APICall] | 所有 API 调用记录 |
生命周期方法
| 方法 | 签名 | 说明 |
|---|---|---|
start | async start() -> None | 启动 BotClient(非阻塞) |
stop | async stop() -> None | 停止 BotClient |
__aenter__ | async __aenter__() -> TestHarness | async with 入口 |
__aexit__ | async __aexit__(...) -> None | async with 出口 |
事件注入方法
| 方法 | 签名 | 说明 |
|---|---|---|
inject | async inject(event_data: BaseEventData) -> None | 注入单个事件 |
inject_many | async inject_many(events: List[BaseEventData]) -> None | 依次注入多个事件 |
settle | async settle(delay: float = 0.05) -> None | 等待 handler 执行 |
wait_event | async wait_event(predicate: Callable[[Event], bool] = None, timeout: float = 2.0) -> Event | 等待特定事件 |
API 断言方法
| 方法 | 签名 | 说明 |
|---|---|---|
api_called | api_called(action: str) -> bool | 检查 action 是否被调用 |
api_call_count | api_call_count(action: str) -> int | 获取调用次数 |
get_api_calls | get_api_calls(action: str) -> List[APICall] | 获取所有匹配调用 |
last_api_call | last_api_call(action: str = None) -> Optional[APICall] | 获取最近一次调用 |
reset_api | reset_api() -> None | 清空调用记录 |
使用示例
async with TestHarness() as h:
await h.inject(group_message("hello"))
await h.settle()
assert h.api_called("send_group_msg")
call = h.last_api_call("send_group_msg")
print(call.action, call.args, call.kwargs)PluginTestHarness
from ncatbot.testing import PluginTestHarness继承 TestHarness,增加插件选择性加载和查询能力。
构造
PluginTestHarness(
plugin_names: List[str], # 要加载的插件名列表
plugin_dir: Path, # 插件根目录
*,
skip_builtin: bool = True, # 是否跳过内置插件
skip_pip: bool = True, # 是否跳过 pip 依赖安装
) -> PluginTestHarness| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
plugin_names | List[str] | 必填 | 要加载的插件名(manifest.toml 中的 name) |
plugin_dir | Path | 必填 | 包含插件文件夹的父目录 |
skip_builtin | bool | True | 跳过内置插件加载 |
skip_pip | bool | True | 跳过 pip 依赖安装 |
额外 Properties
| 属性 | 类型 | 说明 |
|---|---|---|
loaded_plugins | List[str] | 已加载的插件名列表 |
继承自 TestHarness 的所有 properties 和方法同样可用。
插件操作方法
| 方法 | 签名 | 说明 |
|---|---|---|
get_plugin | get_plugin(name: str) -> Optional[BasePlugin] | 获取插件实例 |
plugin_config | plugin_config(name: str) -> dict | 获取插件配置 |
plugin_data | plugin_data(name: str) -> dict | 获取插件数据 |
reload_plugin | async reload_plugin(name: str) -> bool | 热重载插件 |
plugin_config和plugin_data在插件未加载时抛出KeyError。
start 流程
PluginTestHarness.start() 执行以下步骤:
- 启动核心基础设施(adapter / API / dispatcher / services)
- 配置 PluginLoader 依赖注入
- 如果
skip_builtin=False,加载内置插件 - 通过
loader.load_selected()加载指定插件及其传递依赖 - 启动后台监听任务
使用示例
async with PluginTestHarness(
plugin_names=["hello_world"],
plugin_dir=Path("examples/qq/01_hello_world"),
) as h:
# 验证加载
assert "hello_world" in h.loaded_plugins
# 注入事件
await h.inject(group_message("hello", group_id="100", user_id="99"))
await h.settle()
# 断言
assert h.api_called("send_group_msg")
# 查询插件状态
plugin = h.get_plugin("hello_world")
config = h.plugin_config("hello_world")相关文档
| 文档 | 说明 |
|---|---|
| Factory + Scenario + Mock | 事件工厂、场景构建器、Mock API |
| 测试指南 | 教程风格入门 |
| Harness 使用详解 | 实践指南 |
版权所有
版权归属:huan-yp
