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,
) -> GroupMessageEventData

private_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,
) -> PrivateMessageEventData

friend_request

friend_request(
    user_id: str = "99999",
    comment: str = "请求加好友",
    flag: str = "flag_123",
    *,
    self_id: str = "10001",
    **extra: Any,
) -> FriendRequestEventData

group_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,
) -> GroupRequestEventData

NapCat 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,
) -> GroupIncreaseNoticeEventData

group_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,
) -> GroupDecreaseNoticeEventData

group_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,
) -> GroupBanNoticeEventData

group_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,
) -> GroupMsgEmojiLikeNoticeEventData

poke

poke(
    user_id: str = "99999",
    target_id: str = "10001",
    group_id: str = "100200",
    *,
    self_id: str = "10001",
    **extra: Any,
) -> PokeNotifyEventData

group_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,
) -> GroupUploadNoticeEventData

group_admin

group_admin(
    user_id: str = "99999",
    group_id: str = "100200",
    sub_type: str = "set",
    *,
    self_id: str = "10001",
    **extra: Any,
) -> GroupAdminNoticeEventData

friend_add

friend_add(
    user_id: str = "99999",
    *,
    self_id: str = "10001",
    **extra: Any,
) -> FriendAddNoticeEventData

group_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,
) -> GroupRecallNoticeEventData

friend_recall

friend_recall(
    user_id: str = "99999",
    message_id: Optional[str] = None,
    *,
    self_id: str = "10001",
    **extra: Any,
) -> FriendRecallNoticeEventData

lucky_king

lucky_king(
    user_id: str = "99999",
    target_id: str = "88888",
    group_id: str = "100200",
    *,
    self_id: str = "10001",
    **extra: Any,
) -> LuckyKingNotifyEventData

honor

honor(
    user_id: str = "99999",
    group_id: str = "100200",
    honor_type: str = "talkative",
    *,
    self_id: str = "10001",
    **extra: Any,
) -> HonorNotifyEventData

Bilibili 事件工厂

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,
) -> DanmuMsgEventData

super_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,
) -> SuperChatEventData

gift

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,
) -> GiftEventData

private_message (Bilibili)

private_message(
    text: str = "私信内容",
    *, user_id: str = "88888", user_name: str = "私信用户", **extra,
) -> BiliPrivateMessageEventData

comment

comment(
    content: str = "评论内容",
    *, resource_id: str = "BV1234", resource_type: str = "video",
    comment_id: str = "c_001", user_id: str = "88888",
    user_name: str = "评论用户", **extra,
) -> BiliCommentEventData

dynamic

dynamic(
    text: str = "动态内容",
    *, uid: str = "88888", user_name: str = "动态用户",
    dynamic_id: str = "dyn_001",
    dynamic_type: str = "DYNAMIC_TYPE_WORD", **extra,
) -> BiliDynamicEventData

GitHub 事件工厂

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,
) -> GitHubIssueEventData

issue_closed

issue_closed(
    title: str = "Bug report",
    *, repo: str = "owner/repo", issue_number: int = 1,
    login: str = "test-user", **extra,
) -> GitHubIssueEventData

issue_comment

issue_comment(
    body: str = "LGTM",
    *, repo: str = "owner/repo", issue_number: int = 1,
    comment_id: str = "c_001", login: str = "test-user", **extra,
) -> GitHubIssueCommentEventData

pr_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,
) -> GitHubPREventData

push

push(
    ref: str = "refs/heads/main",
    *, repo: str = "owner/repo", login: str = "test-user",
    before: str = "0" * 40, after: str = "a" * 40, **extra,
) -> GitHubPushEventData

star

star(
    *, repo: str = "owner/repo", login: str = "test-user",
    action: str = "created", **extra,
) -> GitHubStarEventData

release_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,
) -> GitHubReleaseEventData

Scenario

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 APICallAssertion

Fluent 断言 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_text

def 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 MockAdapter

BaseAdapter 的内存实现，无网络通信。根据 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

许可证：署名 4.0 国际 (CC-BY-4.0)