Factory + Scenario + Mock
约 2083 字大约 7 分钟
2026-03-19
事件工厂函数、场景构建器、Mock 适配器、Fluent 断言完整 API 参考
QQ 事件工厂
from ncatbot.testing.factories.qq import group_message, private_message, ...所有函数返回经 model_validate 验证的 Pydantic 模型实例。
group_message
group_message(
text: str = "hello",
*,
group_id: str = "100200",
user_id: str = "99999",
nickname: str = "测试用户",
message_id: Optional[str] = None, # 自动递增
self_id: str = "10001",
message: Optional[list] = None, # 自定义消息段
raw_message: Optional[str] = None, # 默认等于 text
sub_type: str = "normal",
**extra: Any,
) -> GroupMessageEventDataprivate_message
private_message(
text: str = "hello",
*,
user_id: str = "99999",
nickname: str = "测试用户",
message_id: Optional[str] = None,
self_id: str = "10001",
message: Optional[list] = None,
raw_message: Optional[str] = None,
sub_type: str = "friend",
**extra: Any,
) -> PrivateMessageEventDatafriend_request
friend_request(
user_id: str = "99999",
comment: str = "请求加好友",
flag: str = "flag_123",
*,
self_id: str = "10001",
**extra: Any,
) -> FriendRequestEventDatagroup_request
group_request(
user_id: str = "99999",
group_id: str = "100200",
comment: str = "请求加群",
flag: str = "flag_456",
sub_type: str = "add",
*,
self_id: str = "10001",
**extra: Any,
) -> GroupRequestEventDataNapCat comment 格式
NapCat 适配器下,event.comment 的实际格式为 问题:xxx\n答案:yyy。直接把 comment 当纯答案使用会导致正则匹配失败。使用 napcat_comment() 构造真实格式的 comment:
napcat_comment
napcat_comment(
answer: str,
question: str = "问题",
) -> str快捷构造 NapCat 格式的 comment 字符串,返回 "问题:{question}\n答案:{answer}"。
示例:
from ncatbot.testing.factories.qq import group_request, napcat_comment
# 模拟 NapCat 下的加群请求
await harness.inject(group_request(
comment=napcat_comment("东南大学 计小软"),
flag="flag_1",
))group_increase
group_increase(
user_id: str = "99999",
group_id: str = "100200",
operator_id: str = "10001",
sub_type: str = "approve",
*,
self_id: str = "10001",
**extra: Any,
) -> GroupIncreaseNoticeEventDatagroup_decrease
group_decrease(
user_id: str = "99999",
group_id: str = "100200",
operator_id: str = "10001",
sub_type: str = "kick",
*,
self_id: str = "10001",
**extra: Any,
) -> GroupDecreaseNoticeEventDatagroup_ban
group_ban(
user_id: str = "99999",
group_id: str = "100200",
operator_id: str = "10001",
duration: int = 600,
sub_type: str = "ban",
*,
self_id: str = "10001",
**extra: Any,
) -> GroupBanNoticeEventDatagroup_msg_emoji_like
group_msg_emoji_like(
user_id: str = "99999",
group_id: str = "100200",
message_id: Optional[str] = None,
likes: Optional[List[Dict[str, Any]]] = None,
is_add: bool = True,
message_seq: Optional[int] = None,
*,
self_id: str = "10001",
**extra: Any,
) -> GroupMsgEmojiLikeNoticeEventDatapoke
poke(
user_id: str = "99999",
target_id: str = "10001",
group_id: str = "100200",
*,
self_id: str = "10001",
**extra: Any,
) -> PokeNotifyEventDatagroup_upload
group_upload(
user_id: str = "99999",
group_id: str = "100200",
*,
file_id: str = "file_001",
file_name: str = "test.txt",
file_size: int = 1024,
busid: int = 0,
self_id: str = "10001",
**extra: Any,
) -> GroupUploadNoticeEventDatagroup_admin
group_admin(
user_id: str = "99999",
group_id: str = "100200",
sub_type: str = "set",
*,
self_id: str = "10001",
**extra: Any,
) -> GroupAdminNoticeEventDatafriend_add
friend_add(
user_id: str = "99999",
*,
self_id: str = "10001",
**extra: Any,
) -> FriendAddNoticeEventDatagroup_recall
group_recall(
user_id: str = "99999",
group_id: str = "100200",
operator_id: str = "10001",
message_id: Optional[str] = None,
*,
self_id: str = "10001",
**extra: Any,
) -> GroupRecallNoticeEventDatafriend_recall
friend_recall(
user_id: str = "99999",
message_id: Optional[str] = None,
*,
self_id: str = "10001",
**extra: Any,
) -> FriendRecallNoticeEventDatalucky_king
lucky_king(
user_id: str = "99999",
target_id: str = "88888",
group_id: str = "100200",
*,
self_id: str = "10001",
**extra: Any,
) -> LuckyKingNotifyEventDatahonor
honor(
user_id: str = "99999",
group_id: str = "100200",
honor_type: str = "talkative",
*,
self_id: str = "10001",
**extra: Any,
) -> HonorNotifyEventDataBilibili 事件工厂
from ncatbot.testing.factories.bilibili import danmu, super_chat, gift, ...danmu
danmu(
text: str = "弹幕内容",
*, room_id: str = "12345", user_id: str = "88888",
user_name: str = "测试弹幕用户", **extra,
) -> DanmuMsgEventDatasuper_chat
super_chat(
content: str = "SC 内容", price: int = 30,
*, room_id: str = "12345", user_id: str = "88888",
user_name: str = "SC 用户", duration: int = 60, **extra,
) -> SuperChatEventDatagift
gift(
gift_name: str = "辣条", num: int = 1,
*, room_id: str = "12345", user_id: str = "88888",
user_name: str = "送礼用户", gift_id: str = "1",
price: int = 100, coin_type: str = "silver", **extra,
) -> GiftEventDataprivate_message (Bilibili)
private_message(
text: str = "私信内容",
*, user_id: str = "88888", user_name: str = "私信用户", **extra,
) -> BiliPrivateMessageEventDatacomment
comment(
content: str = "评论内容",
*, resource_id: str = "BV1234", resource_type: str = "video",
comment_id: str = "c_001", user_id: str = "88888",
user_name: str = "评论用户", **extra,
) -> BiliCommentEventDatadynamic
dynamic(
text: str = "动态内容",
*, uid: str = "88888", user_name: str = "动态用户",
dynamic_id: str = "dyn_001",
dynamic_type: str = "DYNAMIC_TYPE_WORD", **extra,
) -> BiliDynamicEventDataGitHub 事件工厂
from ncatbot.testing.factories.github import issue_opened, pr_opened, push, ...issue_opened
issue_opened(
title: str = "Bug report", body: str = "",
*, repo: str = "owner/repo", issue_number: int = 1,
login: str = "test-user", labels: Optional[List[str]] = None, **extra,
) -> GitHubIssueEventDataissue_closed
issue_closed(
title: str = "Bug report",
*, repo: str = "owner/repo", issue_number: int = 1,
login: str = "test-user", **extra,
) -> GitHubIssueEventDataissue_comment
issue_comment(
body: str = "LGTM",
*, repo: str = "owner/repo", issue_number: int = 1,
comment_id: str = "c_001", login: str = "test-user", **extra,
) -> GitHubIssueCommentEventDatapr_opened
pr_opened(
title: str = "Fix bug", body: str = "",
*, repo: str = "owner/repo", pr_number: int = 1,
head_ref: str = "feature", base_ref: str = "main",
login: str = "test-user", **extra,
) -> GitHubPREventDatapush
push(
ref: str = "refs/heads/main",
*, repo: str = "owner/repo", login: str = "test-user",
before: str = "0" * 40, after: str = "a" * 40, **extra,
) -> GitHubPushEventDatastar
star(
*, repo: str = "owner/repo", login: str = "test-user",
action: str = "created", **extra,
) -> GitHubStarEventDatarelease_published
release_published(
tag_name: str = "v1.0.0", name: str = "Release v1.0.0", body: str = "",
*, repo: str = "owner/repo", login: str = "test-user", **extra,
) -> GitHubReleaseEventDataScenario
from ncatbot.testing import Scenario链式测试场景构建器。所有链式方法返回 self。
构造
Scenario(name: str = "") -> Scenario| 参数 | 类型 | 说明 |
|---|---|---|
name | str | 场景名称(出现在失败消息中) |
链式方法
| 方法 | 签名 | 说明 |
|---|---|---|
on | on(platform: str) -> Scenario | 切换后续断言步骤的平台作用域 |
inject | inject(event_data: BaseEventData) -> Scenario | 注入事件步骤 |
inject_many | inject_many(events: List[BaseEventData]) -> Scenario | 注入多事件步骤 |
settle | settle(delay: float = 0.05) -> Scenario | 等待步骤 |
assert_api_called | assert_api_called(action: str, **match) -> Scenario | API 调用断言(可选 params 子集匹配) |
assert_api_not_called | assert_api_not_called(action: str) -> Scenario | API 未调用断言 |
assert_api_count | assert_api_count(action: str, count: int) -> Scenario | 调用次数断言 |
assert_api_params | assert_api_params(action: str, **params) -> Scenario | params 子集匹配断言 |
assert_api_text | assert_api_text(action: str, *fragments) -> Scenario | 文本内容断言(跨平台感知) |
assert_api_where | assert_api_where(action: str, predicate, desc="") -> Scenario | 自定义条件断言 |
assert_that | assert_that(predicate: Callable[[TestHarness], None], desc="") -> Scenario | 自定义断言 |
reset_api | reset_api() -> Scenario | 清空调用记录 |
执行
async run(harness: TestHarness) -> None按顺序执行所有步骤。失败时抛出 AssertionError,包含场景名和步骤编号。
示例
from ncatbot.testing import Scenario
from ncatbot.testing.factories.qq import group_message
await (
Scenario("注册流程")
.inject(group_message("注册", group_id="100", user_id="99"))
.settle()
.assert_api_called("send_group_msg")
.assert_api_text("send_group_msg", "请输入")
.reset_api()
.inject(group_message("张三", group_id="100", user_id="99"))
.settle()
.assert_api_params("send_group_msg", group_id="100")
.run(harness)
)APICall
from ncatbot.adapter.mock.api_base import APICall@dataclass
class APICall:
action: str # API 方法名(如 "send_group_msg")
params: Dict[str, Any] = field(...) # 所有参数按名存储所有参数以关键字形式统一存入
params字典。无args/kwargs/timestamp字段。
MockAPIBase
from ncatbot.adapter.mock import MockAPIBase所有平台 Mock API 的共享基类。MockBotAPI、MockBiliAPI、MockGitHubAPI 均继承此类。
调用记录方法
| 方法 | 签名 | 说明 |
|---|---|---|
calls | @property -> List[APICall] | 所有调用记录 |
called | called(action: str) -> bool | 是否被调用 |
call_count | call_count(action: str) -> int | 调用次数 |
get_calls | get_calls(action: str) -> List[APICall] | 获取匹配调用 |
last_call | last_call(action: str = None) -> Optional[APICall] | 最近一次调用 |
reset | reset() -> None | 清空记录 |
响应配置
set_response(action: str, response: Any) -> None未配置的 action 返回 {}。
MockBotAPI
from ncatbot.adapter.mock import MockBotAPI继承 MockAPIBase + IQQAPIClient。所有 QQ API 方法以 self._record("action", key=val, ...) 形式录制。
已实现的 API 方法
| 分类 | action 名称 |
|---|---|
| 消息 | send_private_msg, send_group_msg, delete_msg, send_forward_msg |
| 群管理 | set_group_kick, set_group_ban, set_group_whole_ban, set_group_admin, set_group_card, set_group_name, set_group_leave, set_group_special_title |
| 请求 | set_friend_add_request, set_group_add_request |
| 查询 | get_login_info, get_stranger_info, get_friend_list, get_group_info, get_group_list, get_group_member_info, get_group_member_list, get_msg, get_forward_msg |
| 文件 | upload_group_file, get_group_root_files, get_group_file_url, delete_group_file |
| 互动 | send_like, send_poke |
MockBiliAPI
from ncatbot.adapter.mock import MockBiliAPI继承 MockAPIBase + IBiliAPIClient。Bilibili 直播/视频相关 API mock。
MockGitHubAPI
from ncatbot.adapter.mock import MockGitHubAPI继承 MockAPIBase + IGitHubAPIClient。GitHub Issue/PR/Release 相关 API mock。
APICallAssertion
from ncatbot.testing import APICallAssertionFluent 断言 API,通过 h.assert_api(action) 或 h.on(platform).assert_api(action) 获取。
方法
| 方法 | 返回 | 说明 |
|---|---|---|
called() | self | 断言至少被调用一次 |
not_called() | self | 断言未被调用 |
times(n) | self | 断言调用次数 |
with_params(**expected) | self | 断言任一调用的 params 包含 expected 子集 |
with_text(*fragments) | self | 断言任一调用的文本包含所有 fragment(跨平台感知) |
where(predicate) | self | 断言任一调用满足自定义条件 |
last | APICall | 最后一次匹配调用(property) |
calls | List[APICall] | 所有匹配调用(property) |
链式用法
h.assert_api("send_group_msg").called().with_params(group_id="100").with_text("hello")PlatformScope
from ncatbot.testing import PlatformScope限定到单个平台的断言作用域,通过 h.on(platform) 获取。
| 方法/属性 | 签名 | 说明 |
|---|---|---|
assert_api | assert_api(action: str) -> APICallAssertion | 该平台范围内的断言 |
mock_api | @property -> MockAPIBase | 该平台的 Mock API |
reset | reset() -> None | 清空该平台调用记录 |
extract_text
from ncatbot.testing import extract_textdef extract_text(call: APICall) -> str从 APICall 中提取文本内容(跨平台感知):
- QQ
message(list[segment]) → 拼接type=text的data.text - Bilibili
text/content→ 直接取字符串 - GitHub
body→ 直接取字符串 - 兜底 →
str(params)
MockAdapter
from ncatbot.adapter import MockAdapterBaseAdapter 的内存实现,无网络通信。根据 platform 参数自动选择对应的 Mock API 子类。
| 属性/方法 | 签名 | 说明 |
|---|---|---|
mock_api | @property -> MockAPIBase | Mock API 实例 |
connected | @property -> bool | 是否已连接 |
inject_event | async inject_event(data: BaseEventData) -> None | 注入事件到 dispatcher |
stop | stop() -> None | 停止 listen 循环 |
get_api | get_api() -> IAPIClient | 返回 Mock API |
通常通过
TestHarness.adapter_for(platform)或TestHarness.mock_api_for(platform)访问。
插件发现
discover_testable_plugins
discover_testable_plugins(plugins_dir: Path) -> List[PluginManifest]扫描目录下所有含 manifest.toml 的子文件夹,返回解析后的 PluginManifest 列表。
generate_smoke_tests
generate_smoke_tests(manifests: List[PluginManifest]) -> str批量生成冒烟测试代码(完整 pytest 文件)。
generate_smoke_test
generate_smoke_test(manifest: PluginManifest) -> str为单个插件生成冒烟测试代码片段。
conftest_plugin.py — pytest 插件
CLI 选项
| 选项 | 说明 |
|---|---|
--plugin-dir=PATH | 插件根目录路径 |
Markers
| Marker | 说明 |
|---|---|
@pytest.mark.plugin(name="xxx") | 标记特定插件测试 |
@pytest.mark.plugin_names(names) | 指定要加载的插件名列表 |
@pytest.mark.plugins_dir(dir) | 指定插件目录 |
Fixtures
| Fixture | 说明 |
|---|---|
plugins_dir | 从 --plugin-dir 获取的 Path |
辅助函数
get_testable_plugin_names(plugins_dir: str) -> List[str]返回目录下所有可测试插件名。
相关文档
| 文档 | 说明 |
|---|---|
| TestHarness + PluginTestHarness | 编排器 API |
| 测试指南:工厂与场景 | 教程风格的用法说明 |
版权所有
版权归属:MI
