K2CrO4/NeoBot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
501d06d4ddd8a714ccd6671fd14497f37a65be12
NeoBot/tests/test_event_factory.py
baby2016 651d982e19 更新/help (#31) 
* 滚木

* 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: 更新依赖项

* refactor(handler): 移除TYPE_CHECKING并直接导入Bot类

简化类型注解，直接导入Bot类而非使用TYPE_CHECKING条件导入，提高代码可读性和维护性

* fix(command_manager): 修复插件卸载时元信息移除不精确的问题

修复 CommandManager 中 unload_plugin 方法移除插件元信息时使用 startswith 导致可能误删其他插件的问题，改为精确匹配
同时调整相关测试用例验证精确匹配行为

* refactor: 清理未使用的导入和更新文档结构

docs: 添加config_models.py到项目结构文档
docs: 调整数据目录位置到core/data下
docs: 更新权限管理器文档描述

* 文档更新

* 更新thpic插件 支持一次返回多张图

* feat: 添加测试覆盖率并修复相关问题

refactor(redis_manager): 移除冗余的ConnectionError处理
refactor(event_handler): 优化Bot类型注解
refactor(factory): 移除未使用的GroupCardNoticeEvent

test: 添加全面的单元测试覆盖
- 添加test_import.py测试模块导入
- 添加test_debug.py测试插件加载调试
- 添加test_plugin_error.py测试错误处理
- 添加test_config_loader.py测试配置加载
- 添加test_redis_manager.py测试Redis管理
- 添加test_bot.py测试Bot功能
- 扩展test_models.py测试消息模型
- 添加test_plugin_manager_coverage.py测试插件管理
- 添加test_executor.py测试代码执行器
- 添加test_ws.py测试WebSocket
- 添加test_api.py测试API接口
- 添加test_core_managers.py测试核心管理模块

fix(plugin_manager): 修复插件加载日志变量问题

覆盖率已到达86%（忽略插件）

* 更新/help指令，现在会发送图片

---------

Co-authored-by: K2cr2O1 <2221577113@qq.com>
Co-authored-by: 镀铬酸钾 <148796996+K2cr2O1@users.noreply.github.com>
2026-01-10 20:39:52 +08:00

431 lines
15 KiB
Python
Raw Blame History

import pytest
from models.events.factory import EventFactory
from models.events.base import EventType
from models.events.message import GroupMessageEvent, PrivateMessageEvent
from models.events.notice import (
FriendAddNoticeEvent, FriendRecallNoticeEvent, GroupRecallNoticeEvent,
GroupIncreaseNoticeEvent, GroupDecreaseNoticeEvent, GroupAdminNoticeEvent,
GroupBanNoticeEvent, GroupUploadNoticeEvent, PokeNotifyEvent,
LuckyKingNotifyEvent, HonorNotifyEvent, GroupCardNoticeEvent,
OfflineFileNoticeEvent, ClientStatusNoticeEvent, EssenceNoticeEvent,
NotifyNoticeEvent
)
from models.events.request import FriendRequestEvent, GroupRequestEvent
from models.events.meta import HeartbeatEvent, LifeCycleEvent
class TestEventFactory:
def test_create_private_message_event(self):
"""测试创建私聊消息事件。"""
data = {
"post_type": EventType.MESSAGE,
"message_type": "private",
"time": 1234567890,
"self_id": 10000,
"message_id": 123,
"user_id": 20000,
"message": [{"type": "text", "data": {"text": "Hello"}}],
"raw_message": "Hello",
"font": 12,
"sender": {"user_id": 20000, "nickname": "TestUser"}
}
event = EventFactory.create_event(data)
assert isinstance(event, PrivateMessageEvent)
assert event.message_type == "private"
assert event.user_id == 20000
assert len(event.message) == 1
assert event.message[0].type == "text"
assert event.message[0].data["text"] == "Hello"
def test_create_group_message_event(self):
"""测试创建群消息事件。"""
data = {
"post_type": EventType.MESSAGE,
"message_type": "group",
"time": 1234567890,
"self_id": 10000,
"message_id": 123,
"user_id": 20000,
"group_id": 30000,
"message": [{"type": "text", "data": {"text": "Hello"}}],
"raw_message": "Hello",
"font": 12,
"sender": {"user_id": 20000, "nickname": "TestUser", "role": "member"}
}
event = EventFactory.create_event(data)
assert isinstance(event, GroupMessageEvent)
assert event.message_type == "group"
assert event.group_id == 30000
assert event.user_id == 20000
def test_create_group_message_with_anonymous(self):
"""测试创建匿名群消息事件。"""
data = {
"post_type": EventType.MESSAGE,
"message_type": "group",
"time": 1234567890,
"self_id": 10000,
"message_id": 123,
"user_id": 20000,
"group_id": 30000,
"anonymous": {"id": 12345, "name": "Anonymous", "flag": "flag123"},
"message": [{"type": "text", "data": {"text": "Hello"}}],
"raw_message": "Hello",
"font": 12,
"sender": {"user_id": 20000, "nickname": "TestUser", "role": "member"}
}
event = EventFactory.create_event(data)
assert isinstance(event, GroupMessageEvent)
assert event.anonymous is not None
assert event.anonymous.id == 12345
assert event.anonymous.name == "Anonymous"
assert event.anonymous.flag == "flag123"
def test_create_friend_add_notice(self):
"""测试创建好友添加通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "friend_add",
"time": 1234567890,
"self_id": 10000,
"user_id": 20000
}
event = EventFactory.create_event(data)
assert isinstance(event, FriendAddNoticeEvent)
assert event.notice_type == "friend_add"
assert event.user_id == 20000
def test_create_friend_recall_notice(self):
"""测试创建好友消息撤回通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "friend_recall",
"time": 1234567890,
"self_id": 10000,
"user_id": 20000,
"message_id": 123
}
event = EventFactory.create_event(data)
assert isinstance(event, FriendRecallNoticeEvent)
assert event.notice_type == "friend_recall"
assert event.message_id == 123
def test_create_group_recall_notice(self):
"""测试创建群消息撤回通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "group_recall",
"time": 1234567890,
"self_id": 10000,
"group_id": 30000,
"user_id": 20000,
"operator_id": 40000,
"message_id": 123
}
event = EventFactory.create_event(data)
assert isinstance(event, GroupRecallNoticeEvent)
assert event.notice_type == "group_recall"
assert event.group_id == 30000
assert event.operator_id == 40000
def test_create_group_increase_notice(self):
"""测试创建群成员增加通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "group_increase",
"time": 1234567890,
"self_id": 10000,
"group_id": 30000,
"user_id": 20000,
"operator_id": 40000,
"sub_type": "approve"
}
event = EventFactory.create_event(data)
assert isinstance(event, GroupIncreaseNoticeEvent)
assert event.notice_type == "group_increase"
assert event.sub_type == "approve"
def test_create_group_decrease_notice(self):
"""测试创建群成员减少通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "group_decrease",
"time": 1234567890,
"self_id": 10000,
"group_id": 30000,
"user_id": 20000,
"operator_id": 40000,
"sub_type": "kick"
}
event = EventFactory.create_event(data)
assert isinstance(event, GroupDecreaseNoticeEvent)
assert event.notice_type == "group_decrease"
assert event.sub_type == "kick"
def test_create_group_admin_notice(self):
"""测试创建群管理员变更通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "group_admin",
"time": 1234567890,
"self_id": 10000,
"group_id": 30000,
"user_id": 20000,
"sub_type": "set"
}
event = EventFactory.create_event(data)
assert isinstance(event, GroupAdminNoticeEvent)
assert event.notice_type == "group_admin"
assert event.sub_type == "set"
def test_create_group_ban_notice(self):
"""测试创建群成员禁言通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "group_ban",
"time": 1234567890,
"self_id": 10000,
"group_id": 30000,
"user_id": 20000,
"operator_id": 40000,
"duration": 3600,
"sub_type": "ban"
}
event = EventFactory.create_event(data)
assert isinstance(event, GroupBanNoticeEvent)
assert event.notice_type == "group_ban"
assert event.duration == 3600
def test_create_group_upload_notice(self):
"""测试创建群文件上传通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "group_upload",
"time": 1234567890,
"self_id": 10000,
"group_id": 30000,
"user_id": 20000,
"file": {"id": "file123", "name": "test.txt", "size": 1024, "busid": 1}
}
event = EventFactory.create_event(data)
assert isinstance(event, GroupUploadNoticeEvent)
assert event.notice_type == "group_upload"
assert event.file.name == "test.txt"
assert event.file.size == 1024
def test_create_poke_notify_event(self):
"""测试创建戳一戳通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "notify",
"sub_type": "poke",
"time": 1234567890,
"self_id": 10000,
"group_id": 30000,
"user_id": 20000,
"target_id": 40000
}
event = EventFactory.create_event(data)
assert isinstance(event, PokeNotifyEvent)
assert event.notice_type == "notify"
assert event.sub_type == "poke"
def test_create_lucky_king_notify_event(self):
"""测试创建运气王通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "notify",
"sub_type": "lucky_king",
"time": 1234567890,
"self_id": 10000,
"group_id": 30000,
"user_id": 20000,
"target_id": 40000
}
event = EventFactory.create_event(data)
assert isinstance(event, LuckyKingNotifyEvent)
assert event.sub_type == "lucky_king"
def test_create_honor_notify_event(self):
"""测试创建荣誉变更通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "notify",
"sub_type": "honor",
"time": 1234567890,
"self_id": 10000,
"group_id": 30000,
"user_id": 20000,
"honor_type": "talkative"
}
event = EventFactory.create_event(data)
assert isinstance(event, HonorNotifyEvent)
assert event.sub_type == "honor"
assert event.honor_type == "talkative"
def test_create_unknown_notify_event(self):
"""测试创建未知类型的通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "notify",
"sub_type": "unknown",
"time": 1234567890,
"self_id": 10000,
"user_id": 20000
}
event = EventFactory.create_event(data)
assert isinstance(event, NotifyNoticeEvent)
assert event.notice_type == "notify"
assert event.sub_type == "unknown"
def test_create_group_card_notice(self):
"""测试创建群名片变更通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "group_card",
"time": 1234567890,
"self_id": 10000,
"group_id": 30000,
"user_id": 20000,
"card_new": "NewCard",
"card_old": "OldCard"
}
event = EventFactory.create_event(data)
assert isinstance(event, GroupCardNoticeEvent)
assert event.notice_type == "group_card"
assert event.card_new == "NewCard"
assert event.card_old == "OldCard"
def test_create_offline_file_notice(self):
"""测试创建离线文件通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "offline_file",
"time": 1234567890,
"self_id": 10000,
"user_id": 20000,
"file": {"name": "test.txt", "size": 1024, "url": "http://example.com/test.txt"}
}
event = EventFactory.create_event(data)
assert isinstance(event, OfflineFileNoticeEvent)
assert event.notice_type == "offline_file"
assert event.file.name == "test.txt"
def test_create_client_status_notice(self):
"""测试创建客户端状态通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "client_status",
"time": 1234567890,
"self_id": 10000,
"client": {"online": True, "status": "normal"}
}
event = EventFactory.create_event(data)
assert isinstance(event, ClientStatusNoticeEvent)
assert event.notice_type == "client_status"
assert event.client.online is True
def test_create_essence_notice(self):
"""测试创建精华消息通知事件。"""
data = {
"post_type": EventType.NOTICE,
"notice_type": "essence",
"time": 1234567890,
"self_id": 10000,
"group_id": 30000,
"sender_id": 20000,
"operator_id": 40000,
"message_id": 123,
"sub_type": "add"
}
event = EventFactory.create_event(data)
assert isinstance(event, EssenceNoticeEvent)
assert event.notice_type == "essence"
assert event.sub_type == "add"
def test_create_friend_request_event(self):
"""测试创建好友请求事件。"""
data = {
"post_type": EventType.REQUEST,
"request_type": "friend",
"time": 1234567890,
"self_id": 10000,
"user_id": 20000,
"comment": "Hello",
"flag": "flag123"
}
event = EventFactory.create_event(data)
assert isinstance(event, FriendRequestEvent)
assert event.request_type == "friend"
assert event.comment == "Hello"
def test_create_group_request_event(self):
"""测试创建群请求事件。"""
data = {
"post_type": EventType.REQUEST,
"request_type": "group",
"sub_type": "add",
"time": 1234567890,
"self_id": 10000,
"group_id": 30000,
"user_id": 20000,
"comment": "Hello",
"flag": "flag123"
}
event = EventFactory.create_event(data)
assert isinstance(event, GroupRequestEvent)
assert event.request_type == "group"
assert event.sub_type == "add"
def test_create_heartbeat_event(self):
"""测试创建心跳元事件。"""
data = {
"post_type": EventType.META,
"meta_event_type": "heartbeat",
"time": 1234567890,
"self_id": 10000,
"status": {"online": True, "good": True},
"interval": 1000
}
event = EventFactory.create_event(data)
assert isinstance(event, HeartbeatEvent)
assert event.meta_event_type == "heartbeat"
assert event.status.online is True
assert event.interval == 1000
def test_create_lifecycle_event(self):
"""测试创建生命周期元事件。"""
data = {
"post_type": EventType.META,
"meta_event_type": "lifecycle",
"time": 1234567890,
"self_id": 10000,
"sub_type": "enable"
}
event = EventFactory.create_event(data)
assert isinstance(event, LifeCycleEvent)
assert event.meta_event_type == "lifecycle"
assert event.sub_type == "enable"
def test_create_unknown_event_type(self):
"""测试创建未知类型事件时抛出异常。"""
data = {
"post_type": "unknown",
"time": 1234567890,
"self_id": 10000
}
with pytest.raises(ValueError, match="Unknown event type: unknown"):
EventFactory.create_event(data)
def test_create_unknown_message_type(self):
"""测试创建未知消息类型时抛出异常。"""
data = {
"post_type": EventType.MESSAGE,
"message_type": "unknown",
"time": 1234567890,
"self_id": 10000,
"message": "Hello"
}
with pytest.raises(ValueError, match="Unknown message type: unknown"):
EventFactory.create_event(data)
Reference in New Issue View Git Blame Copy Permalink