TermControlAgent/tests/test_shell_command_wait_time.py

"""
测试Shell命令执行等待时间功能
"""
import asyncio
import time
import pytest
from app.services.shell_command_service import ShellCommandService
from app.models.adb_models import (
    ShellCommandTask, DeviceShellTask, UnifiedShellCommandRequest
)

@pytest.mark.asyncio
async def test_shell_command_wait_time():
    """测试Shell命令执行等待时间功能"""
    service = ShellCommandService()
    
    # 创建测试请求
    request = UnifiedShellCommandRequest(
        tasks=[
            DeviceShellTask(
                device_id="test_device_001",
                commands=[
                    ShellCommandTask(
                        command="echo 'test1'",
                        timeout=30,
                        wait_time=1.0,  # 等待1秒
                        description="测试命令1"
                    ),
                    ShellCommandTask(
                        command="echo 'test2'",
                        timeout=30,
                        wait_time=0.5,  # 等待0.5秒
                        description="测试命令2"
                    ),
                    ShellCommandTask(
                        command="echo 'test3'",
                        timeout=30,
                        wait_time=2.0,  # 等待2秒
                        description="测试命令3"
                    )
                ],
                sequential=True
            )
        ],
        parallel_devices=False,
        parallel_commands=False
    )
    
    # 记录开始时间
    start_time = time.time()
    
    # 执行命令
    result = await service.execute_unified_shell_commands(request)
    
    # 记录结束时间
    end_time = time.time()
    total_time = end_time - start_time
    
    # 验证结果
    assert result["success"] is True
    assert "data" in result
    assert result["data"]["total_devices"] == 1
    assert result["data"]["total_commands"] == 3
    
    # 验证等待时间（总时间应该至少包含所有等待时间）
    # 3个命令的等待时间总和：1.0 + 0.5 + 2.0 = 3.5秒
    # 加上执行时间，总时间应该大于3.5秒
    assert total_time >= 3.5, f"总执行时间 {total_time} 秒应该大于等于3.5秒"
    
    # 验证设备结果
    device_results = result["data"]["device_results"]
    assert len(device_results) == 1
    
    device_result = device_results[0]
    assert device_result["device_id"] == "test_device_001"
    assert device_result["total_commands"] == 3
    assert device_result["success_commands"] == 3
    
    # 验证命令结果
    command_results = device_result["results"]
    assert len(command_results) == 3
    
    # 验证每个命令的等待时间
    for i, cmd_result in enumerate(command_results):
        expected_wait_times = [1.0, 0.5, 2.0]
        assert cmd_result["wait_time"] == expected_wait_times[i]
        assert cmd_result["command"] == f"echo 'test{i+1}'"

@pytest.mark.asyncio
async def test_shell_command_no_wait_time():
    """测试Shell命令执行无等待时间功能"""
    service = ShellCommandService()
    
    # 创建测试请求（无等待时间）
    request = UnifiedShellCommandRequest(
        tasks=[
            DeviceShellTask(
                device_id="test_device_002",
                commands=[
                    ShellCommandTask(
                        command="echo 'test1'",
                        timeout=30,
                        wait_time=0.0,  # 无等待时间
                        description="测试命令1"
                    ),
                    ShellCommandTask(
                        command="echo 'test2'",
                        timeout=30,
                        wait_time=0.0,  # 无等待时间
                        description="测试命令2"
                    )
                ],
                sequential=True
            )
        ],
        parallel_devices=False,
        parallel_commands=False
    )
    
    # 记录开始时间
    start_time = time.time()
    
    # 执行命令
    result = await service.execute_unified_shell_commands(request)
    
    # 记录结束时间
    end_time = time.time()
    total_time = end_time - start_time
    
    # 验证结果
    assert result["success"] is True
    assert "data" in result
    assert result["data"]["total_devices"] == 1
    assert result["data"]["total_commands"] == 2
    
    # 验证无等待时间时执行时间应该很短
    assert total_time < 1.0, f"无等待时间时总执行时间 {total_time} 秒应该小于1秒"
    
    # 验证设备结果
    device_results = result["data"]["device_results"]
    assert len(device_results) == 1
    
    device_result = device_results[0]
    assert device_result["device_id"] == "test_device_002"
    assert device_result["total_commands"] == 2
    assert device_result["success_commands"] == 2
    
    # 验证命令结果
    command_results = device_result["results"]
    assert len(command_results) == 2
    
    # 验证每个命令的等待时间
    for cmd_result in command_results:
        assert cmd_result["wait_time"] == 0.0

if __name__ == "__main__":
    # 运行测试
    asyncio.run(test_shell_command_wait_time())
    asyncio.run(test_shell_command_no_wait_time())
    print("所有测试通过！")