396 lines
15 KiB
Python
396 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
import os
|
||
import argparse
|
||
import sys
|
||
import time
|
||
import json
|
||
import re
|
||
import logging
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
from dataclasses import dataclass, asdict
|
||
|
||
# 找到项目根目录
|
||
ROOT = Path(__file__).resolve().parent
|
||
# 确保根目录在Python路径中
|
||
if str(ROOT) not in sys.path:
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
# 导入统一的日志配置模块
|
||
from logging_config import setup_logging, get_logger
|
||
from config_loader import load_config as load_runtime_config
|
||
from utils_android.device_config import ensure_android_wireless_connected, get_android_device_serial
|
||
|
||
# 导入 DroidBot
|
||
from DroidBot.droidbot import DroidBot
|
||
from DroidBot.exceptions import ADBException, AppCrashException, ExplorationStuckException, AppNeedUpdateException, AppLaunchErrorException
|
||
from DroidBot.guiagent_core.decision_maker import GuiAgentDecisionMaker
|
||
|
||
|
||
from result_codes import ResultCode, ErrorCategory, ErrorInfo, STUCK_TO_ERROR
|
||
|
||
EXIT_ERROR_GENERAL = ResultCode.ERROR_GENERAL
|
||
EXIT_SUCCESS = ResultCode.SUCCESS
|
||
EXIT_ERROR_USER = ResultCode.ERROR_USER
|
||
EXIT_ADB_ERROR = ResultCode.ADB_ERROR
|
||
EXIT_NETWORK_ERROR = ResultCode.NETWORK_ERROR
|
||
EXIT_SIMULATOR_ERROR = ResultCode.SIMULATOR_ERROR
|
||
EXIT_EXPLORATION_STUCK = ResultCode.EXPLORATION_STUCK
|
||
EXIT_APP_CRASH_ERROR = ResultCode.APP_CRASH
|
||
EXIT_APP_NEED_UPDATE = ResultCode.APP_NEED_UPDATE
|
||
EXIT_APP_LAUNCH_ERROR = ResultCode.APP_LAUNCH_ERROR
|
||
|
||
STUCK_REASON_TO_ERROR_TYPE = {k: v[0].name for k, v in STUCK_TO_ERROR.items()}
|
||
|
||
|
||
@dataclass
|
||
class TaskResult:
|
||
"""单次测试任务的执行结果"""
|
||
status: str # "SUCCESS" | "FAILED" | "INTERRUPTED"
|
||
exit_code: int # 退出码
|
||
error_reason: str # 错误原因（成功时为空字符串）
|
||
droidbot_steps: int # DroidBot 执行步数
|
||
guiagent_steps: int # GuiAgent 执行步数
|
||
total_steps: int # 总步数
|
||
duration_seconds: float # 执行耗时（秒）
|
||
guiagent_message: str = "" # GuiAgent 登录/注册失败消息，用于worker.report汇报
|
||
login_count: int = 0 # 进入登录场景的次数
|
||
register_count: int = 0 # 进入注册场景的次数
|
||
stuck_reason_code: int = 0 # 卡住原因代码 (0=无/其他, 1=登录注册, 2=启动异常, 3=正常, 4-12=其他原因)
|
||
num_nodes: int = 0 # UTG 节点数
|
||
num_reached_activities: int = 0 # UTG 已到达 Activity 数
|
||
app_num_total_activities: int = 0 # 应用总 Activity 数
|
||
|
||
|
||
def load_config(json_path: str = None):
|
||
"""从JSON文件加载配置"""
|
||
try:
|
||
return load_runtime_config(json_path)
|
||
except Exception as e:
|
||
target = json_path or "layered autool config"
|
||
print(f"[ERROR] 无法从 {target} 加载配置: {e}")
|
||
raise e
|
||
|
||
|
||
class TeeOutput:
|
||
"""将输出同时写入控制台和文件，包括标准输出、标准错误和所有logging输出"""
|
||
def __init__(self, file_path):
|
||
self.file_path = file_path
|
||
self.file = open(file_path, 'w', encoding='utf-8', buffering=1) # 行缓冲
|
||
self.stdout = sys.stdout
|
||
self.stderr = sys.stderr
|
||
self.log_handler = None
|
||
|
||
def write(self, message):
|
||
try:
|
||
self.stdout.write(message)
|
||
self.file.write(message)
|
||
except Exception:
|
||
pass # 忽略写入错误，防止程序崩溃
|
||
|
||
def flush(self):
|
||
try:
|
||
self.stdout.flush()
|
||
self.file.flush()
|
||
except Exception:
|
||
pass
|
||
|
||
def fileno(self):
|
||
"""返回文件描述符，使某些库兼容"""
|
||
return self.stdout.fileno()
|
||
|
||
def isatty(self):
|
||
"""检查是否为终端"""
|
||
return self.stdout.isatty()
|
||
|
||
def setup_logging(self):
|
||
"""设置根日志记录器的文件处理器，捕获所有logging模块输出"""
|
||
import logging
|
||
self.log_handler = logging.FileHandler(self.file_path, mode='a', encoding='utf-8')
|
||
self.log_handler.setLevel(logging.DEBUG) # 捕获所有级别
|
||
formatter = logging.Formatter('%(asctime)s %(name)-12s %(levelname)-8s %(message)s')
|
||
self.log_handler.setFormatter(formatter)
|
||
logging.getLogger().addHandler(self.log_handler)
|
||
|
||
def close(self):
|
||
"""清理资源"""
|
||
import logging
|
||
if self.log_handler:
|
||
logging.getLogger().removeHandler(self.log_handler)
|
||
self.log_handler.close()
|
||
self.log_handler = None
|
||
if self.file and not self.file.closed:
|
||
self.file.close()
|
||
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||
self.close()
|
||
return False
|
||
|
||
|
||
class TaskRunner:
|
||
"""任务执行器：负责构建路径、直接调用 DroidBot 模块并执行"""
|
||
|
||
def __init__(self, config: dict):
|
||
self.config = config
|
||
|
||
def _build_droidbot_kwargs(self, package_name: str, app_name: str, output_dir: str, **extra_kwargs) -> dict:
|
||
"""构建 DroidBot 初始化所需的参数"""
|
||
# 统一路径格式为正斜杠
|
||
output_dir = output_dir.replace('\\', '/')
|
||
|
||
device_serial = get_android_device_serial(self.config)
|
||
ensure_android_wireless_connected(self.config, device_serial)
|
||
|
||
mumu_config = self.config.get("mumu", {})
|
||
kwargs = {
|
||
'package_name': package_name,
|
||
'app_name': app_name,
|
||
'device_serial': device_serial,
|
||
'is_emulator': self.config.get("IS_EMULATOR", False),
|
||
'output_dir': output_dir,
|
||
'policy_name': self.config.get("POLICY", "memory_guided"),
|
||
'random_input': False,
|
||
'event_interval': 0,
|
||
'timeout': self.config.get("TIMEOUT", 3600),
|
||
'event_count': 10000,
|
||
'cv_mode': self.config.get("CV_MODE", False),
|
||
'debug_mode': False,
|
||
'keep_app': self.config.get("KEEP_APP", True),
|
||
'keep_env': False,
|
||
'profiling_method': None,
|
||
'grant_perm': False,
|
||
'enable_accessibility_hard': False,
|
||
'humanoid': None,
|
||
'ignore_ad': False,
|
||
'replay_output': None,
|
||
'enable_guiagent': self.config.get("ENABLE_GUIAGENT", False),
|
||
'platform': "android",
|
||
'pcap_callback': extra_kwargs.get('pcap_callback'),
|
||
'mumu_manager_path': mumu_config.get("manager_path", ""),
|
||
'mumu_vm_index': mumu_config.get("default_vm_index", 2),
|
||
}
|
||
return kwargs
|
||
|
||
def run_task(self, package_name: str, app_name: str, output_dir: str, **kwargs) -> TaskResult:
|
||
"""执行单个测试任务，返回 TaskResult"""
|
||
print(f"\n{'='*50}")
|
||
print(f"[INFO] 准备测试应用包名: {package_name}")
|
||
|
||
start_time = time.time()
|
||
ret_code = EXIT_ERROR_GENERAL
|
||
status = "FAILED"
|
||
error_reason = ""
|
||
|
||
# 重置全局步数计数器，确保每轮任务独立统计
|
||
GuiAgentDecisionMaker.total_steps = 0
|
||
|
||
# 初始化步数统计，确保 finally 块中能访问到
|
||
droidbot_steps = 0
|
||
guiagent_steps = 0
|
||
total_steps = 0
|
||
user_interrupted = False # 追踪用户中断，用于向上传递
|
||
|
||
os.makedirs(output_dir, exist_ok=True)
|
||
print(f"[INFO] 输出目录已创建: {output_dir}")
|
||
|
||
# 将 **kwargs (如 pcap_callback) 合并到构建参数中
|
||
droidbot_kwargs = self._build_droidbot_kwargs(package_name, app_name, output_dir)
|
||
enable_app_block = kwargs.pop('enable_app_block', False)
|
||
droidbot_kwargs.update(kwargs)
|
||
droidbot_kwargs['enable_app_block'] = enable_app_block
|
||
if enable_app_block:
|
||
droidbot_kwargs['timeout'] = self.config.get("BLOCK_TIMEOUT", 1000)
|
||
print(f"[DEBUG] DroidBot 初始化参数: {droidbot_kwargs}")
|
||
|
||
cmd_log_path = os.path.join(output_dir, "cmd_output.txt")
|
||
|
||
print(f"[INFO] 开始运行DroidBot")
|
||
print(f"{'='*50}\n")
|
||
|
||
# 设置输出重定向
|
||
tee = TeeOutput(cmd_log_path)
|
||
old_stdout = sys.stdout
|
||
old_stderr = sys.stderr
|
||
sys.stdout = tee
|
||
sys.stderr = tee
|
||
tee.setup_logging() # 设置日志处理器，捕获所有logging模块输出
|
||
|
||
try:
|
||
# 直接创建并运行DroidBot实例
|
||
droidbot = DroidBot(**droidbot_kwargs)
|
||
droidbot.start()
|
||
ret_code = EXIT_SUCCESS
|
||
status = "SUCCESS"
|
||
error_reason = ""
|
||
|
||
# 统计步数
|
||
droidbot_steps = droidbot.input_manager.total_exploring_steps
|
||
guiagent_steps = GuiAgentDecisionMaker.total_steps
|
||
total_steps = droidbot_steps + guiagent_steps
|
||
|
||
print(f"\n[SUCCESS] 测试任务完成: {app_name}")
|
||
|
||
except KeyboardInterrupt:
|
||
print(f"\n[WARN] 用户手动中断测试: {package_name}")
|
||
ret_code = EXIT_ERROR_USER
|
||
status = "INTERRUPTED"
|
||
error_reason = "用户手动中断"
|
||
user_interrupted = True # 标记需要向上传递中断
|
||
except ADBException as e:
|
||
import traceback
|
||
print(f"\n[ERROR] ADB 断联: {e}")
|
||
print(traceback.format_exc())
|
||
ret_code = EXIT_ADB_ERROR
|
||
status = "FAILED"
|
||
error_reason = f"ADB 断联: {e}"
|
||
except AppCrashException as e:
|
||
import traceback
|
||
print(f"\n[ERROR] 应用闪退: {e}")
|
||
print(traceback.format_exc())
|
||
ret_code = EXIT_APP_CRASH_ERROR
|
||
status = "FAILED"
|
||
error_reason = f"应用闪退: {e}"
|
||
except AppNeedUpdateException as e:
|
||
import traceback
|
||
print(f"\n[ERROR] 应用需更新: {e}")
|
||
print(traceback.format_exc())
|
||
ret_code = EXIT_APP_NEED_UPDATE
|
||
status = "FAILED"
|
||
error_reason = f"应用需更新: {e}"
|
||
except AppLaunchErrorException as e:
|
||
import traceback
|
||
print(f"\n[ERROR] 启动异常: {e}")
|
||
print(traceback.format_exc())
|
||
ret_code = EXIT_APP_LAUNCH_ERROR
|
||
status = "FAILED"
|
||
error_reason = f"启动异常: {e}"
|
||
except ExplorationStuckException as e:
|
||
import traceback
|
||
print(f"\n[ERROR] 探索停滞: {e}")
|
||
print(traceback.format_exc())
|
||
ret_code = EXIT_EXPLORATION_STUCK
|
||
status = "FAILED"
|
||
error_reason = f"探索停滞: {e}"
|
||
except Exception as e:
|
||
import traceback
|
||
print(f"\n[ERROR] DroidBot 执行异常: {e}")
|
||
print(traceback.format_exc())
|
||
ret_code = EXIT_ERROR_GENERAL
|
||
status = "FAILED"
|
||
error_reason = f"执行异常: {e}"
|
||
finally:
|
||
# 尝试获取步数（即使异常也尽量获取）
|
||
try:
|
||
guiagent_steps = GuiAgentDecisionMaker.total_steps
|
||
except Exception:
|
||
pass
|
||
total_steps = droidbot_steps + guiagent_steps
|
||
|
||
# 读取 GuiAgent 登录/注册失败消息
|
||
guiagent_message = ""
|
||
login_count = 0
|
||
register_count = 0
|
||
stuck_reason_code = 0
|
||
try:
|
||
if droidbot and droidbot.input_manager and droidbot.input_manager.policy:
|
||
guiagent_message = getattr(droidbot.input_manager.policy, 'guiagent_message', '') or ''
|
||
login_count = getattr(droidbot.input_manager.policy, 'login_count', 0) or 0
|
||
register_count = getattr(droidbot.input_manager.policy, 'register_count', 0) or 0
|
||
stuck_reason_code = getattr(droidbot.input_manager.policy, 'stuck_reason_code', 0) or 0
|
||
except Exception:
|
||
pass
|
||
|
||
# 解析 UTG 文件
|
||
num_nodes = 0
|
||
num_reached_activities = 0
|
||
app_num_total_activities = 0
|
||
try:
|
||
from utils_android.Manager.utg_parser import parse_utg_js
|
||
utg_path = os.path.join(output_dir, 'utg.js')
|
||
if os.path.exists(utg_path):
|
||
utg_result = parse_utg_js(utg_path, package_name)
|
||
num_nodes = utg_result['num_nodes']
|
||
num_reached_activities = utg_result['num_reached_activities']
|
||
app_num_total_activities = utg_result['app_num_total_activities']
|
||
print(f"[INFO] UTG 节点数: {num_nodes}, 已到达Activity: {num_reached_activities}/{app_num_total_activities}")
|
||
except Exception as e:
|
||
print(f"[WARN] 解析 UTG 文件失败: {e}")
|
||
|
||
print(f"[INFO] DroidBot 步数: {droidbot_steps}")
|
||
print(f"[INFO] GUI Agent 步数: {guiagent_steps}")
|
||
print(f"[INFO] 总步数: {total_steps}")
|
||
# 恢复原来的stdout/stderr
|
||
sys.stdout = old_stdout
|
||
sys.stderr = old_stderr
|
||
tee.close()
|
||
|
||
end_time = time.time()
|
||
duration = end_time - start_time
|
||
hours, rem = divmod(duration, 3600)
|
||
minutes, seconds = divmod(rem, 60)
|
||
time_str = f"{int(hours)}小时 {int(minutes)}分 {int(seconds)}秒"
|
||
print(f"[INFO] 任务耗时: {time_str}\n")
|
||
|
||
if os.path.exists(output_dir):
|
||
try:
|
||
time_file_path = os.path.join(output_dir, "time_cost.txt")
|
||
with open(time_file_path, "w", encoding="utf-8") as f:
|
||
f.write(f"Test Duration: {time_str}\n")
|
||
f.write(f"Start Time: {datetime.fromtimestamp(start_time).strftime('%Y-%m-%d %H:%M:%S')}\n")
|
||
f.write(f"End Time: {datetime.fromtimestamp(end_time).strftime('%Y-%m-%d %H:%M:%S')}\n")
|
||
except Exception as e:
|
||
print(f"[ERROR] 无法记录耗时文件: {e}")
|
||
|
||
result = TaskResult(
|
||
status=status,
|
||
exit_code=ret_code,
|
||
error_reason=error_reason,
|
||
droidbot_steps=droidbot_steps,
|
||
guiagent_steps=guiagent_steps,
|
||
total_steps=total_steps,
|
||
duration_seconds=duration,
|
||
guiagent_message=guiagent_message,
|
||
login_count=login_count,
|
||
register_count=register_count,
|
||
stuck_reason_code=stuck_reason_code,
|
||
num_nodes=num_nodes,
|
||
num_reached_activities=num_reached_activities,
|
||
app_num_total_activities=app_num_total_activities
|
||
)
|
||
|
||
# 如果是用户中断，返回结果后重新抛出异常，让上层 batch_run 能够捕获并退出
|
||
if user_interrupted:
|
||
raise KeyboardInterrupt(result)
|
||
|
||
return result
|
||
|
||
|
||
def main():
|
||
logger = get_logger(__name__)
|
||
logger.info("开始测试任务")
|
||
|
||
package = "com.google.android.youtube"
|
||
app_name = "YouTube"
|
||
config = load_config()
|
||
task_runner = TaskRunner(config)
|
||
output_dir = "test_output"
|
||
result = task_runner.run_task(package, app_name, output_dir)
|
||
|
||
logger.info(f"测试完成: status={result.status}, total_steps={result.total_steps}")
|
||
return result
|
||
|
||
|
||
if __name__ == "__main__":
|
||
setup_logging(level=logging.INFO, enable_file_handler=False)
|
||
# from pyinstrument import Profiler
|
||
# profiler = Profiler()
|
||
# profiler.start()
|
||
|
||
main()
|
||
|
||
# profiler.stop()
|
||
# profiler.write_html("report.html")
|