K2CrO4/NeoBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Dev (#28)

Browse Source 
* 滚木

* 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: 更新依赖项
This commit is contained in:
镀铬酸钾
2026-01-09 00:20:56 +08:00
committed by GitHub
parent 6d7dfc179d
commit fa81229f6f
42 changed files with 1461 additions and 697 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
tests/test_basic.py Normal file
View File

@@ -0,0 +1,37 @@
import pytest
import sys
import os
# 确保项目根目录在 sys.path 中
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
def test_import_core():
"""测试核心模块是否可以被导入"""
try:
import core
import core.bot
import core.ws
except ImportError as e:
pytest.fail(f"无法导入核心模块: {e}")
def test_plugin_manager_path():
"""测试插件管理器路径逻辑是否正确"""
from core.managers.plugin_manager import PluginManager
# Mock command manager
pm = PluginManager(None)
# 我们无法直接测试 load_all_plugins 的内部路径变量，
# 但我们可以检查它是否能找到 plugins 目录而不报错
# 这里我们简单地断言 PluginManager 类存在且可以实例化
assert pm is not None
def test_config_loader_exists():
"""测试配置加载器是否存在"""
# 注意：导入 config_loader 会尝试读取 config.toml
# 如果 config.toml 不存在，这可能会失败。
# 这是一个已知的设计问题，但在测试环境中我们假设 config.toml 存在或被 mock
if os.path.exists("config.toml"):
from core.config_loader import global_config
assert global_config is not None
else:
pytest.skip("config.toml 不存在，跳过配置加载测试")

114
tests/test_command_manager.py Normal file
View File

@@ -0,0 +1,114 @@
import pytest
import asyncio
from unittest.mock import AsyncMock, MagicMock, patch
from core.managers.command_manager import CommandManager
from models.events.message import GroupMessageEvent
from models.message import MessageSegment
@pytest.fixture
def mock_bot():
bot = AsyncMock()
bot.self_id = 123456
return bot
@pytest.fixture
def command_manager():
# 创建一个新的 CommandManager 实例用于测试，避免单例状态污染
return CommandManager(prefixes=("/",))
@pytest.mark.asyncio
async def test_command_registration_and_execution(command_manager, mock_bot):
"""测试命令注册和执行"""
# 定义一个命令处理函数
handler_mock = AsyncMock()
# 注册命令
@command_manager.command("test")
async def test_command(bot, event):
await handler_mock(bot, event)
# 构造触发命令的事件
event = MagicMock(spec=GroupMessageEvent)
event.post_type = "message"
event.message_type = "group"
event.raw_message = "/test"
event.message = [MessageSegment.text("/test")]
event.user_id = 111
event.group_id = 222
# 处理事件
await command_manager.handle_event(mock_bot, event)
# 验证处理函数被调用
handler_mock.assert_called_once_with(mock_bot, event)
@pytest.mark.asyncio
async def test_command_prefix_match(command_manager, mock_bot):
"""测试命令前缀匹配"""
handler_mock = AsyncMock()
@command_manager.command("hello")
async def hello_command(bot, event):
await handler_mock(bot, event)
# 1. 正确的前缀
event1 = MagicMock(spec=GroupMessageEvent)
event1.post_type = "message"
event1.raw_message = "/hello"
event1.message = [MessageSegment.text("/hello")]
await command_manager.handle_event(mock_bot, event1)
handler_mock.assert_called_once()
handler_mock.reset_mock()
# 2. 错误的前缀 (应该忽略)
event2 = MagicMock(spec=GroupMessageEvent)
event2.post_type = "message"
event2.raw_message = ".hello" # 假设前缀是 /
event2.message = [MessageSegment.text(".hello")]
await command_manager.handle_event(mock_bot, event2)
handler_mock.assert_not_called()
@pytest.mark.asyncio
async def test_ignore_self_message(command_manager, mock_bot):
"""测试忽略自身消息"""
# 模拟配置
with patch("core.managers.command_manager.global_config") as mock_config:
mock_config.bot.ignore_self_message = True
event = MagicMock(spec=GroupMessageEvent)
event.post_type = "message"
event.user_id = 123456 # 与 bot.self_id 相同
event.self_id = 123456
# Mock handle 方法来检测是否被调用
command_manager.message_handler.handle = AsyncMock()
await command_manager.handle_event(mock_bot, event)
# 应该直接返回，不调用 handler
command_manager.message_handler.handle.assert_not_called()
@pytest.mark.asyncio
async def test_help_command(command_manager, mock_bot):
"""测试内置 help 命令"""
# 注册一个测试插件信息
command_manager.plugins["test_plugin"] = {
"name": "测试插件",
"description": "这是一个测试",
"usage": "/test"
}
event = MagicMock(spec=GroupMessageEvent)
event.post_type = "message"
event.raw_message = "/help"
event.message = [MessageSegment.text("/help")]
await command_manager.handle_event(mock_bot, event)
# 验证 bot.send 被调用，且内容包含插件信息
mock_bot.send.assert_called_once()
args, _ = mock_bot.send.call_args
sent_msg = args[1]
assert "测试插件" in sent_msg
assert "这是一个测试" in sent_msg

141
tests/test_event_factory.py Normal file
View File

@@ -0,0 +1,141 @@
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)

194
tests/test_event_handler.py Normal file
View File

@@ -0,0 +1,194 @@
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from core.handlers.event_handler import MessageHandler, NoticeHandler, RequestHandler
from models.events.message import GroupMessageEvent
from models.events.notice import GroupIncreaseNoticeEvent
from models.events.request import FriendRequestEvent
@pytest.fixture
def mock_bot():
bot = AsyncMock()
return bot
@pytest.mark.asyncio
async def test_message_handler_run_handler_injection(mock_bot):
"""测试参数注入"""
handler = MessageHandler(prefixes=("/",))
# 1. 测试注入 bot 和 event
async def func1(bot, event):
assert bot == mock_bot
assert event.user_id == 123
return True
event = MagicMock(spec=GroupMessageEvent)
event.user_id = 123
result = await handler._run_handler(func1, mock_bot, event)
assert result is True
# 2. 测试注入 args
async def func2(args):
assert args == ["arg1", "arg2"]
return True
result = await handler._run_handler(func2, mock_bot, event, args=["arg1", "arg2"])
assert result is True
@pytest.mark.asyncio
async def test_message_handler_command_parsing(mock_bot):
"""测试命令解析"""
handler = MessageHandler(prefixes=("/",))
mock_func = AsyncMock()
handler.commands["test"] = {
"func": mock_func,
"permission": None,
"override_permission_check": False,
"plugin_name": "test_plugin"
}
event = MagicMock(spec=GroupMessageEvent)
event.raw_message = "/test arg1 arg2"
event.user_id = 123
# Mock permission manager
with patch("core.managers.permission_manager.PermissionManager.check_permission", new_callable=AsyncMock) as mock_perm:
mock_perm.return_value = True
await handler.handle(mock_bot, event)
mock_func.assert_called_once()
# 验证 args 参数是否正确传递
call_args = mock_func.call_args
if "args" in call_args.kwargs:
assert call_args.kwargs["args"] == ["arg1", "arg2"]
@pytest.mark.asyncio
async def test_notice_handler(mock_bot):
"""测试通知事件分发"""
handler = NoticeHandler()
mock_func = AsyncMock()
handler.handlers.append({
"type": "group_increase",
"func": mock_func,
"plugin_name": "test_plugin"
})
event = MagicMock(spec=GroupIncreaseNoticeEvent)
event.notice_type = "group_increase"
await handler.handle(mock_bot, event)
mock_func.assert_called_once()
@pytest.mark.asyncio
async def test_sync_handler_execution(mock_bot):
"""测试同步处理函数的执行"""
handler = MessageHandler(prefixes=("/",))
def sync_func(event):
return True
event = MagicMock(spec=GroupMessageEvent)
# 同步函数应该在线程池中运行
result = await handler._run_handler(sync_func, mock_bot, event)
assert result is True
@pytest.mark.asyncio
async def test_message_handler_management(mock_bot):
"""测试消息处理器的管理（注册、卸载、清空）"""
handler = MessageHandler(prefixes=("/",))
# 测试 on_message 装饰器
@handler.on_message()
async def msg_handler(event):
pass
assert len(handler.message_handlers) == 1
# 测试 command 装饰器
@handler.command("cmd1", "cmd2")
async def cmd_handler(event):
pass
assert len(handler.commands) == 2
assert "cmd1" in handler.commands
assert "cmd2" in handler.commands
# 测试 unregister_by_plugin_name
# 直接从已注册的处理器中获取 plugin_name
if handler.message_handlers:
plugin_name = handler.message_handlers[0]["plugin_name"]
handler.unregister_by_plugin_name(plugin_name)
assert len(handler.message_handlers) == 0
assert len(handler.commands) == 0
# 测试 clear
handler.commands["cmd"] = {}
handler.message_handlers.append({})
handler.clear()
assert len(handler.commands) == 0
assert len(handler.message_handlers) == 0
@pytest.mark.asyncio
async def test_request_handler(mock_bot):
"""测试请求事件处理器"""
handler = RequestHandler()
mock_func = AsyncMock()
# 测试 register 装饰器
@handler.register("friend")
async def req_handler(event):
await mock_func(event)
assert len(handler.handlers) == 1
event = MagicMock(spec=FriendRequestEvent)
event.request_type = "friend"
await handler.handle(mock_bot, event)
mock_func.assert_called_once()
# 测试 unregister 和 clear
import inspect
module = inspect.getmodule(req_handler)
plugin_name = module.__name__
handler.unregister_by_plugin_name(plugin_name)
assert len(handler.handlers) == 0
handler.handlers.append({})
handler.clear()
assert len(handler.handlers) == 0
@pytest.mark.asyncio
async def test_permission_denied(mock_bot):
"""测试权限不足的情况"""
handler = MessageHandler(prefixes=("/",))
mock_func = AsyncMock()
handler.commands["admin_cmd"] = {
"func": mock_func,
"permission": "ADMIN", # 假设 Permission.ADMIN
"override_permission_check": False,
"plugin_name": "test_plugin"
}
event = MagicMock(spec=GroupMessageEvent)
event.raw_message = "/admin_cmd"
event.user_id = 123
# Mock permission manager returning False
with patch("core.managers.permission_manager.PermissionManager.check_permission", new_callable=AsyncMock) as mock_perm:
mock_perm.return_value = False
await handler.handle(mock_bot, event)
mock_func.assert_not_called()
# 应该发送拒绝消息
mock_bot.send.assert_called_once()

75
tests/test_models.py Normal file
View File

@@ -0,0 +1,75 @@
import pytest
from models.message import MessageSegment
from models.objects import GroupInfo, StrangerInfo
class TestMessageSegment:
def test_text_segment(self):
seg = MessageSegment.text("Hello")
assert seg.type == "text"
assert seg.data["text"] == "Hello"
assert str(seg) == "Hello"
def test_at_segment(self):
seg = MessageSegment.at(123456)
assert seg.type == "at"
assert seg.data["qq"] == "123456"
assert str(seg) == "[CQ:at,qq=123456]"
def test_image_segment(self):
seg = MessageSegment.image("http://example.com/img.jpg", cache=False, proxy=False)
assert seg.type == "image"
assert seg.data["file"] == "http://example.com/img.jpg"
assert str(seg) == "[CQ:image,file=http://example.com/img.jpg,cache=0,proxy=0]"
def test_face_segment(self):
seg = MessageSegment.face(123)
assert seg.type == "face"
assert seg.data["id"] == "123"
assert str(seg) == "[CQ:face,id=123]"
def test_reply_segment(self):
seg = MessageSegment.reply(1001)
assert seg.type == "reply"
assert seg.data["id"] == "1001"
assert str(seg) == "[CQ:reply,id=1001]"
def test_add_segments(self):
seg1 = MessageSegment.text("Hello ")
seg2 = MessageSegment.at(123)
combined = seg1 + seg2
assert isinstance(combined, list)
assert len(combined) == 2
assert combined[0] == seg1
assert combined[1] == seg2
def test_add_segment_and_string(self):
seg = MessageSegment.at(123)
combined = seg + " Hello"
assert isinstance(combined, list)
assert len(combined) == 2
assert combined[0] == seg
assert combined[1].type == "text"
assert combined[1].data["text"] == " Hello"
class TestObjects:
def test_group_info(self):
data = {
"group_id": 123456,
"group_name": "Test Group",
"member_count": 10,
"max_member_count": 100
}
group = GroupInfo(**data)
assert group.group_id == 123456
assert group.group_name == "Test Group"
def test_stranger_info(self):
data = {
"user_id": 111111,
"nickname": "Stranger",
"sex": "male",
"age": 18
}
user = StrangerInfo(**data)
assert user.user_id == 111111
assert user.nickname == "Stranger"