TermControlAgent/tests/test_websocket_device_status_check.py

#!/usr/bin/env python3
"""
WebSocket设备状态检查功能单元测试
专门测试_async_check_current_device_status方法
"""
import asyncio
import sys
import os
from unittest.mock import AsyncMock, MagicMock, patch

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

from app.core.websocket.manager import WebSocketManager
from app.core.device.manager import DeviceManager
from app.utils.structured_log import get_structured_logger, LogLevel

logger = get_structured_logger(__name__, LogLevel.INFO)

async def test_device_status_check_basic():
    """测试设备状态检查基本功能"""
    logger.info("=== 测试设备状态检查基本功能 ===")
    
    websocket_manager = WebSocketManager()
    device_manager = DeviceManager()
    
    try:
        # Mock设备管理器方法
        with patch.object(device_manager, 'get_all_devices_unified') as mock_get_devices, \
             patch.object(device_manager, 'handle_auto_discovered_device_event') as mock_handle_event:
            
            # 设置Mock返回值
            mock_get_devices.return_value = [
                {
                    "device_id": "device_001",
                    "status": "device",
                    "source": "auto_discovered",
                    "device_info": {"model": "Test Device 1"}
                },
                {
                    "device_id": "device_002",
                    "status": "offline",
                    "source": "auto_discovered",
                    "device_info": {"model": "Test Device 2"}
                }
            ]
            
            # 调用设备状态检查方法
            await websocket_manager._async_check_current_device_status(
                "test_client", device_manager
            )
            
            # 验证结果
            mock_get_devices.assert_called_once()
            mock_handle_event.assert_called_once_with(
                device_id="device_001",
                status="device",
                device_info={"model": "Test Device 1"}
            )
            
            logger.info("✅ 设备状态检查基本功能测试通过")
            return True
            
    except Exception as e:
        logger.error(f"❌ 设备状态检查基本功能测试失败: {e}")
        return False
    finally:
        await websocket_manager.cleanup()
        await device_manager.cleanup()

async def test_device_status_check_registered_devices():
    """测试已注册设备的跳过逻辑"""
    logger.info("=== 测试已注册设备的跳过逻辑 ===")
    
    websocket_manager = WebSocketManager()
    device_manager = DeviceManager()
    
    try:
        # Mock设备管理器方法
        with patch.object(device_manager, 'get_all_devices_unified') as mock_get_devices, \
             patch.object(device_manager, 'handle_auto_discovered_device_event') as mock_handle_event:
            
            # 设置Mock返回值 - 包含已注册设备
            mock_get_devices.return_value = [
                {
                    "device_id": "device_001",
                    "status": "device",
                    "source": "auto_discovered",  # 自动发现设备
                    "device_info": {"model": "Test Device 1"}
                },
                {
                    "device_id": "device_002",
                    "status": "device",
                    "source": "registered",  # 已注册设备，应该被跳过
                    "device_info": {"model": "Test Device 2"}
                }
            ]
            
            # 调用设备状态检查方法
            await websocket_manager._async_check_current_device_status(
                "test_client", device_manager
            )
            
            # 验证结果 - 只处理自动发现的设备
            mock_get_devices.assert_called_once()
            mock_handle_event.assert_called_once_with(
                device_id="device_001",
                status="device",
                device_info={"model": "Test Device 1"}
            )
            
            logger.info("✅ 已注册设备跳过逻辑测试通过")
            return True
            
    except Exception as e:
        logger.error(f"❌ 已注册设备跳过逻辑测试失败: {e}")
        return False
    finally:
        await websocket_manager.cleanup()
        await device_manager.cleanup()

async def test_device_status_check_empty_list():
    """测试空设备列表处理"""
    logger.info("=== 测试空设备列表处理 ===")
    
    websocket_manager = WebSocketManager()
    device_manager = DeviceManager()
    
    try:
        # Mock设备管理器方法
        with patch.object(device_manager, 'get_all_devices_unified') as mock_get_devices, \
             patch.object(device_manager, 'handle_auto_discovered_device_event') as mock_handle_event:
            
            # 设置Mock返回值 - 空列表
            mock_get_devices.return_value = []
            
            # 调用设备状态检查方法
            await websocket_manager._async_check_current_device_status(
                "test_client", device_manager
            )
            
            # 验证结果
            mock_get_devices.assert_called_once()
            mock_handle_event.assert_not_called()
            
            logger.info("✅ 空设备列表处理测试通过")
            return True
            
    except Exception as e:
        logger.error(f"❌ 空设备列表处理测试失败: {e}")
        return False
    finally:
        await websocket_manager.cleanup()
        await device_manager.cleanup()

async def test_device_status_check_error_handling():
    """测试错误处理"""
    logger.info("=== 测试错误处理 ===")
    
    websocket_manager = WebSocketManager()
    device_manager = DeviceManager()
    
    try:
        # Mock设备管理器方法 - 抛出异常
        with patch.object(device_manager, 'get_all_devices_unified', side_effect=Exception("测试异常")):
            
            # 调用设备状态检查方法 - 应该能处理异常
            await websocket_manager._async_check_current_device_status(
                "test_client", device_manager
            )
            
            logger.info("✅ 错误处理测试通过")
            return True
            
    except Exception as e:
        logger.error(f"❌ 错误处理测试失败: {e}")
        return False
    finally:
        await websocket_manager.cleanup()
        await device_manager.cleanup()

async def main():
    """主测试函数"""
    logger.info("开始WebSocket设备状态检查功能单元测试")
    logger.info("=" * 50)
    
    test_cases = [
        ("设备状态检查基本功能", test_device_status_check_basic),
        ("已注册设备跳过逻辑", test_device_status_check_registered_devices),
        ("空设备列表处理", test_device_status_check_empty_list),
        ("错误处理", test_device_status_check_error_handling),
    ]
    
    passed = 0
    total = len(test_cases)
    
    for test_name, test_func in test_cases:
        logger.info(f"\n🧪 运行测试: {test_name}")
        try:
            result = await test_func()
            if result:
                passed += 1
                logger.info(f"✅ {test_name} - 通过")
            else:
                logger.error(f"❌ {test_name} - 失败")
        except Exception as e:
            logger.error(f"❌ {test_name} - 异常: {e}")
    
    logger.info("\n" + "=" * 50)
    logger.info(f"测试结果: {passed}/{total} 通过")
    logger.info(f"成功率: {(passed/total*100):.1f}%")
    
    if passed == total:
        logger.info("🎉 所有测试通过！设备状态检查功能正常")
        return True
    else:
        logger.error("❌ 部分测试失败，请检查实现")
        return False

if __name__ == "__main__":
    try:
        success = asyncio.run(main())
        sys.exit(0 if success else 1)
    except KeyboardInterrupt:
        logger.info("\n⏹️  测试被用户中断")
        sys.exit(1)
    except Exception as e:
        logger.error(f"\n❌ 测试执行失败: {e}")
        sys.exit(1)