aiagent/test_adb_tool.py

#!/usr/bin/env python3
"""
ADB工具验证脚本
用于直接测试 adb_log_tool 是否能正常调用 ADB 命令
"""
import asyncio
import json
import sys
import os

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

from app.services.builtin_tools import adb_log_tool


async def test_adb_devices():
    """测试列出设备"""
    print("=" * 60)
    print("测试 1: 列出连接的设备 (adb devices)")
    print("=" * 60)
    try:
        result = await adb_log_tool(command="devices")
        result_data = json.loads(result)
        print(f"✅ 执行成功")
        print(f"结果:\n{json.dumps(result_data, ensure_ascii=False, indent=2)}")
        return True
    except Exception as e:
        print(f"❌ 执行失败: {str(e)}")
        return False


async def test_adb_logcat_recent():
    """测试获取最近日志"""
    print("\n" + "=" * 60)
    print("测试 2: 获取最近日志 (adb logcat -d -t 10)")
    print("=" * 60)
    try:
        result = await adb_log_tool(
            command="logcat",
            max_lines=10
        )
        result_data = json.loads(result)
        print(f"✅ 执行成功")
        if "error" in result_data:
            print(f"⚠️  返回错误: {result_data['error']}")
        else:
            print(f"日志行数: {result_data.get('line_count', 0)}")
            if result_data.get('logs'):
                print(f"前3行日志预览:")
                for i, log in enumerate(result_data['logs'][:3], 1):
                    print(f"  {i}. {log[:100]}...")
        return True
    except Exception as e:
        print(f"❌ 执行失败: {str(e)}")
        return False


async def test_adb_logcat_with_filter():
    """测试带过滤的日志获取"""
    print("\n" + "=" * 60)
    print("测试 3: 获取错误级别日志 (adb logcat -d *:E -t 5)")
    print("=" * 60)
    try:
        result = await adb_log_tool(
            command="logcat",
            level="E",
            max_lines=5
        )
        result_data = json.loads(result)
        print(f"✅ 执行成功")
        if "error" in result_data:
            print(f"⚠️  返回错误: {result_data['error']}")
        else:
            print(f"错误日志行数: {result_data.get('line_count', 0)}")
            if result_data.get('logs'):
                print(f"错误日志预览:")
                for i, log in enumerate(result_data['logs'][:3], 1):
                    print(f"  {i}. {log[:100]}...")
        return True
    except Exception as e:
        print(f"❌ 执行失败: {str(e)}")
        return False


async def test_adb_shell():
    """测试执行shell命令"""
    print("\n" + "=" * 60)
    print("测试 4: 执行shell命令 (adb shell getprop ro.build.version.release)")
    print("=" * 60)
    try:
        result = await adb_log_tool(
            command="shell",
            filter_tag="getprop ro.build.version.release"
        )
        result_data = json.loads(result)
        print(f"✅ 执行成功")
        if "error" in result_data:
            print(f"⚠️  返回错误: {result_data['error']}")
        else:
            print(f"命令输出:\n{result_data.get('output', '')}")
        return True
    except Exception as e:
        print(f"❌ 执行失败: {str(e)}")
        return False


async def test_invalid_command():
    """测试无效命令"""
    print("\n" + "=" * 60)
    print("测试 5: 测试无效命令 (验证错误处理)")
    print("=" * 60)
    try:
        result = await adb_log_tool(command="invalid_command")
        result_data = json.loads(result)
        if "error" in result_data:
            print(f"✅ 正确返回错误: {result_data['error']}")
            return True
        else:
            print(f"⚠️  应该返回错误但未返回")
            return False
    except Exception as e:
        print(f"❌ 执行失败: {str(e)}")
        return False


async def main():
    """主测试函数"""
    print("\n" + "🔧 ADB工具验证测试")
    print("=" * 60)
    print("此脚本将测试 adb_log_tool 的各种功能")
    print("请确保:")
    print("  1. 已安装 Android SDK Platform Tools")
    print("  2. adb 命令在 PATH 中")
    print("  3. 已连接 Android 设备或启动模拟器")
    print("=" * 60)
    print("\n开始测试...\n")
    
    results = []
    
    # 运行所有测试
    results.append(("列出设备", await test_adb_devices()))
    results.append(("获取最近日志", await test_adb_logcat_recent()))
    results.append(("获取错误日志", await test_adb_logcat_with_filter()))
    results.append(("执行shell命令", await test_adb_shell()))
    results.append(("错误处理", await test_invalid_command()))
    
    # 汇总结果
    print("\n" + "=" * 60)
    print("测试结果汇总")
    print("=" * 60)
    passed = sum(1 for _, result in results if result)
    total = len(results)
    
    for test_name, result in results:
        status = "✅ 通过" if result else "❌ 失败"
        print(f"{status} - {test_name}")
    
    print(f"\n总计: {passed}/{total} 测试通过")
    
    if passed == total:
        print("\n🎉 所有测试通过！ADB工具工作正常。")
        return 0
    else:
        print(f"\n⚠️  有 {total - passed} 个测试失败，请检查:")
        print("  1. ADB 是否正确安装")
        print("  2. 设备是否已连接 (运行 'adb devices' 检查)")
        print("  3. 设备是否已启用 USB 调试")
        return 1


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