Mofox-Core/tests/test_wakeup_system.py

import pytest
import time
from unittest.mock import Mock, patch
import sys
import os

# 添加项目根目录到Python路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))

from src.chat.chat_loop.wakeup_manager import WakeUpManager
from src.chat.chat_loop.hfc_context import HfcContext
from src.config.official_configs import WakeUpSystemConfig


class TestWakeUpManager:
    """唤醒度管理器测试类"""
    
    @pytest.fixture
    def mock_context(self):
        """创建模拟的HFC上下文"""
        context = Mock(spec=HfcContext)
        context.stream_id = "test_chat_123"
        context.log_prefix = "[TEST]"
        context.running = True
        return context
    
    @pytest.fixture
    def wakeup_config(self):
        """创建测试用的唤醒度配置"""
        return WakeUpSystemConfig(
            enable=True,
            wakeup_threshold=15.0,
            private_message_increment=3.0,
            group_mention_increment=2.0,
            decay_rate=0.2,
            decay_interval=30.0,
            angry_duration=300.0  # 5分钟
        )
    
    @pytest.fixture
    def wakeup_manager(self, mock_context, wakeup_config):
        """创建唤醒度管理器实例"""
        with patch('src.chat.chat_loop.wakeup_manager.global_config') as mock_global_config:
            mock_global_config.wakeup_system = wakeup_config
            manager = WakeUpManager(mock_context)
            return manager
    
    def test_initialization(self, wakeup_manager, wakeup_config):
        """测试初始化"""
        assert wakeup_manager.wakeup_value == 0.0
        assert wakeup_manager.is_angry == False
        assert wakeup_manager.wakeup_threshold == wakeup_config.wakeup_threshold
        assert wakeup_manager.private_message_increment == wakeup_config.private_message_increment
        assert wakeup_manager.group_mention_increment == wakeup_config.group_mention_increment
        assert wakeup_manager.decay_rate == wakeup_config.decay_rate
        assert wakeup_manager.decay_interval == wakeup_config.decay_interval
        assert wakeup_manager.angry_duration == wakeup_config.angry_duration
        assert wakeup_manager.enabled == wakeup_config.enable
    
    @patch('src.manager.schedule_manager.schedule_manager')
    @patch('src.mood.mood_manager.mood_manager')
    def test_private_message_wakeup_accumulation(self, mock_mood_manager, mock_schedule_manager, wakeup_manager):
        """测试私聊消息唤醒度累积"""
        # 模拟休眠状态
        mock_schedule_manager.is_sleeping.return_value = True
        
        # 发送5条私聊消息 (5 * 3.0 = 15.0，达到阈值)
        for i in range(4):
            result = wakeup_manager.add_wakeup_value(is_private_chat=True)
            assert result == False  # 前4条消息不应该触发唤醒
            assert wakeup_manager.wakeup_value == (i + 1) * 3.0
        
        # 第5条消息应该触发唤醒
        result = wakeup_manager.add_wakeup_value(is_private_chat=True)
        assert result == True
        assert wakeup_manager.is_angry == True
        assert wakeup_manager.wakeup_value == 0.0  # 唤醒后重置
        
        # 验证情绪管理器被调用
        mock_mood_manager.set_angry_from_wakeup.assert_called_once_with("test_chat_123")
    
    @patch('src.manager.schedule_manager.schedule_manager')
    @patch('src.mood.mood_manager.mood_manager')
    def test_group_mention_wakeup_accumulation(self, mock_mood_manager, mock_schedule_manager, wakeup_manager):
        """测试群聊艾特消息唤醒度累积"""
        # 模拟休眠状态
        mock_schedule_manager.is_sleeping.return_value = True
        
        # 发送7条群聊艾特消息 (7 * 2.0 = 14.0，未达到阈值)
        for i in range(7):
            result = wakeup_manager.add_wakeup_value(is_private_chat=False, is_mentioned=True)
            assert result == False
            assert wakeup_manager.wakeup_value == (i + 1) * 2.0
        
        # 第8条消息应该触发唤醒 (8 * 2.0 = 16.0，超过阈值15.0)
        result = wakeup_manager.add_wakeup_value(is_private_chat=False, is_mentioned=True)
        assert result == True
        assert wakeup_manager.is_angry == True
        assert wakeup_manager.wakeup_value == 0.0
        
        # 验证情绪管理器被调用
        mock_mood_manager.set_angry_from_wakeup.assert_called_once_with("test_chat_123")
    
    @patch('src.manager.schedule_manager.schedule_manager')
    def test_group_message_without_mention(self, mock_schedule_manager, wakeup_manager):
        """测试群聊未艾特消息不增加唤醒度"""
        # 模拟休眠状态
        mock_schedule_manager.is_sleeping.return_value = True
        
        # 发送群聊消息但未被艾特
        result = wakeup_manager.add_wakeup_value(is_private_chat=False, is_mentioned=False)
        assert result == False
        assert wakeup_manager.wakeup_value == 0.0  # 不应该增加
    
    @patch('src.manager.schedule_manager.schedule_manager')
    def test_no_accumulation_when_not_sleeping(self, mock_schedule_manager, wakeup_manager):
        """测试非休眠状态下不累积唤醒度"""
        # 模拟非休眠状态
        mock_schedule_manager.is_sleeping.return_value = False
        
        # 发送私聊消息
        result = wakeup_manager.add_wakeup_value(is_private_chat=True)
        assert result == False
        assert wakeup_manager.wakeup_value == 0.0  # 不应该增加
    
    def test_disabled_system(self, mock_context):
        """测试系统禁用时的行为"""
        disabled_config = WakeUpSystemConfig(enable=False)
        
        with patch('src.chat.chat_loop.wakeup_manager.global_config') as mock_global_config:
            mock_global_config.wakeup_system = disabled_config
            manager = WakeUpManager(mock_context)
            
            with patch('src.manager.schedule_manager.schedule_manager') as mock_schedule_manager:
                mock_schedule_manager.is_sleeping.return_value = True
                
                # 即使发送消息也不应该累积唤醒度
                result = manager.add_wakeup_value(is_private_chat=True)
                assert result == False
                assert manager.wakeup_value == 0.0
    
    @patch('src.mood.mood_manager.mood_manager')
    def test_angry_state_expiration(self, mock_mood_manager, wakeup_manager):
        """测试愤怒状态过期"""
        # 手动设置愤怒状态
        wakeup_manager.is_angry = True
        wakeup_manager.angry_start_time = time.time() - 400  # 400秒前开始愤怒（超过300秒持续时间）
        
        # 检查愤怒状态应该已过期
        is_angry = wakeup_manager.is_in_angry_state()
        assert is_angry == False
        assert wakeup_manager.is_angry == False
        
        # 验证情绪管理器被调用清除愤怒状态
        mock_mood_manager.clear_angry_from_wakeup.assert_called_once_with("test_chat_123")
    
    def test_angry_prompt_addition(self, wakeup_manager):
        """测试愤怒状态提示词"""
        # 非愤怒状态
        prompt = wakeup_manager.get_angry_prompt_addition()
        assert prompt == ""
        
        # 愤怒状态
        wakeup_manager.is_angry = True
        wakeup_manager.angry_start_time = time.time()
        prompt = wakeup_manager.get_angry_prompt_addition()
        assert "吵醒" in prompt and "生气" in prompt
    
    def test_status_info(self, wakeup_manager):
        """测试状态信息获取"""
        # 设置一些状态
        wakeup_manager.wakeup_value = 10.5
        wakeup_manager.is_angry = True
        wakeup_manager.angry_start_time = time.time()
        
        status = wakeup_manager.get_status_info()
        
        assert status["wakeup_value"] == 10.5
        assert status["wakeup_threshold"] == 15.0
        assert status["is_angry"] == True
        assert status["angry_remaining_time"] > 0
    
    @pytest.mark.asyncio
    async def test_decay_loop(self, wakeup_manager):
        """测试衰减循环"""
        # 设置初始唤醒度
        wakeup_manager.wakeup_value = 5.0
        
        # 模拟一次衰减
        with patch('asyncio.sleep') as mock_sleep:
            # 创建一个会立即停止的衰减循环
            wakeup_manager.context.running = False
            
            # 手动调用衰减逻辑
            if wakeup_manager.wakeup_value > 0:
                old_value = wakeup_manager.wakeup_value
                wakeup_manager.wakeup_value = max(0, wakeup_manager.wakeup_value - wakeup_manager.decay_rate)
                
            assert wakeup_manager.wakeup_value == 4.8  # 5.0 - 0.2 = 4.8
    
    @pytest.mark.asyncio
    @patch('src.mood.mood_manager.mood_manager')
    async def test_angry_state_expiration_in_decay_loop(self, mock_mood_manager, wakeup_manager):
        """测试衰减循环中愤怒状态过期"""
        # 设置过期的愤怒状态
        wakeup_manager.is_angry = True
        wakeup_manager.angry_start_time = time.time() - 400  # 400秒前
        
        # 手动调用衰减循环中的愤怒状态检查逻辑
        current_time = time.time()
        if wakeup_manager.is_angry and current_time - wakeup_manager.angry_start_time >= wakeup_manager.angry_duration:
            wakeup_manager.is_angry = False
            mock_mood_manager.clear_angry_from_wakeup(wakeup_manager.context.stream_id)
        
        assert wakeup_manager.is_angry == False
        mock_mood_manager.clear_angry_from_wakeup.assert_called_once_with("test_chat_123")
    
    @pytest.mark.asyncio
    async def test_start_stop_lifecycle(self, wakeup_manager):
        """测试启动和停止生命周期"""
        # 测试启动
        await wakeup_manager.start()
        assert wakeup_manager._decay_task is not None
        assert not wakeup_manager._decay_task.done()
        
        # 测试停止
        await wakeup_manager.stop()
        assert wakeup_manager._decay_task.cancelled()
    
    @pytest.mark.asyncio
    async def test_disabled_system_start(self, mock_context):
        """测试禁用系统的启动行为"""
        disabled_config = WakeUpSystemConfig(enable=False)
        
        with patch('src.chat.chat_loop.wakeup_manager.global_config') as mock_global_config:
            mock_global_config.wakeup_system = disabled_config
            manager = WakeUpManager(mock_context)
            
            await manager.start()
            assert manager._decay_task is None  # 禁用时不应该创建衰减任务


class TestWakeUpSystemIntegration:
    """唤醒度系统集成测试"""
    
    @patch('src.manager.schedule_manager.schedule_manager')
    @patch('src.mood.mood_manager.mood_manager')
    def test_mixed_message_types(self, mock_mood_manager, mock_schedule_manager):
        """测试混合消息类型的唤醒度累积"""
        mock_schedule_manager.is_sleeping.return_value = True
        
        # 创建配置和管理器
        config = WakeUpSystemConfig(
            enable=True,
            wakeup_threshold=10.0,  # 降低阈值便于测试
            private_message_increment=3.0,
            group_mention_increment=2.0,
            decay_rate=0.2,
            decay_interval=30.0,
            angry_duration=300.0
        )
        
        context = Mock(spec=HfcContext)
        context.stream_id = "test_mixed"
        context.log_prefix = "[MIXED]"
        context.running = True
        
        with patch('src.chat.chat_loop.wakeup_manager.global_config') as mock_global_config:
            mock_global_config.wakeup_system = config
            manager = WakeUpManager(context)
            
            # 发送2条私聊消息 (2 * 3.0 = 6.0)
            manager.add_wakeup_value(is_private_chat=True)
            manager.add_wakeup_value(is_private_chat=True)
            assert manager.wakeup_value == 6.0
            
            # 发送2条群聊艾特消息 (2 * 2.0 = 4.0, 总计10.0)
            manager.add_wakeup_value(is_private_chat=False, is_mentioned=True)
            assert manager.wakeup_value == 8.0
            
            # 最后一条消息触发唤醒
            result = manager.add_wakeup_value(is_private_chat=False, is_mentioned=True)
            assert result == True
            assert manager.is_angry == True
            assert manager.wakeup_value == 0.0
            
            mock_mood_manager.set_angry_from_wakeup.assert_called_once_with("test_mixed")


if __name__ == "__main__":
    # 运行测试
    pytest.main([__file__, "-v"])