TermControlAgent/test_logging.py

#!/usr/bin/env python3
"""
测试结构化日志系统
"""
import asyncio
import os
from app.utils.log import get_enhanced_logger, LogLevel, set_request_context
from app.services.enhanced_adb_service import EnhancedAdbService


async def test_structured_logging():
    """测试结构化日志系统"""
    print("=== 测试结构化日志系统 ===")
    
    # 设置请求上下文
    set_request_context("test-request-123", "test-user-456", "test-session-789")
    
    # 获取日志记录器
    logger = get_enhanced_logger("test_module", LogLevel.DEBUG)
    
    # 测试不同级别的日志
    logger.trace("这是TRACE级别的日志", extra_data="trace_info")
    logger.debug("这是DEBUG级别的日志", debug_info="debug_data")
    logger.info("这是INFO级别的日志", user_action="login")
    logger.warning("这是WARNING级别的日志", warning_type="deprecated")
    logger.error("这是ERROR级别的日志", error_code=500)
    logger.critical("这是CRITICAL级别的日志", critical_reason="system_failure")
    
    # 测试异常日志
    try:
        raise ValueError("测试异常")
    except Exception as e:
        logger.exception("捕获到异常", error_type=type(e).__name__)
    
    print("结构化日志测试完成")


async def test_adb_service_logging():
    """测试ADB服务的日志记录"""
    print("\n=== 测试ADB服务日志记录 ===")
    
    # 设置请求上下文
    set_request_context("adb-test-request", "adb-test-user")
    
    # 创建ADB服务实例
    service = EnhancedAdbService()
    
    try:
        # 测试获取服务器信息
        logger = get_enhanced_logger("test_adb_service")
        logger.info("开始测试ADB服务")
        
        # 注意：这里只是测试日志，实际ADB连接可能失败
        logger.debug("尝试获取ADB服务器信息")
        
        # 模拟一些操作
        logger.info("模拟设备列表获取", device_count=3)
        logger.debug("模拟Shell命令执行", command="ls /sdcard", device_serial="test_device")
        
    except Exception as e:
        logger.error("ADB服务测试失败", error=str(e))
    
    print("ADB服务日志测试完成")


def test_log_file_creation():
    """测试日志文件创建"""
    print("\n=== 测试日志文件创建 ===")
    
    # 检查logs目录是否存在
    if os.path.exists("logs"):
        print("✓ logs目录已存在")
        
        # 列出日志文件
        log_files = [f for f in os.listdir("logs") if f.endswith('.log')]
        print(f"发现日志文件: {log_files}")
        
        # 显示最新的日志内容
        if log_files:
            latest_log = max(log_files, key=lambda x: os.path.getmtime(os.path.join("logs", x)))
            log_path = os.path.join("logs", latest_log)
            print(f"最新日志文件: {latest_log}")
            
            # 读取最后几行
            try:
                with open(log_path, 'r', encoding='utf-8') as f:
                    lines = f.readlines()
                    print(f"日志文件总行数: {len(lines)}")
                    if lines:
                        print("最后5行日志:")
                        for line in lines[-5:]:
                            print(f"  {line.strip()}")
            except Exception as e:
                print(f"读取日志文件失败: {e}")
    else:
        print("✗ logs目录不存在")


async def main():
    """主测试函数"""
    print("开始测试结构化日志系统...")
    
    # 测试结构化日志
    await test_structured_logging()
    
    # 测试ADB服务日志
    await test_adb_service_logging()
    
    # 测试日志文件创建
    test_log_file_creation()
    
    print("\n=== 测试完成 ===")
    print("请检查logs目录中的日志文件查看结构化日志输出")


if __name__ == "__main__":
    asyncio.run(main())