NeoBot/tests/test_core_managers.py

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"