Files
NeoBot/tests/test_event_factory.py
镀铬酸钾 fa81229f6f Dev (#28)
* 滚木

* feat: 重构核心架构,增强类型安全与插件管理

本次提交对核心模块进行了深度重构,引入 Pydantic 增强配置管理的类型安全性,并全面优化了插件管理系统。

主要变更详情:

1. 核心架构与配置
   - 重构配置加载模块:引入 Pydantic 模型 (`core/config_models.py`),提供严格的配置项类型检查、验证及默认值管理。
   - 统一模块结构:规范化模块导入路径,移除冗余的 `__init__.py` 文件,提升项目结构的清晰度。
   - 性能优化:集成 Redis 缓存支持 (`RedisManager`),有效降低高频 API 调用开销,提升响应速度。

2. 插件系统升级
   - 实现热重载机制:新增插件文件变更监听功能,支持开发过程中自动重载插件,提升开发效率。
   - 优化生命周期管理:改进插件加载与卸载逻辑,支持精确卸载指定插件及其关联的命令、事件处理器和定时任务。

3. 功能特性增强
   - 新增媒体 API:引入 `MediaAPI` 模块,封装图片、语音等富媒体资源的获取与处理接口。
   - 完善权限体系:重构权限管理系统,实现管理员与操作员的分级控制,支持更细粒度的命令权限校验。

4. 代码质量与稳定性
   - 全面类型修复:解决 `mypy` 静态类型检查发现的大量类型错误(包括 `CommandManager`、`EventFactory` 及 `Bot` API 签名不匹配问题)。
   - 增强错误处理:优化消息处理管道的异常捕获机制,完善关键路径的日志记录,提升系统运行稳定性。

* feat: 添加测试用例并优化代码结构

refactor(permission_manager): 调整初始化顺序和逻辑
fix(admin_manager): 修复初始化逻辑和目录创建问题
feat(ws): 优化Bot实例初始化条件
feat(message): 增强MessageSegment功能并添加测试
feat(events): 支持字符串格式的消息解析
test: 添加核心功能测试用例
refactor(plugin_manager): 改进插件路径处理
style: 清理无用导入和代码
chore: 更新依赖项
2026-01-09 00:20:56 +08:00

142 lines
4.8 KiB
Python

import pytest
from models.events.factory import EventFactory, EventType
from models.events.message import GroupMessageEvent, PrivateMessageEvent
from models.events.notice import GroupIncreaseNoticeEvent
from models.events.request import FriendRequestEvent
from models.events.meta import HeartbeatEvent
from models.message import MessageSegment
class TestEventFactory:
def test_create_group_message_event_list(self):
"""测试创建群消息事件 (message 为列表格式)"""
data = {
"post_type": "message",
"message_type": "group",
"time": 1600000000,
"self_id": 123456,
"sub_type": "normal",
"message_id": 1001,
"user_id": 111111,
"group_id": 222222,
"message": [
{"type": "text", "data": {"text": "Hello"}}
],
"raw_message": "Hello",
"font": 0,
"sender": {
"user_id": 111111,
"nickname": "User",
"role": "member"
}
}
event = EventFactory.create_event(data)
assert isinstance(event, GroupMessageEvent)
assert event.group_id == 222222
assert len(event.message) == 1
assert event.message[0].type == "text"
assert event.message[0].data["text"] == "Hello"
def test_create_group_message_event_str(self):
"""测试创建群消息事件 (message 为字符串格式)"""
data = {
"post_type": "message",
"message_type": "group",
"time": 1600000000,
"self_id": 123456,
"sub_type": "normal",
"message_id": 1002,
"user_id": 111111,
"group_id": 222222,
"message": "Hello World",
"raw_message": "Hello World",
"font": 0,
"sender": {
"user_id": 111111,
"nickname": "User"
}
}
event = EventFactory.create_event(data)
assert isinstance(event, GroupMessageEvent)
assert len(event.message) == 1
assert event.message[0].type == "text"
assert event.message[0].data["text"] == "Hello World"
def test_create_private_message_event(self):
"""测试创建私聊消息事件"""
data = {
"post_type": "message",
"message_type": "private",
"time": 1600000000,
"self_id": 123456,
"sub_type": "friend",
"message_id": 2001,
"user_id": 333333,
"message": "Private Msg",
"raw_message": "Private Msg",
"font": 0,
"sender": {
"user_id": 333333,
"nickname": "Friend"
}
}
event = EventFactory.create_event(data)
assert isinstance(event, PrivateMessageEvent)
assert event.user_id == 333333
def test_create_notice_event(self):
"""测试创建通知事件 (群成员增加)"""
data = {
"post_type": "notice",
"notice_type": "group_increase",
"sub_type": "approve",
"group_id": 222222,
"operator_id": 444444,
"user_id": 555555,
"time": 1600000000,
"self_id": 123456
}
event = EventFactory.create_event(data)
assert isinstance(event, GroupIncreaseNoticeEvent)
assert event.group_id == 222222
assert event.user_id == 555555
def test_create_request_event(self):
"""测试创建请求事件 (加好友)"""
data = {
"post_type": "request",
"request_type": "friend",
"user_id": 666666,
"comment": "Add me",
"flag": "flag_123",
"time": 1600000000,
"self_id": 123456
}
event = EventFactory.create_event(data)
assert isinstance(event, FriendRequestEvent)
assert event.user_id == 666666
assert event.comment == "Add me"
def test_create_meta_event(self):
"""测试创建元事件 (心跳)"""
data = {
"post_type": "meta_event",
"meta_event_type": "heartbeat",
"time": 1600000000,
"self_id": 123456,
"status": {"online": True, "good": True},
"interval": 5000
}
event = EventFactory.create_event(data)
assert isinstance(event, HeartbeatEvent)
assert event.interval == 5000
def test_unknown_event_type(self):
"""测试未知事件类型"""
data = {
"post_type": "unknown_type",
"time": 1600000000,
"self_id": 123456
}
with pytest.raises(ValueError, match="Unknown event type"):
EventFactory.create_event(data)