更新/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
parent 5f16c288bf
commit 651d982e19
20 changed files with 2077 additions and 124 deletions
--- a/tests/test_core_managers.py
+++ b/tests/test_core_managers.py
@@ -0,0 +1,290 @@
+
+import json
+import os
+import tempfile
+import pytest
+from unittest.mock import MagicMock, patch, AsyncMock
+
+from core.managers.permission_manager import PermissionManager
+from core.managers.admin_manager import AdminManager
+from core.permission import Permission
+
+# --- Fixtures ---
+
+@pytest.fixture
+def mock_redis():
+    """Mock RedisManager to avoid real Redis connection"""
+    with patch("core.managers.redis_manager.redis_manager") as mock:
+        mock.redis = AsyncMock()
+        # Mock sismember to return False by default
+        mock.redis.sismember.return_value = False
+        yield mock
+
+@pytest.fixture
+def temp_data_dir():
+    """Create a temporary directory for data files"""
+    with tempfile.TemporaryDirectory() as tmpdirname:
+        yield tmpdirname
+
+@pytest.fixture
+def admin_manager(temp_data_dir, mock_redis):
+    """Create an AdminManager instance with temporary data file"""
+    # Reset singleton instance if it exists
+    if hasattr(AdminManager, "_instance"):
+        del AdminManager._instance
+    
+    # Patch the data file path
+    with patch("core.managers.admin_manager.AdminManager.__init__", return_value=None) as mock_init:
+        manager = AdminManager()
+        # Manually initialize necessary attributes since we mocked __init__
+        manager.data_file = os.path.join(temp_data_dir, "admin.json")
+        manager._admins = set()
+        # Call the real __init__ logic we want to test (partially) or just setup state
+        # Actually, it's better to let __init__ run but patch the path inside it.
+        # But AdminManager is a Singleton, which makes it tricky.
+        pass
+
+    # Let's try a different approach: Patch the class attribute or use a fresh instance logic
+    # Since Singleton logic might prevent re-init, we force it.
+    
+    # Re-create properly
+    if hasattr(AdminManager, "_instance"):
+        del AdminManager._instance
+        
+    with patch("core.managers.admin_manager.os.path.dirname") as mock_dirname:
+        # We want os.path.join(..., "data", "admin.json") to resolve to our temp file
+        # But the path construction is hardcoded.
+        # Instead, we can patch the `data_file` attribute after init if we can.
+        
+        # Easiest way: Subclass or modify the instance after creation, 
+        # but __init__ runs immediately.
+        
+        # Let's patch `os.path.abspath` to redirect the base path? 
+        # No, let's just patch the `data_file` attribute on the instance.
+        
+        manager = AdminManager()
+        manager.data_file = os.path.join(temp_data_dir, "admin.json")
+        manager._admins = set() # Reset in-memory state
+        
+        return manager
+
+@pytest.fixture
+def permission_manager(temp_data_dir, admin_manager):
+    """Create a PermissionManager instance with temporary data file"""
+    if hasattr(PermissionManager, "_instance"):
+        del PermissionManager._instance
+        
+    manager = PermissionManager()
+    manager.data_file = os.path.join(temp_data_dir, "permissions.json")
+    manager._data = {"users": {}} # Reset in-memory state
+    
+    # Ensure admin_manager is linked correctly if needed (it's imported globally in permission_manager)
+    # We need to patch the global admin_manager used in permission_manager
+    with patch("core.managers.permission_manager.admin_manager", admin_manager):
+        yield manager
+
+
+# --- AdminManager Tests ---
+
+@pytest.mark.asyncio
+async def test_admin_manager_load_save(admin_manager):
+    """Test loading and saving admins to file"""
+    # Test adding and saving
+    await admin_manager.add_admin(123456)
+    assert 123456 in admin_manager._admins
+    
+    # Verify file content
+    with open(admin_manager.data_file, "r", encoding="utf-8") as f:
+        data = json.load(f)
+        assert "123456" in data["admins"]
+        
+    # Test loading
+    # Clear memory
+    admin_manager._admins.clear()
+    await admin_manager._load_from_file()
+    assert 123456 in admin_manager._admins
+
+@pytest.mark.asyncio
+async def test_admin_manager_operations(admin_manager, mock_redis):
+    """Test add, remove, and is_admin operations"""
+    user_id = 1001
+    
+    # Initially not admin
+    assert not await admin_manager.is_admin(user_id)
+    
+    # Add admin
+    success = await admin_manager.add_admin(user_id)
+    assert success
+    assert await admin_manager.is_admin(user_id)
+    mock_redis.redis.sadd.assert_called()
+    
+    # Add duplicate
+    success = await admin_manager.add_admin(user_id)
+    assert not success
+    
+    # Remove admin
+    success = await admin_manager.remove_admin(user_id)
+    assert success
+    assert not await admin_manager.is_admin(user_id)
+    mock_redis.redis.srem.assert_called()
+    
+    # Remove non-existent
+    success = await admin_manager.remove_admin(user_id)
+    assert not success
+
+@pytest.mark.asyncio
+async def test_admin_manager_sync_redis(admin_manager, mock_redis):
+    """Test syncing to Redis"""
+    admin_manager._admins = {111, 222}
+    await admin_manager._sync_to_redis()
+    
+    mock_redis.redis.delete.assert_called_with(admin_manager._REDIS_KEY)
+    
+    # Check sadd call args manually because set order is not guaranteed
+    args, _ = mock_redis.redis.sadd.call_args
+    assert args[0] == admin_manager._REDIS_KEY
+    assert set(args[1:]) == {111, 222}
+
+
+# --- PermissionManager Tests ---
+
+@pytest.mark.asyncio
+async def test_permission_manager_load_save(permission_manager):
+    """Test loading and saving permissions"""
+    user_id = 2001
+    permission_manager.set_user_permission(user_id, Permission.OP)
+    
+    # Verify memory
+    assert permission_manager._data["users"][str(user_id)] == "op"
+    
+    # Verify file
+    with open(permission_manager.data_file, "r", encoding="utf-8") as f:
+        data = json.load(f)
+        assert data["users"][str(user_id)] == "op"
+        
+    # Test load
+    permission_manager._data["users"] = {}
+    permission_manager.load()
+    assert permission_manager._data["users"][str(user_id)] == "op"
+
+@pytest.mark.asyncio
+async def test_permission_check_flow(permission_manager, admin_manager):
+    """Test permission checking logic including admin fallback"""
+    admin_id = 8888
+    op_id = 6666
+    user_id = 1111
+    
+    # Setup admin
+    await admin_manager.add_admin(admin_id)
+    
+    # Setup OP
+    permission_manager.set_user_permission(op_id, Permission.OP)
+    
+    # Test Admin (should be ADMIN even if not in permissions.json)
+    perm = await permission_manager.get_user_permission(admin_id)
+    assert perm == Permission.ADMIN
+    assert await permission_manager.check_permission(admin_id, Permission.ADMIN)
+    assert await permission_manager.check_permission(admin_id, Permission.OP)
+    
+    # Test OP
+    perm = await permission_manager.get_user_permission(op_id)
+    assert perm == Permission.OP
+    assert not await permission_manager.check_permission(op_id, Permission.ADMIN)
+    assert await permission_manager.check_permission(op_id, Permission.OP)
+    assert await permission_manager.check_permission(op_id, Permission.USER)
+    
+    # Test User (Default)
+    perm = await permission_manager.get_user_permission(user_id)
+    assert perm == Permission.USER
+    assert not await permission_manager.check_permission(user_id, Permission.OP)
+    assert await permission_manager.check_permission(user_id, Permission.USER)
+
+@pytest.mark.asyncio
+async def test_get_all_user_permissions(permission_manager, admin_manager):
+    """Test merging of admin and permission data"""
+    admin_id = 9999
+    op_id = 7777
+    
+    await admin_manager.add_admin(admin_id)
+    permission_manager.set_user_permission(op_id, Permission.OP)
+    
+    all_perms = await permission_manager.get_all_user_permissions()
+    
+    assert str(admin_id) in all_perms
+    assert all_perms[str(admin_id)] == "admin"
+    assert str(op_id) in all_perms
+    assert all_perms[str(op_id)] == "op"
+
+def test_remove_user(permission_manager):
+    """Test removing user permission"""
+    user_id = 3001
+    permission_manager.set_user_permission(user_id, Permission.OP)
+    assert str(user_id) in permission_manager._data["users"]
+    
+    permission_manager.remove_user(user_id)
+    assert str(user_id) not in permission_manager._data["users"]
+
+@pytest.mark.asyncio
+async def test_permission_manager_load_error(permission_manager):
+    """Test loading permissions with invalid file"""
+    # Write invalid JSON
+    with open(permission_manager.data_file, "w", encoding="utf-8") as f:
+        f.write("{invalid_json")
+        
+    # Should not raise exception, but log error (we can't easily check log here without more mocking)
+    # But we can check that data remains empty or default
+    permission_manager._data["users"] = {}
+    permission_manager.load()
+    assert permission_manager._data["users"] == {}
+
+@pytest.mark.asyncio
+async def test_admin_manager_redis_error(admin_manager, mock_redis):
+    """Test Redis errors are handled gracefully"""
+    mock_redis.redis.sadd.side_effect = Exception("Redis error")
+    
+    # Should not raise exception
+    success = await admin_manager.add_admin(123)
+    assert not success  # Or however it handles it - let's check implementation
+    # Looking at code: try...except Exception... return False
+    
+    mock_redis.redis.srem.side_effect = Exception("Redis error")
+    success = await admin_manager.remove_admin(123)
+    assert not success
+
+def test_permission_manager_utils(permission_manager):
+    """Test utility methods like get_all_users and clear_all"""
+    permission_manager.set_user_permission(123, Permission.OP)
+    permission_manager.set_user_permission(456, Permission.USER)
+    
+    users = permission_manager.get_all_users()
+    assert "123" in users
+    assert "456" in users
+    
+    permission_manager.clear_all()
+    assert len(permission_manager.get_all_users()) == 0
+
+@pytest.mark.asyncio
+async def test_require_admin_decorator(permission_manager, admin_manager):
+    """Test the require_admin decorator"""
+    from core.managers.permission_manager import require_admin
+    from models.events.message import MessageEvent
+    
+    # Mock event
+    mock_event = MagicMock(spec=MessageEvent)
+    mock_event.user_id = 12345
+    mock_event.reply = AsyncMock()
+    
+    # Define decorated function
+    @require_admin
+    async def protected_func(event, *args):
+        return "success"
+    
+    # Test without permission
+    result = await protected_func(mock_event)
+    assert result is None
+    mock_event.reply.assert_called_with("抱歉，您没有权限执行此命令。")
+    
+    # Test with permission
+    await admin_manager.add_admin(12345)
+    result = await protected_func(mock_event)
+    assert result == "success"