TermControlAgent/tests/test_input_output_logging.py

#!/usr/bin/env python3
"""
输入输出日志记录测试
演示各种输入输出操作的日志记录
"""

import asyncio
import json
from app.utils.structured_log import get_structured_logger, LogLevel
from app.services.auto_discovery_adb_service import AutoDiscoveryAdbService
from app.services.ssh_service import SSHService

# 创建日志记录器
logger = get_structured_logger("InputOutputTest", LogLevel.DEBUG)

async def test_adb_input_output():
    """测试ADB输入输出日志"""
    logger.info("开始测试ADB输入输出日志")
    
    # 创建ADB服务
    adb_service = AutoDiscoveryAdbService()
    
    # 模拟设备ID
    device_id = "test_device_001"
    
    try:
        # 测试文本输入
        logger.info("测试文本输入操作", device_id=device_id, text="Hello World")
        
        # 模拟输入文本
        result = await adb_service.input_text(device_id, "Hello World", clear_first=True)
        logger.info("文本输入结果", 
                   device_id=device_id, 
                   success=result.success, 
                   output_length=len(result.output) if result.output else 0)
        
        # 测试点击操作
        logger.info("测试点击操作", device_id=device_id, x=100, y=200)
        
        # 模拟点击
        click_result = await adb_service.click_device(device_id, 100, 200)
        logger.info("点击操作结果", 
                   device_id=device_id, 
                   success=click_result.success,
                   coordinates={"x": 100, "y": 200})
        
        # 测试截图操作
        logger.info("测试截图操作", device_id=device_id)
        
        # 模拟截图
        screenshot_result = await adb_service.screenshot(device_id)
        logger.info("截图操作结果", 
                   device_id=device_id, 
                   success=screenshot_result.success,
                   output_size=len(screenshot_result.output) if screenshot_result.output else 0)
        
        # 测试日志获取
        logger.info("测试日志获取", device_id=device_id, package_name="com.example.app")
        
        # 模拟获取日志
        logcat_result = await adb_service.get_logcat(device_id, "com.example.app", "D", 50)
        logger.info("日志获取结果", 
                   device_id=device_id, 
                   success=logcat_result.success,
                   log_lines=len(logcat_result.output.splitlines()) if logcat_result.output else 0)
        
    except Exception as e:
        logger.error("ADB输入输出测试失败", device_id=device_id, error=str(e))
    
    logger.info("ADB输入输出测试完成")

async def test_ssh_input_output():
    """测试SSH输入输出日志"""
    logger.info("开始测试SSH输入输出日志")
    
    # 创建SSH服务
    ssh_service = SSHService()
    
    # 模拟设备ID
    device_id = "ssh_device_001"
    
    try:
        # 测试SSH命令执行
        logger.info("测试SSH命令执行", device_id=device_id, command="ls -la")
        
        # 模拟SSH连接配置
        ssh_config = {
            "hostname": "192.168.1.100",
            "port": 22,
            "username": "testuser",
            "password": "testpass"
        }
        
        # 模拟命令执行
        exec_result = await ssh_service.execute_command(
            device_id=device_id,
            command="ls -la",
            working_directory="/home/testuser",
            timeout=30
        )
        
        logger.info("SSH命令执行结果", 
                   device_id=device_id, 
                   success=exec_result.success,
                   exit_code=exec_result.exit_code,
                   stdout_length=len(exec_result.stdout),
                   stderr_length=len(exec_result.stderr),
                   execution_time=exec_result.execution_time)
        
        # 测试文件上传
        logger.info("测试文件上传", device_id=device_id, local_path="/tmp/test.txt", remote_path="/home/testuser/test.txt")
        
        # 模拟文件上传
        upload_result = await ssh_service.upload_file(device_id, "/tmp/test.txt", "/home/testuser/test.txt")
        logger.info("文件上传结果", 
                   device_id=device_id, 
                   success=upload_result.get("success", False),
                   message=upload_result.get("message", ""))
        
        # 测试文件下载
        logger.info("测试文件下载", device_id=device_id, remote_path="/home/testuser/output.txt", local_path="/tmp/downloaded.txt")
        
        # 模拟文件下载
        download_result = await ssh_service.download_file(device_id, "/home/testuser/output.txt", "/tmp/downloaded.txt")
        logger.info("文件下载结果", 
                   device_id=device_id, 
                   success=download_result.get("success", False),
                   message=download_result.get("message", ""))
        
    except Exception as e:
        logger.error("SSH输入输出测试失败", device_id=device_id, error=str(e))
    
    logger.info("SSH输入输出测试完成")

async def test_serial_input_output():
    """测试串口输入输出日志"""
    logger.info("开始测试串口输入输出日志")
    
    try:
        from app.utils.serial_utils import SerialManager
        
        # 创建串口管理器
        serial_manager = SerialManager()
        
        # 模拟串口配置
        port_config = {
            "port": "COM3",
            "baudrate": 115200,
            "timeout": 1
        }
        
        logger.info("测试串口连接", port_config=port_config)
        
        # 模拟串口操作
        try:
            # 模拟发送命令
            command = "AT+CGSN"
            logger.info("发送串口命令", command=command, port=port_config["port"])
            
            # 模拟接收响应
            response = "123456789012345"
            logger.info("接收串口响应", 
                       command=command, 
                       response=response, 
                       response_length=len(response),
                       port=port_config["port"])
            
            # 模拟错误情况
            error_command = "AT+INVALID"
            logger.warning("发送无效命令", command=error_command, port=port_config["port"])
            
            error_response = "ERROR"
            logger.warning("接收错误响应", 
                          command=error_command, 
                          response=error_response,
                          port=port_config["port"])
            
        except Exception as e:
            logger.error("串口操作失败", port=port_config["port"], error=str(e))
            
    except ImportError:
        logger.warning("串口模块不可用，跳过串口测试")
    except Exception as e:
        logger.error("串口输入输出测试失败", error=str(e))
    
    logger.info("串口输入输出测试完成")

async def test_websocket_input_output():
    """测试WebSocket输入输出日志"""
    logger.info("开始测试WebSocket输入输出日志")
    
    try:

        from app.core.websocket.client import WebSocketClient
        
        # 模拟WebSocket客户端
        client_name = "test_websocket_client"
        logger.info("创建WebSocket客户端", client_name=client_name)
        
        # 模拟WebSocket消息
        message = {
            "type": "command",
            "data": "test_message",
            "timestamp": "2025-08-12T09:45:00Z"
        }
        
        logger.info("发送WebSocket消息", 
                   client_name=client_name, 
                   message_type=message["type"],
                   message_data=message["data"])
        
        # 模拟接收响应
        response = {
            "type": "response",
            "data": "message_received",
            "timestamp": "2025-08-12T09:45:01Z"
        }
        
        logger.info("接收WebSocket响应", 
                   client_name=client_name, 
                   response_type=response["type"],
                   response_data=response["data"])
        
        # 模拟连接状态变化
        logger.info("WebSocket连接状态", client_name=client_name, status="connected")
        
        # 模拟心跳消息
        heartbeat_msg = {"type": "heartbeat", "timestamp": "2025-08-12T09:45:02Z"}
        logger.debug("发送心跳消息", client_name=client_name, heartbeat_data=heartbeat_msg)
        
    except ImportError:
        logger.warning("WebSocket模块不可用，跳过WebSocket测试")
    except Exception as e:
        logger.error("WebSocket输入输出测试失败", error=str(e))
    
    logger.info("WebSocket输入输出测试完成")

async def main():
    """主测试函数"""
    logger.info("开始输入输出日志记录测试")
    
    # 设置请求上下文
    from app.utils.structured_log import set_request_context
    set_request_context("test_request_001", "test_user", "test_session")
    
    try:
        # 测试各种输入输出操作
        await test_adb_input_output()
        await test_ssh_input_output()
        await test_serial_input_output()
        await test_websocket_input_output()
        
        logger.info("所有输入输出日志记录测试完成")
        
    except Exception as e:
        logger.error("输入输出日志记录测试失败", error=str(e))
    
    finally:
        # 清除上下文
        from app.utils.structured_log import clear_log_context
        clear_log_context()

if __name__ == "__main__":
    # 运行测试
    asyncio.run(main())