qihuaiflow/tests/astrbot-channel-optimization-test-case-report.md

测试用例 — AstrBot频道优化（P0-P2 全部完成）

测试日期: 2026-06-01
测试工程师: 测试工程师
测试依据: 项目负责人指定的AstrBot频道优化需求 + 前后端代码分析
需求范围: 统一Webhook端点(6平台)、消息组件系统(5种组件)、平台注册机制(6个适配器)、MessageSession统一标识、Pipeline管道(5 Stage)、Redis事件总线、二维码localhost根因修复(config.py三级防护)

---

1. 需求依据

• PRD 路径: /data/share/project/docs/PRD.md
• 本轮优化范围（项目负责人指定）:
  1. 统一Webhook端点 — 为6个IM平台（企业微信/飞书/钉钉/Telegram/Slack/WhatsApp）提供统一的 /webhook/{platform} 入口
  2. 消息组件系统 — 5种消息组件（Plain/Image/At/Reply/File），支持序列化/反序列化
  3. 平台注册机制 — 使用 @register_platform_adapter 装饰器标准化注册6个适配器
  4. MessageSession统一会话标识 — platform:type:session_id 格式
  5. Pipeline管道 — 5 Stage 链式处理（WakingCheck→RateLimit→ContentSafety→Process→Respond）
  6. Redis事件总线 — 基于Redis List的生产者-消费者消息队列
  7. 二维码localhost根因修复 — config.py 三级防护（环境变量→.env→硬编码域名）

---

2. 代码功能模块概览

模块	职责	代码位置
Config配置模块	APP_ACCESS_URL三级防护，全局配置管理	app/config.py:36-63
Pipeline管道系统	5 Stage链式消息处理，PipelineScheduler调度器	app/pipeline.py (349行)
Redis事件总线	MessageEvent序列化，生产者-消费者模式	app/event_bus.py (293行)
消息组件系统	5种组件(Plain/Image/At/Reply/File)，序列化工厂	app/message_components.py (340行)
平台注册机制	@register_platform_adapter装饰器，全局注册表	app/platform_adapter.py (126行)
6个平台适配器	wecom/feishu/dingtalk/telegram/slack/whatsapp	app/platform_adapters/*.py
MessageSession	统一会话标识 platform:type:session_id	app/message_session.py (55行)
统一Webhook端点	POST /webhook/{platform}，适配器+EventBus集成	app/api/channels.py:1453-1579
频道QR二维码	微信ClawBot API优先，本地qrcode库回退	app/api/channels.py:420-498
Webhook平台列表	GET /webhook/platforms 列出所有支持平台	app/api/channels.py:1582-1597

---

3. 需求与代码差异说明

3.1 PRD中未在代码中体现的内容

无。本轮优化的7个功能点在代码中均已有完整实现。

3.2 PRD中未覆盖但代码中存在的功能模块

编号	模块名称	代码位置	建议
N1	旧版webhook处理函数（_handle_*_webhook）	app/api/channels.py:1600-1918	与适配器模式重复，本次已统一到适配器架构
N2	Pipeline单元测试（自测代码）	app/pipeline.py:202-347	内建9个单元测试，建议产品确认纳入CI
N3	EventBus单元测试（自测代码）	app/event_bus.py:182-292	内建7个单元测试，建议产品确认纳入CI
N4	频道消息组件与webhook解析的桥接	跨文件	适配器 parse_webhook() 返回dict而非BaseMessageComponent链，存在集成断点

---

4. 测试用例

4.1 统一Webhook端点 (Q1)

TC-WH-01 [P0] 验证统一Webhook端点存在且路径正确

• 前置条件: 后端服务运行在18080端口
• 步骤: [1] 向 POST /app/api/channels/webhook/{platform} 各平台发送请求 [2] 验证返回404之外的HTTP状态码（400表示平台不支持但端点存在） [3] 逐个平台验证：wecom/feishu/dingtalk/telegram/slack/whatsapp
• 预期结果: 所有6个平台的webhook端点均可访问（返回400"不支持的平台"或正常响应）
• 备注: 请求体为空时，FastAPI可能返回422（Body缺失），这也是端点存在的证明

TC-WH-02 [P0] 验证Webhook端点消息接收和记录

• 前置条件: 已有活跃的微信频道
• 步骤: [1] 构造合法的微信平台webhook消息payload [2] 发送POST到 /app/api/channels/webhook/wecom [3] 验证返回 code: 0 且 data.handled=true [4] 验证返回包含 message_id、session 和 platform 字段 [5] 检查数据库中 channel_messages 表存在对应记录
• 预期结果: 消息成功接收并写入数据库；返回标准webhook响应格式
• 备注: 验证从webhook入站到数据库记录的完整链路

TC-WH-03 [P0] 验证Webhook平台验证（飞书challenge/ Slack url_verification）

• 前置条件: 后端服务运行
• 步骤: [1] 向 /webhook/feishu 发送 {"challenge": "test_challenge_123"} 的POST请求 [2] 验证返回 {"challenge": "test_challenge_123"} [3] 向 /webhook/slack 发送 {"type": "url_verification", "challenge": "slack_challenge_456"} [4] 验证返回 {"challenge": "slack_challenge_456"}
• 预期结果: 飞书和Slack的URL验证正确返回challenge值
• 备注: URL验证是各平台接入的第一步，必须正确处理

TC-WH-04 [P1] 验证Webhook对未注册平台的拒绝

• 前置条件: 后端服务运行
• 步骤: [1] 向 /webhook/unknown_platform 发送POST请求 [2] 验证返回400状态码 [3] 验证错误消息包含"不支持的平台"和可用平台列表
• 预期结果: 未注册平台被正确拒绝，返回明确错误

TC-WH-05 [P1] 验证无活跃频道时Webhook行为

• 前置条件: 所有wecom频道已断开或删除
• 步骤: [1] 向 /webhook/wecom 发送消息payload [2] 验证返回 code: 0 且 data.handled=false [3] 验证消息包含"无活跃的wecom频道"
• 预期结果: 优雅处理无活跃频道的情况，不报500错误

TC-WH-06 [P0] 验证Webhook支持平台列表接口

• 前置条件: 6个适配器已注册
• 步骤: [1] 发送GET请求到 /app/api/channels/webhook/platforms [2] 验证返回 code: 0 [3] 验证 data.platforms 包含6个平台条目 [4] 验证每个平台条目包含name/display_name/description/support能力字段
• 预期结果: 返回完整的6个平台元数据

---

4.2 消息组件系统 (Q2)

TC-MC-01 [P0] 验证Plain文本组件序列化/反序列化

• 前置条件: 消息组件系统代码可导入
• 步骤: [1] 创建 Plain(text="你好世界") [2] 调用 to_dict() 序列化 [3] 验证返回 {"type": "plain", "text": "你好世界"} [4] 调用 Plain.from_dict(serialized) 反序列化 [5] 验证 plain_obj.text == "你好世界"
• 预期结果: Plain组件序列化/反序列化正确

TC-MC-02 [P0] 验证Image图片组件序列化/反序列化

• 前置条件: 同上
• 步骤: [1] 创建 Image(url="https://example.com/img.png", width=800, height=600) [2] 调用 to_dict() 验证url/width/height字段 [3] 调用 Image.from_dict() 反序列化验证
• 预期结果: Image组件序列化/反序列化正确

TC-MC-03 [P0] 验证At提及组件序列化/反序列化

• 前置条件: 同上
• 步骤: [1] 创建 At(user_id="user_001", display_name="张三") [2] 序列化验证user_id和display_name [3] 反序列化验证 [4] 测试不带display_name的场景
• 预期结果: At组件正确处理user_id和可选的display_name

TC-MC-04 [P0] 验证Reply回复组件序列化/反序列化

• 前置条件: 同上
• 步骤: [1] 创建 Reply(message_id="msg_001", content_preview="原消息预览") [2] 序列化验证message_id和content_preview [3] 反序列化验证
• 预期结果: Reply组件序列化/反序列化正确

TC-MC-05 [P0] 验证File文件组件序列化/反序列化

• 前置条件: 同上
• 步骤: [1] 创建 File(filename="report.pdf", url="https://example.com/report.pdf", size=102400) [2] 序列化验证filename/url/size [3] 验证不带size时的序列化
• 预期结果: File组件序列化/反序列化正确

TC-MC-06 [P0] 验证消息链整体序列化/反序列化 (message_chain_to_dict/from_dict)

• 前置条件: 同上
• 步骤: [1] 构建复合消息链: [Reply(...), Plain(...), At(...), Image(...)] [2] 调用 message_chain_to_dict(chain) 生成字典列表 [3] 调用 message_chain_from_dict(dict_list) 恢复组件链 [4] 验证每个组件的类型和关键字段正确
• 预期结果: 完整消息链循环序列化正确

TC-MC-07 [P1] 验证辅助提取函数

• 前置条件: 同上
• 步骤: [1] 构建包含多个组件的消息链 [2] 测试 extract_plain_text() 正确拼接所有Plain文本 [3] 测试 extract_image_urls() 提取所有图片URL [4] 测试 extract_at_user_ids() 提取所有@用户ID [5] 测试 has_component_type() 对存在和不存在的类型
• 预期结果: 所有辅助提取函数正确

TC-MC-08 [P1] 验证未知组件类型降级处理

• 前置条件: 同上
• 步骤: [1] 构造 [{"type": "unknown_type", "data": "test"}] 传入 message_chain_from_dict [2] 验证返回结果为Plain组件，text为原始dict的字符串表示
• 预期结果: 未知类型优雅降级，不抛异常

---

4.3 平台注册机制 (Q3)

TC-PR-01 [P0] 验证6个适配器均已正确注册

• 前置条件: 6个适配器文件已导入
• 步骤: [1] 从 platform_adapter 导入 platform_registry 和 platform_cls_map [2] 验证 len(platform_registry) == 6 [3] 验证注册表包含所有6个平台: {"wecom", "feishu", "dingtalk", "telegram", "slack", "whatsapp"} [4] 验证 platform_cls_map 中的每个类都是 BasePlatformAdapter 的子类
• 预期结果: 6个适配器全部正确注册，name和class映射完整

TC-PR-02 [P0] 验证平台适配器工厂函数

• 前置条件: 适配器已注册
• 步骤: [1] 对每个平台调用 get_platform_adapter(name, config={}) [2] 验证返回的实例是其对应适配器类的实例 [3] 验证每个适配器 get_status() 返回 "stopped" [4] 调用 start() 后验证 get_status() 返回 "running"
• 预期结果: 工厂函数正确创建适配器实例，状态管理正常

TC-PR-03 [P0] 验证平台元数据完整性

• 前置条件: 适配器已注册
• 步骤: [1] 对每个平台获取 platform_registry[name] [2] 验证元数据包含: name/display_name/description/icon/version [3] 验证能力标记: support_streaming/support_image/support_file/support_voice [4] 验证不同平台的能力标记符合实际情况: 飞书streaming=true, 企业微信streaming=false等
• 预期结果: 所有平台元数据字段完整正确

TC-PR-04 [P0] 验证list_platforms API

• 前置条件: 适配器已注册
• 步骤: [1] 调用 list_platforms() [2] 验证返回6个PlatformMetadata实例 [3] 验证每个实例的name字段唯一且非空
• 预期结果: 列出所有已注册平台

TC-PR-05 [P1] 验证未注册平台的错误处理

• 前置条件: 适配器已注册
• 步骤: [1] 调用 get_platform_adapter("unknown_platform", {}) [2] 验证抛出 ValueError 异常 [3] 验证异常消息包含"未知平台"和可用平台列表
• 预期结果: 未注册平台被正确拒绝，有清晰错误提示

TC-PR-06 [P1] 验证装饰器中非BasePlatformAdapter子类的拒绝

• 前置条件: 适配器代码
• 步骤: [1] 确认代码中 register_platform_adapter 装饰器的类型检查逻辑存在 [2] 验证 issubclass(cls, BasePlatformAdapter) 检查在第93行
• 预期结果: 装饰器强制类型约束，非适配器子类会抛出TypeError

---

4.4 MessageSession统一会话标识 (Q4)

TC-MS-01 [P0] 验证MessageSession创建和字符串化

• 前置条件: message_session模块可用
• 步骤: [1] 创建 MessageSession(platform_name="wecom", session_type=SessionType.GROUP, session_id="group_123") [2] 验证 unified_msg_origin 返回 "wecom:group:group_123" [3] 创建 MessageSession(platform_name="feishu", session_type=SessionType.FRIEND, session_id="user_456") [4] 验证 unified_msg_origin 返回 "feishu:friend:user_456"
• 预期结果: 统一会话标识格式化为 platform:type:session_id

TC-MS-02 [P0] 验证from_str解析

• 前置条件: 同上
• 步骤: [1] 调用 MessageSession.from_str("telegram:friend:user_789") [2] 验证 platform_name="telegram", session_type=SessionType.FRIEND, session_id="user_789" [3] 调用 MessageSession.from_str("slack:channel:ch_001") [4] 验证 session_type=SessionType.CHANNEL
• 预期结果: 正确从字符串解析，所有SessionType枚举值均支持

TC-MS-03 [P0] 验证from_channel_event自动推断

• 前置条件: 同上
• 步骤: [1] 有group_id时: from_channel_event("wecom", "ch_001", "user_001", group_id="grp_001") [2] 验证返回GROUP类型，session_id="grp_001" [3] 无group_id时: from_channel_event("feishu", "ch_002", "user_002") [4] 验证返回FRIEND类型，session_id="user_002"
• 预期结果: 自动根据group_id判断会话类型

TC-MS-04 [P1] 验证hashable/equality行为

• 前置条件: 同上
• 步骤: [1] 创建 ms1 = MessageSession("wecom", SessionType.FRIEND, "user_001") [2] 创建 ms2 = MessageSession("wecom", SessionType.FRIEND, "user_001") [3] 验证 ms1 == ms2 为True [4] 验证 hash(ms1) == hash(ms2) [5] 验证可以放入set去重: len({ms1, ms2}) == 1
• 预期结果: MessageSession支持相等比较和哈希，可用作集合元素

TC-MS-05 [P1] 验证from_str格式错误处理

• 前置条件: 同上
• 步骤: [1] 调用 MessageSession.from_str("invalid_format") [2] 验证抛出 ValueError 异常 [3] 验证异常消息包含正确的格式说明
• 预期结果: 格式错误的字符串被正确拒绝

---

4.5 Pipeline管道系统 (Q5)

TC-PL-01 [P0] 验证Pipeline默认5阶段按顺序执行

• 前置条件: pipeline模块可用
• 步骤: [1] 创建 PipelineScheduler() 默认管道 [2] 构造完整上下文（含message/platform/sender_id/is_private等） [3] 调用 pipeline.execute(context) [4] 验证返回 success=True [5] 验证 pipeline_log 包含5条记录，顺序为: waking_check → rate_limit_check → content_safety_check → process → respond
• 预期结果: 5个阶段按正确顺序执行，pipeline_log记录完整

TC-PL-02 [P0] 验证WakingCheck阶段：私聊自动唤醒

• 前置条件: 同上
• 步骤: [1] 创建私聊场景: is_private=True [2] 执行WakingCheckStage.process() [3] 验证 should_respond=True, stop=False
• 预期结果: 私聊消息自动唤醒AI响应

TC-PL-03 [P0] 验证WakingCheck阶段：群聊@bot唤醒

• 前置条件: 同上
• 步骤: [1] 创建群聊场景: is_private=False, message="@bot 帮我查询" [2] 执行WakingCheckStage.process() [3] 验证 should_respond=True, stop=False
• 预期结果: 群聊中@bot后唤醒AI

TC-PL-04 [P0] 验证WakingCheck阶段：群聊不@bot则停止

• 前置条件: 同上
• 步骤: [1] 创建群聊场景: is_private=False, message="今天天气不错" [2] 执行WakingCheckStage.process() [3] 验证 should_respond=False, stop=True
• 预期结果: 群聊中未@bot时停止管道

TC-PL-05 [P0] 验证RateLimit阶段：限流生效

• 前置条件: 同上
• 步骤: [1] 设置 RateLimitCheckStage(max_per_minute=3) [2] 同一sender_id连续发送4次 [3] 验证前3次返回success=True [4] 验证第4次返回success=False, error含"频繁", stop=True
• 预期结果: 超过限制后被正确限流

TC-PL-06 [P1] 验证ContentSafety阶段：正常内容通过

• 前置条件: 同上
• 步骤: [1] 发送内容为"正常的聊天内容" [2] 验证success=True, stop=False
• 预期结果: 正常内容不受影响

TC-PL-07 [P1] 验证Pipeline阶段异常处理

• 前置条件: 同上
• 步骤: [1] 创建抛出异常的BrokenStage [2] 执行PipelineScheduler(stages=[BrokenStage()]) [3] 验证返回success=False [4] 验证error消息包含阶段名和异常信息
• 预期结果: stage异常被正确捕获，返回结构化错误

TC-PL-08 [P1] 验证Pipeline阶段动态增删

• 前置条件: 同上
• 步骤: [1] 创建空stages的管道 [2] 调用 add_stage() 添加WakingCheckStage [3] 调用 add_stage() 插入到位置0 [4] 验证阶段顺序正确 [5] 调用 remove_stage("waking_check") 移除 [6] 验证移除后管道阶段列表正确
• 预期结果: 动态增删阶段正常

---

4.6 Redis事件总线 (Q6)

TC-EB-01 [P0] 验证MessageEvent创建和序列化

• 前置条件: Redis可用
• 步骤: [1] 创建MessageEvent(platform="feishu", channel_id="ch_001", content="测试", sender_id="user_001") [2] 调用 to_dict() 序列化 [3] 验证返回字典包含所有9个字段 [4] 验证timestamp字段存在且为整数
• 预期结果: MessageEvent序列化包含完整字段，含自动生成的timestamp

TC-EB-02 [P0] 验证MessageEvent反序列化

• 前置条件: 同上
• 步骤: [1] 从to_dict结果调用 MessageEvent.from_dict(dict_data) [2] 验证所有字段完全恢复 [3] 特别验证platform/channel_id/content/sender_id正确
• 预期结果: 序列化→反序列化循环后数据一致

TC-EB-03 [P0] 验证RedisEventBus事件发布

• 前置条件: Redis可用
• 步骤: [1] 创建RedisEventBus实例 [2] 创建MessageEvent并调用 bus.publish(event) [3] 验证返回 True（发布成功） [4] 调用 bus.queue_size() 验证队列长度 ≥1
• 预期结果: 事件成功写入Redis List

TC-EB-04 [P0] 验证RedisEventBus事件消费

• 前置条件: Redis可用
• 步骤: [1] 创建RedisEventBus并启动1个消费者 [2] 注册feishu平台的handler [3] 发布一个feishu平台事件 [4] 等待1秒后验证handler被调用 [5] 验证handler接收到的事件字段正确 [6] 调用 bus.stop() 停止消费者
• 预期结果: 事件被正确消费并分发给对应平台handler

TC-EB-05 [P0] 验证全局单例

• 前置条件: 同上
• 步骤: [1] 调用 get_event_bus() 两次 [2] 验证两个引用指向同一个实例（bus1 is bus2）
• 预期结果: 全局单例正确

TC-EB-06 [P1] 验证订阅通配符 *

• 前置条件: Redis可用
• 步骤: [1] 注册 * 通配符handler [2] 发布不同平台（feishu/wecom）的事件 [3] 验证通配符handler都能接收到 [4] 清理
• 预期结果: 通配符订阅接收所有平台事件

TC-EB-07 [P1] 验证多消费者并发

• 前置条件: Redis可用
• 步骤: [1] 启动2个消费者线程 [2] 发布多个事件 [3] 验证所有事件被消费，不重复不丢失 [4] 停止消费者
• 预期结果: 多消费者并发正常工作

TC-EB-08 [P1] 验证Redis队列清理

• 前置条件: Redis可用
• 步骤: [1] 清理测试key: DEL aiflow:event_queue [2] 验证 queue_size() == 0
• 预期结果: 队列清理后为空

---

4.7 二维码URL非localhost (Q7)

TC-QR-01 [P0] 验证config.py三级防护机制存在

• 前置条件: config.py可读取
• 步骤: [1] 检查 APP_ACCESS_URL 赋值逻辑 [2] 验证第40-41行：首选环境变量 os.getenv("APP_ACCESS_URL") [3] 验证第43-58行：临时域名检测（team-coding/localhost/127.0.0.1/0.0.0.0），检测到后从.env覆盖 [4] 验证第61-63行：最终兜底为 https://bot.qihuyun.com
• 预期结果: 三级防护完整：环境变量→.env→硬编码正式域名

TC-QR-02 [P0] 验证二维码生成API返回非localhost的URL

• 前置条件: 频道已创建
• 步骤: [1] 调用 POST /app/api/channels/authorize/scan?platform=wechat 生成二维码 [2] 验证返回的 qr_url 字段存在（base64 data URI） [3] 通过API调用确认绑定URL使用 APP_ACCESS_URL，非localhost [4] 验证API响应中的二维码图片不编码localhost地址
• 预期结果: 二维码包含生产地址，不含localhost

TC-QR-03 [P1] 验证临时域名检测模式列表

• 前置条件: config.py可读取
• 步骤: [1] 验证 _TEMP_DOMAIN_PATTERNS 包含: "team-coding", "localhost", "127.0.0.1", "0.0.0.0" [2] 测试 "http://s-3686137677310083075-773.team-coding:8080" → 应被识别为临时 [3] 测试 "https://bot.qihuyun.com" → 不应被识别为临时 [4] 测试 "http://localhost:8080" → 应被识别为临时
• 预期结果: 临时域名检测准确

TC-QR-04 [P1] 验证.env文件不存在的容错

• 前置条件: 模拟场景
• 步骤: [1] 确认代码中 .env 读取在try-except块中 [2] 验证当 .env 不存在或读取失败时不抛异常 [3] 验证最终仍走到兜底域名
• 预期结果: .env缺失不会导致启动失败

---

5. 测试用例统计

模块	P0用例	P1用例	合计
Q1: 统一Webhook端点	4	2	6
Q2: 消息组件系统	6	2	8
Q3: 平台注册机制	4	2	6
Q4: MessageSession	3	2	5
Q5: Pipeline管道	5	3	8
Q6: Redis事件总线	5	3	8
Q7: 二维码URL	2	2	4
合计	29	16	45

---

文档状态: 已完成
创建日期: 2026-06-01
基于: 后端代码功能分析 + 项目负责人指定优化范围