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%（忽略插件）
2026-01-09 23:18:58 +08:00
parent ec3a1c8eac
commit 8508fc95f5
17 changed files with 2057 additions and 112 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"