1167 lines
48 KiB
Python
1167 lines
48 KiB
Python
# -*- encoding=utf8 -*-
|
||
from __future__ import annotations
|
||
|
||
from airtest.core.api import *
|
||
from airtest.cli.parser import cli_setup
|
||
from poco.drivers.android.uiautomation import AndroidUiautomationPoco
|
||
import subprocess
|
||
import os
|
||
import re
|
||
import csv
|
||
import json
|
||
import traceback
|
||
import time
|
||
import sys
|
||
from dataclasses import dataclass
|
||
from datetime import datetime
|
||
|
||
# 导入统一的日志配置模块
|
||
from logging_config import setup_logging, get_logger
|
||
|
||
from start_test import (
|
||
TaskRunner,
|
||
TaskResult,
|
||
load_config,
|
||
EXIT_SUCCESS,
|
||
EXIT_ERROR_GENERAL,
|
||
EXIT_ERROR_USER,
|
||
EXIT_ADB_ERROR,
|
||
EXIT_NETWORK_ERROR,
|
||
EXIT_SIMULATOR_ERROR,
|
||
EXIT_EXPLORATION_STUCK,
|
||
EXIT_APP_CRASH_ERROR,
|
||
EXIT_APP_NEED_UPDATE,
|
||
EXIT_APP_LAUNCH_ERROR,
|
||
STUCK_REASON_TO_ERROR_TYPE,
|
||
)
|
||
from result_codes import ErrorInfo, STUCK_TO_ERROR, ErrorCategory, InfraError, AppError, BusinessError, DownloadError
|
||
from utils_android.Manager.mumu_manager import MuMuManager
|
||
from utils_android.Manager.physical_device_manager import PhysicalDeviceManager
|
||
from utils_android.Manager.pcapdroid_manager import PCAPManager
|
||
from utils_android.Manager.data_manager import DataManager
|
||
from utils_android.Manager.network_manager import NetworkWatchdog, NetworkException, check_network, wait_for_network
|
||
from utils_android.device_config import (
|
||
build_airtest_android_uri,
|
||
ensure_android_wireless_connected,
|
||
get_android_device_serial,
|
||
)
|
||
from utils_android.download_app.download_controller import (
|
||
download_app_chain,
|
||
stop_services,
|
||
reset_runtime_state as reset_download_runtime_state,
|
||
)
|
||
from utils_android.download_app.country_codes import normalize_country_codes
|
||
from DroidBot.exceptions import ADBException
|
||
|
||
|
||
# ==========================================
|
||
# 0. NullTaskWorker（单应用模式，不连 Redis）
|
||
# ==========================================
|
||
class NullTaskWorker:
|
||
"""不连接 Redis 的 Worker 桩，用于单应用本地测试模式。"""
|
||
|
||
def init(self):
|
||
return None
|
||
|
||
def report(self, report_data):
|
||
status = report_data.get("status", "?")
|
||
task_key = report_data.get("task_key", "?")
|
||
error_info = report_data.get("error", "")
|
||
if error_info:
|
||
print(f"[NullWorker] Report task={task_key} status={status} error={error_info}")
|
||
else:
|
||
print(f"[NullWorker] Report task={task_key} status={status}")
|
||
return None
|
||
|
||
def retry(self):
|
||
return True
|
||
|
||
def event(self, event_data):
|
||
pass
|
||
|
||
|
||
# ==========================================
|
||
# 1. 日志重定向类 (捕获所有 stdout/stderr)
|
||
# ==========================================
|
||
class TeeOutput:
|
||
"""同时将输出写入文件和控制台"""
|
||
def __init__(self, filepath, original_stream):
|
||
self.file = open(filepath, 'a', encoding='utf-8')
|
||
self.original_stream = original_stream
|
||
|
||
def write(self, data):
|
||
self.file.write(data)
|
||
self.file.flush()
|
||
self.original_stream.write(data)
|
||
self.original_stream.flush()
|
||
|
||
def flush(self):
|
||
self.file.flush()
|
||
self.original_stream.flush()
|
||
|
||
def close(self):
|
||
self.file.close()
|
||
|
||
def isatty(self):
|
||
return False
|
||
|
||
|
||
# 配置日志目录和文件路径
|
||
log_dir = os.path.join(os.path.dirname(__file__), 'logs')
|
||
os.makedirs(log_dir, exist_ok=True)
|
||
log_filename = datetime.now().strftime("batch_run_%Y%m%d_%H%M%S.log")
|
||
log_filepath = os.path.join(log_dir, log_filename)
|
||
|
||
# 重定向 stdout 和 stderr 到日志文件
|
||
sys.stdout = TeeOutput(log_filepath, sys.stdout)
|
||
sys.stderr = TeeOutput(log_filepath, sys.stderr)
|
||
|
||
# 使用统一的日志配置
|
||
setup_logging(
|
||
level='INFO',
|
||
log_file=log_filepath,
|
||
debug_mode=False
|
||
)
|
||
logger = get_logger(__name__)
|
||
logger.info(f"日志文件路径: {log_filepath}")
|
||
|
||
|
||
# ==========================================
|
||
# 2. 统计管理 (BatchStatistics)
|
||
# ==========================================
|
||
class BatchStatistics:
|
||
"""批量测试统计管理器"""
|
||
|
||
def __init__(self):
|
||
self.start_time = datetime.now()
|
||
self.records = []
|
||
self.current_round = 0
|
||
|
||
def add_record(self, round_num: int, app_name: str, package_name: str,
|
||
result: TaskResult, is_retry: bool = False):
|
||
"""添加一条测试记录"""
|
||
self.records.append({
|
||
"round": round_num,
|
||
"app_name": app_name,
|
||
"package_name": package_name,
|
||
"status": result.status,
|
||
"exit_code": result.exit_code,
|
||
"error_reason": result.error_reason,
|
||
"droidbot_steps": result.droidbot_steps,
|
||
"guiagent_steps": result.guiagent_steps,
|
||
"total_steps": result.total_steps,
|
||
"duration_seconds": round(result.duration_seconds, 2),
|
||
"is_retry": is_retry,
|
||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
})
|
||
|
||
def get_current_statistics(self, result: TaskResult, round_num: int, is_retry: bool = False, download_source: str = None) -> dict:
|
||
"""获取当前任务的统计数据，用于上报"""
|
||
return {
|
||
"round": round_num,
|
||
"exit_code": result.exit_code,
|
||
"error_reason": result.error_reason,
|
||
"droidbot_steps": result.droidbot_steps,
|
||
"guiagent_steps": result.guiagent_steps,
|
||
"total_steps": result.total_steps,
|
||
"duration_seconds": round(result.duration_seconds, 2),
|
||
"is_retry": is_retry,
|
||
"download_source": download_source,
|
||
"num_nodes": result.num_nodes,
|
||
"num_reached_activities": result.num_reached_activities,
|
||
"app_num_total_activities": result.app_num_total_activities
|
||
}
|
||
|
||
def get_round_summary(self, round_num: int) -> dict:
|
||
"""获取某轮的统计摘要"""
|
||
round_records = [r for r in self.records if r["round"] == round_num]
|
||
success_count = sum(1 for r in round_records if r["status"] == "SUCCESS")
|
||
failed_count = sum(1 for r in round_records if r["status"] == "FAILED")
|
||
return {
|
||
"total": len(round_records),
|
||
"success": success_count,
|
||
"failed": failed_count,
|
||
"retry_count": sum(1 for r in round_records if r["is_retry"])
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class TaskContext:
|
||
task: dict
|
||
app_name: str
|
||
package_name: str
|
||
task_key: str
|
||
country_codes: list
|
||
available_sources: list
|
||
local_apk_dir: str = None
|
||
local_apk_files: list = None
|
||
download_source: str = None
|
||
output_dir: str = None
|
||
final_result: TaskResult = None
|
||
status_to_report: str = "failed"
|
||
error_info: object = None
|
||
collect_retry_count: int = 0
|
||
keep_app_installed: bool = False
|
||
enable_app_block: bool = False
|
||
app_block_mode: str = "set"
|
||
app_block_tag: str = None
|
||
|
||
|
||
class FinalizeRecoveryFailed(RuntimeError):
|
||
"""finalize 阶段环境恢复失败后终止当前轮。"""
|
||
|
||
|
||
# ==========================================
|
||
# 3. 执行部分 (TestExecutor)
|
||
# ==========================================
|
||
class TestExecutor:
|
||
"""自动化测试的执行引擎"""
|
||
BROWSER_PACKAGE_CANDIDATES = (
|
||
"com.android.chrome",
|
||
"com.android.chromium",
|
||
"com.android.browser",
|
||
)
|
||
|
||
def __init__(self, stats: BatchStatistics = None, worker=None):
|
||
self.config = load_config()
|
||
self.data = DataManager(self.config)
|
||
self.device = MuMuManager(self.config) if self.config.get('IS_EMULATOR', True) else PhysicalDeviceManager(self.config)
|
||
self.mumu = self.device
|
||
self.network_watchdog = NetworkWatchdog()
|
||
self._adb_helper = None
|
||
self._pcap = None
|
||
self.task_runner = TaskRunner(self.config)
|
||
self.stats = stats or BatchStatistics()
|
||
self.current_round = 0
|
||
self._base_packages = None
|
||
self.device_serial = get_android_device_serial(self.config)
|
||
self.app_block_app_list = self.config.get('APP_BLOCK_APP_LIST')
|
||
self.app_block_url_lib = self.config.get('APP_BLOCK_URL_LIB')
|
||
self._worker = worker
|
||
|
||
@property
|
||
def adb_helper(self):
|
||
"""延迟初始化 ADB Helper"""
|
||
if self._adb_helper is None:
|
||
from utils_android.download_app.adb_helper import ADBHelper
|
||
self._adb_helper = ADBHelper(self.device_serial)
|
||
return self._adb_helper
|
||
|
||
@property
|
||
def pcap(self):
|
||
"""延迟初始化 PCAPManager（依赖 adb_helper）"""
|
||
if self._pcap is None:
|
||
self._pcap = PCAPManager(self.adb_helper, root_capture=self.device.device_type == 'emulator')
|
||
return self._pcap
|
||
|
||
@property
|
||
def device_label(self) -> str:
|
||
return "模拟器" if self.device.device_type == "emulator" else "真机"
|
||
|
||
@property
|
||
def is_emulator_device(self) -> bool:
|
||
return self.device.device_type == "emulator"
|
||
|
||
def _reset_runtime_state(self):
|
||
"""设备重启后丢弃所有持有旧设备连接的缓存对象。"""
|
||
self._adb_helper = None
|
||
self._pcap = None
|
||
reset_download_runtime_state()
|
||
|
||
def _reinitialize_after_restart(self):
|
||
self._reset_runtime_state()
|
||
print(f"[INFO] {self.device_label}已恢复，重新初始化 Airtest 连接...")
|
||
try:
|
||
return self.setup()
|
||
except ADBException as e:
|
||
logger.error(f"重启后重新初始化 Airtest 失败: {e}")
|
||
return False
|
||
except Exception as e:
|
||
logger.error(f"重启后初始化运行时失败: {e}")
|
||
return False
|
||
|
||
def _restart_after_adb_fault(self):
|
||
print(f"[CRITICAL] 检测到 ADB 断联，正在重启{self.device_label}...")
|
||
logger.error(f"检测到 ADB 断联，触发{self.device_label}重启恢复流程")
|
||
if not self.device.restart():
|
||
self._reset_runtime_state()
|
||
return False
|
||
return self._reinitialize_after_restart()
|
||
|
||
def _load_base_packages(self):
|
||
"""加载纯净镜像的包名白名单（仅加载一次）"""
|
||
if self._base_packages is not None:
|
||
return self._base_packages
|
||
|
||
if not self.is_emulator_device:
|
||
logger.info("当前为真机模式，跳过纯净镜像包名白名单校验")
|
||
self._base_packages = set()
|
||
return self._base_packages
|
||
|
||
packages_file = os.path.join(os.path.dirname(__file__), 'doc', 'android_packages.txt')
|
||
if not os.path.exists(packages_file):
|
||
logger.warning(f"纯净镜像包名文件不存在: {packages_file}")
|
||
self._base_packages = set()
|
||
return self._base_packages
|
||
|
||
packages = set()
|
||
with open(packages_file, 'r', encoding='utf-8') as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line.startswith('package:'):
|
||
packages.add(line.split('package:')[1].strip())
|
||
elif line: # 兼容没有 package: 前缀的格式
|
||
packages.add(line)
|
||
|
||
logger.info(f"已加载纯净镜像包名白名单: {len(packages)} 个包")
|
||
self._base_packages = packages
|
||
return self._base_packages
|
||
|
||
def _get_third_party_packages(self):
|
||
"""获取设备上已安装的第三方包列表（通过 adb_helper 封装调用）"""
|
||
output = self.adb_helper.shell("pm list package")
|
||
return [line.split("package:")[1].strip() for line in output.splitlines() if "package:" in line]
|
||
|
||
def _stop_browser_apps(self):
|
||
"""停止所有已知浏览器候选包，兼容模拟器与真机。"""
|
||
for package_name in self.BROWSER_PACKAGE_CANDIDATES:
|
||
try:
|
||
stop_app(package_name)
|
||
except Exception as exc:
|
||
logger.debug(f"停止浏览器 {package_name} 失败，忽略: {exc}")
|
||
|
||
def _resolve_pocoservice_apk(self) -> str:
|
||
"""定位 Poco 随包提供的 Android 服务 APK。"""
|
||
import poco
|
||
|
||
poco_dir = os.path.dirname(os.path.abspath(poco.__file__))
|
||
return os.path.join(poco_dir, "drivers", "android", "lib", "pocoservice-debug.apk")
|
||
|
||
def uninstall_extra_apps(self, current_package: str):
|
||
"""卸载白名单和当前任务包名之外的多余应用"""
|
||
base_packages = self._load_base_packages()
|
||
if not base_packages:
|
||
logger.warning("白名单为空，跳过卸载多余应用")
|
||
return
|
||
|
||
# 获取当前设备上所有已安装的包
|
||
installed = self._get_third_party_packages()
|
||
# 计算需要卸载的包：不在白名单中，且不是当前任务的包
|
||
to_uninstall = [
|
||
pkg for pkg in installed
|
||
if pkg not in base_packages and pkg != current_package
|
||
]
|
||
|
||
if not to_uninstall:
|
||
logger.info("没有需要卸载的多余应用")
|
||
return
|
||
|
||
logger.info(f"检测到 {len(to_uninstall)} 个多余应用，开始卸载: {to_uninstall}")
|
||
for pkg in to_uninstall:
|
||
if self.adb_helper.uninstall(pkg):
|
||
logger.info(f"已卸载: {pkg}")
|
||
continue
|
||
logger.warning(f"卸载 {pkg} 失败")
|
||
|
||
def setup(self):
|
||
if not cli_setup():
|
||
try:
|
||
ensure_android_wireless_connected(self.config, self.device_serial)
|
||
auto_setup(
|
||
__file__,
|
||
logdir=False,
|
||
devices=[build_airtest_android_uri(self.config, serial=self.device_serial)],
|
||
)
|
||
except Exception as e:
|
||
print(f"设备连接失败: {e}")
|
||
return False
|
||
|
||
if "com.netease.open.pocoservice" not in self._get_third_party_packages():
|
||
install(self._resolve_pocoservice_apk())
|
||
sleep(5)
|
||
|
||
start_app("com.netease.open.pocoservice")
|
||
sleep(2)
|
||
return True
|
||
|
||
def _create_task_context(self, task: dict) -> TaskContext:
|
||
country_codes = normalize_country_codes(task.get('country_codes') or task.get('country_code', ''))
|
||
return TaskContext(
|
||
task=task,
|
||
app_name=task.get('app_name', 'Unknown App'),
|
||
package_name=task.get('package_name', ''),
|
||
task_key=task.get('task_key', ''),
|
||
country_codes=country_codes,
|
||
available_sources=task.get('available_sources', ['google_play', 'local']),
|
||
local_apk_dir=task.get('local_apk_dir', '') or None,
|
||
local_apk_files=task.get('local_apk_files', []) or None,
|
||
error_info=ErrorInfo.success(),
|
||
keep_app_installed=bool(task.get('keep_app_installed', False)),
|
||
enable_app_block=bool(task.get('blocked', False)),
|
||
app_block_mode=task.get('app_block_mode', 'set'),
|
||
app_block_tag=task.get('app_block_tag') or None,
|
||
)
|
||
|
||
def _error_type_str(self, error) -> str:
|
||
if hasattr(error, 'category') and hasattr(error, 'code'):
|
||
return f"{error.category.name}/{error.code}"
|
||
if isinstance(error, dict) and error.get('category') is not None and error.get('code') is not None:
|
||
return f"{error['category']}/{error['code']}"
|
||
return "UNEXPECTED/0"
|
||
|
||
def _report_terminal(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker, *, status: str, failed_stage: str, error, error_message: str = None, download_errors: dict = None, extra_error: dict = None, **extra):
|
||
error_payload = error.to_report_dict() if hasattr(error, 'to_report_dict') else dict(error or {})
|
||
if extra_error:
|
||
error_payload.update(extra_error)
|
||
if error_message is None:
|
||
error_message = error_payload.get('reason', '')
|
||
mon.finish_task(
|
||
status,
|
||
failed_stage=failed_stage,
|
||
error_type=self._error_type_str(error),
|
||
error_message=error_message,
|
||
download_errors=download_errors,
|
||
)
|
||
return worker.report(
|
||
mon.build_report_payload(
|
||
status,
|
||
error_payload,
|
||
download_errors=download_errors,
|
||
**extra,
|
||
)
|
||
)
|
||
|
||
def _prepare_download(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker):
|
||
mon.start_stage("download", "running_download")
|
||
while True:
|
||
device_ok, was_restarted = self.device.ensure_running()
|
||
if not device_ok:
|
||
print(f"[ERROR] 跳过 {ctx.app_name}: {self.device_label}不可用")
|
||
mon.finish_stage("download", status="failed")
|
||
return self._report_terminal(
|
||
ctx,
|
||
mon,
|
||
worker,
|
||
status="failed",
|
||
failed_stage="download",
|
||
error=ErrorInfo.infra(
|
||
InfraError.EMULATOR_START_FAILED if self.is_emulator_device else InfraError.ADB_ERROR
|
||
),
|
||
error_message=f"{self.device_label} unavailable",
|
||
)
|
||
|
||
try:
|
||
if was_restarted:
|
||
mon.mark("device_restarted")
|
||
setup_ok = self._reinitialize_after_restart()
|
||
if not setup_ok:
|
||
print(f"[ERROR] 跳过 {ctx.app_name}: Airtest 重新连接失败")
|
||
mon.finish_stage("download", status="failed")
|
||
return self._report_terminal(
|
||
ctx,
|
||
mon,
|
||
worker,
|
||
status="failed",
|
||
failed_stage="download",
|
||
error=ErrorInfo.infra(InfraError.AIRTEST_INIT_FAILED),
|
||
error_message="airtest reinit failed",
|
||
)
|
||
mon.mark("airtest_reinitialized")
|
||
worker.retry()
|
||
|
||
self.uninstall_extra_apps(ctx.package_name)
|
||
return None
|
||
except ADBException as e:
|
||
logger.error(f"准备下载阶段检测到 ADB 异常: {e}")
|
||
print(f"[ERROR] 清理多余应用时 ADB 断联 {ctx.app_name}，尝试恢复...")
|
||
if not self._restart_after_adb_fault():
|
||
mon.finish_stage("download", status="failed")
|
||
return self._report_terminal(
|
||
ctx,
|
||
mon,
|
||
worker,
|
||
status="failed",
|
||
failed_stage="download",
|
||
error=ErrorInfo.infra(
|
||
InfraError.EMULATOR_RECOVERY_FAILED if self.is_emulator_device else InfraError.ADB_ERROR,
|
||
str(e),
|
||
),
|
||
error_message=f"{self.device_label} recovery failed during cleanup",
|
||
)
|
||
mon.mark("download_recovered")
|
||
worker.retry()
|
||
|
||
def _build_download_kwargs(self, ctx: TaskContext) -> dict:
|
||
kwargs = {
|
||
'app_name': ctx.app_name,
|
||
'country': ctx.country_codes[0],
|
||
'country_codes': ctx.country_codes,
|
||
'local_path': self.config.get('LOCAL_APK_PATH'),
|
||
'apkpure_path': self.config.get('APKPURE_APK_PATH'),
|
||
'target_account': '',
|
||
'available_sources': ctx.available_sources,
|
||
}
|
||
if ctx.local_apk_dir:
|
||
print(f"[Task] 收到本地APK路径: {ctx.local_apk_dir}")
|
||
print(f"[Task] 本地APK文件: {ctx.local_apk_files}")
|
||
kwargs['local_apk_dir'] = ctx.local_apk_dir
|
||
else:
|
||
print(f"[Task] 未收到本地APK路径 (available_sources={ctx.available_sources})")
|
||
if ctx.local_apk_files:
|
||
kwargs['local_apk_files'] = ctx.local_apk_files
|
||
return kwargs
|
||
|
||
def _handle_download_result(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker, dl_success: bool, dl_result: dict):
|
||
if not dl_success:
|
||
error_code = dl_result.get('error_code', DownloadError.OTHER)
|
||
errors_json = dl_result.get('errors', {})
|
||
attempted_countries = dl_result.get('attempted_countries', [])
|
||
mon.finish_stage("download", status="failed", download_source=dl_result.get('source', 'unknown'))
|
||
print(f"[ERROR] 应用下载失败 {ctx.app_name}: {errors_json}")
|
||
return self._report_terminal(
|
||
ctx,
|
||
mon,
|
||
worker,
|
||
status="failed",
|
||
failed_stage="download",
|
||
error=ErrorInfo.download(error_code),
|
||
error_message=json.dumps(errors_json, ensure_ascii=False),
|
||
download_errors=errors_json,
|
||
attempted_countries=attempted_countries,
|
||
)
|
||
|
||
if dl_result.get('run') == 'crash':
|
||
details = dl_result.get('details', 'App crashed')
|
||
crashed_source = dl_result.get('crashed_source', 'unknown')
|
||
mon.finish_stage("download", status="failed", download_source=dl_result.get('source', 'unknown'))
|
||
print(f"[ERROR] 应用安装成功但无法运行 {ctx.app_name}: {details}")
|
||
return self._report_terminal(
|
||
ctx,
|
||
mon,
|
||
worker,
|
||
status="failed",
|
||
failed_stage="download",
|
||
error=ErrorInfo.app(AppError.CRASH, details),
|
||
error_message=details,
|
||
extra_error={"crashed_source": crashed_source},
|
||
)
|
||
|
||
print(f"[INFO] 应用准备就绪 (来源: {dl_result.get('source')}, 状态: {dl_result.get('state')})")
|
||
ctx.download_source = dl_result.get('source', 'unknown')
|
||
mon.finish_stage("download", status="success", download_source=ctx.download_source)
|
||
mon.start_stage("collect", "running_collect")
|
||
return None
|
||
|
||
def _recover_download(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker, dl_kwargs: dict):
|
||
recovery_success = self.device.recover_from_frozen() if self.device.is_player_frozen() else self.device.restart()
|
||
if not recovery_success:
|
||
self._reset_runtime_state()
|
||
mon.finish_stage("download", status="failed")
|
||
print(f"[ERROR] {self.device_label}恢复失败，无法继续测试: {ctx.app_name}")
|
||
return self._report_terminal(
|
||
ctx,
|
||
mon,
|
||
worker,
|
||
status="failed",
|
||
failed_stage="download",
|
||
error=ErrorInfo.infra(
|
||
InfraError.EMULATOR_RECOVERY_FAILED if self.is_emulator_device else InfraError.ADB_ERROR
|
||
),
|
||
error_message=f"{self.device_label} recovery failed during download",
|
||
)
|
||
|
||
mon.mark("download_recovered")
|
||
logger.info(f"{self.device_label}恢复成功，重新初始化连接...")
|
||
if not self._reinitialize_after_restart():
|
||
mon.finish_stage("download", status="failed")
|
||
print(f"[ERROR] 跳过 {ctx.app_name}: Airtest 重新连接失败")
|
||
return self._report_terminal(
|
||
ctx,
|
||
mon,
|
||
worker,
|
||
status="failed",
|
||
failed_stage="download",
|
||
error=ErrorInfo.infra(InfraError.AIRTEST_INIT_FAILED),
|
||
error_message="airtest reinit failed after download recovery",
|
||
)
|
||
worker.retry()
|
||
logger.info("重新尝试下载应用...")
|
||
try:
|
||
dl_success, dl_result = download_app_chain(ctx.package_name, **dl_kwargs)
|
||
return self._handle_download_result(ctx, mon, worker, dl_success, dl_result)
|
||
except Exception as e:
|
||
mon.finish_stage("download", status="failed")
|
||
print(f"[ERROR] 重试下载时出现异常 {ctx.app_name}: {e}")
|
||
return self._report_terminal(
|
||
ctx,
|
||
mon,
|
||
worker,
|
||
status="failed",
|
||
failed_stage="download",
|
||
error=ErrorInfo.download(DownloadError.OTHER, str(e)),
|
||
error_message=str(e),
|
||
)
|
||
|
||
def _execute_download(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker):
|
||
print(f"[INFO] 正在启动智能下载链条: {ctx.app_name} ({ctx.package_name})")
|
||
dl_kwargs = self._build_download_kwargs(ctx)
|
||
try:
|
||
dl_success, dl_result = download_app_chain(ctx.package_name, **dl_kwargs)
|
||
return self._handle_download_result(ctx, mon, worker, dl_success, dl_result)
|
||
except (ADBException, IndexError) as e:
|
||
logger.error(f"下载过程中检测到 ADB 异常: {e}")
|
||
print(f"[ERROR] 下载应用时 ADB 断联 {ctx.app_name}，尝试恢复...")
|
||
return self._recover_download(ctx, mon, worker, dl_kwargs)
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
mon.finish_stage("download", status="failed")
|
||
print(f"[ERROR] 下载过程出现异常 {ctx.app_name}: {e}")
|
||
return self._report_terminal(
|
||
ctx,
|
||
mon,
|
||
worker,
|
||
status="failed",
|
||
failed_stage="download",
|
||
error=ErrorInfo.download(DownloadError.OTHER, str(e)),
|
||
error_message=str(e),
|
||
)
|
||
|
||
def _prepare_collect(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker):
|
||
print("[INFO] 清理下载相关进程 (Google Play, Browser, APKPure, PocoService)...")
|
||
stop_services()
|
||
stop_app("com.android.vending")
|
||
self._stop_browser_apps()
|
||
stop_app("com.apkpure.aegon")
|
||
stop_app("com.netease.open.pocoservice")
|
||
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
ctx.output_dir = os.path.join(self.config['OUTPUT_BASE_DIR'], f"{ctx.package_name}_{timestamp}")
|
||
|
||
# 根据当前任务上下文重新初始化 PCAPManager（应用阻塞时关闭 root 抓包）
|
||
root_capture = self.device.device_type == 'emulator' and not ctx.enable_app_block
|
||
app_blocker = None
|
||
if ctx.enable_app_block and self.app_block_app_list and self.app_block_url_lib:
|
||
from utils_android.Manager.pcapdroid_manager import AppBlocker
|
||
app_blocker = AppBlocker(
|
||
self.adb_helper,
|
||
app_list_file=self.app_block_app_list,
|
||
url_lib_file=self.app_block_url_lib,
|
||
)
|
||
self._pcap = PCAPManager(
|
||
self.adb_helper,
|
||
root_capture=root_capture,
|
||
app_blocker=app_blocker,
|
||
)
|
||
|
||
# 导入应用阻塞规则
|
||
if ctx.enable_app_block:
|
||
print(f"[INFO] 正在导入应用阻塞规则 (package={ctx.package_name}, tag={ctx.app_block_tag}, mode={ctx.app_block_mode})")
|
||
block_ok = self._pcap.import_app_blocklist(
|
||
ctx.package_name,
|
||
mode=ctx.app_block_mode,
|
||
tag=ctx.app_block_tag,
|
||
)
|
||
if block_ok:
|
||
print(f"[INFO] 应用阻塞规则导入成功")
|
||
else:
|
||
print(f"[WARNING] 应用阻塞规则导入失败或该应用无规则")
|
||
|
||
if self.pcap.start_capture(ctx.package_name):
|
||
mon.mark("pcap_started")
|
||
return None
|
||
|
||
mon.finish_stage("collect", status="failed")
|
||
print(f"[ERROR] 跳过 {ctx.app_name}: PCAP 启动失败")
|
||
return self._report_terminal(
|
||
ctx,
|
||
mon,
|
||
worker,
|
||
status="failed",
|
||
failed_stage="collect",
|
||
error=ErrorInfo.infra(InfraError.PCAP_START_FAILED),
|
||
error_message="pcap start failed",
|
||
)
|
||
|
||
def _pcap_sync_wrapper(self, pkg_name):
|
||
print(f"[Callback] 检测到 PCAPDroid 停止，正在推送流量文件: {pkg_name}")
|
||
self.data.sync_latest_traffic_file(pkg_name)
|
||
|
||
def _run_collect_task(self, ctx: TaskContext, mon: TaskMonitor) -> TaskResult:
|
||
try:
|
||
result = self.task_runner.run_task(
|
||
ctx.package_name,
|
||
ctx.app_name,
|
||
ctx.output_dir,
|
||
pcap_callback=self._pcap_sync_wrapper,
|
||
enable_app_block=ctx.enable_app_block,
|
||
)
|
||
except KeyboardInterrupt as e:
|
||
if e.args and isinstance(e.args[0], TaskResult):
|
||
result = e.args[0]
|
||
self.stats.add_record(self.current_round, ctx.app_name, ctx.package_name, result)
|
||
raise
|
||
return result
|
||
|
||
def _retry_collect(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker, recover_fn, failure_error, retry_error_builder):
|
||
if not recover_fn():
|
||
self._reset_runtime_state()
|
||
ctx.status_to_report = 'failed'
|
||
ctx.error_info = failure_error
|
||
return
|
||
|
||
mon.mark("collect_recovered")
|
||
if not self._reinitialize_after_restart():
|
||
ctx.status_to_report = 'failed'
|
||
ctx.error_info = ErrorInfo.infra(InfraError.AIRTEST_INIT_FAILED)
|
||
return
|
||
worker.retry()
|
||
retry_result = self.task_runner.run_task(
|
||
ctx.package_name,
|
||
ctx.app_name,
|
||
ctx.output_dir,
|
||
pcap_callback=self._pcap_sync_wrapper,
|
||
enable_app_block=ctx.enable_app_block,
|
||
)
|
||
self.stats.add_record(self.current_round, ctx.app_name, ctx.package_name, retry_result, is_retry=True)
|
||
ctx.final_result = retry_result
|
||
ctx.status_to_report = 'success' if retry_result.status == 'SUCCESS' else 'failed'
|
||
ctx.collect_retry_count = 1
|
||
if retry_result.status != "SUCCESS":
|
||
ctx.error_info = retry_error_builder(retry_result)
|
||
|
||
def _resolve_collect_outcome(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker):
|
||
result = ctx.final_result
|
||
ctx.status_to_report = 'success' if result.status == 'SUCCESS' else 'failed'
|
||
ctx.error_info = ErrorInfo.success()
|
||
ctx.collect_retry_count = 0
|
||
|
||
if result.exit_code == EXIT_SIMULATOR_ERROR:
|
||
print(f"[CRITICAL] 检测到{self.device_label}异常 (code {result.exit_code})，触发紧急恢复流程...")
|
||
self._retry_collect(
|
||
ctx,
|
||
mon,
|
||
worker,
|
||
self.device.recover if self.is_emulator_device else self.device.restart,
|
||
ErrorInfo.infra(
|
||
InfraError.EMULATOR_RECOVERY_FAILED if self.is_emulator_device else InfraError.ADB_ERROR
|
||
),
|
||
lambda retry_result: ErrorInfo.infra(
|
||
InfraError.EMULATOR_CRASH if self.is_emulator_device else InfraError.ADB_ERROR,
|
||
retry_result.error_reason,
|
||
),
|
||
)
|
||
elif result.exit_code == EXIT_ADB_ERROR:
|
||
print(f"[CRITICAL] 检测到 ADB 断联 (code {result.exit_code})，正在重启{self.device_label}...")
|
||
self._retry_collect(
|
||
ctx,
|
||
mon,
|
||
worker,
|
||
self.device.restart,
|
||
ErrorInfo.infra(
|
||
InfraError.EMULATOR_RECOVERY_FAILED if self.is_emulator_device else InfraError.ADB_ERROR
|
||
),
|
||
lambda retry_result: ErrorInfo.infra(InfraError.ADB_ERROR, retry_result.error_reason),
|
||
)
|
||
elif result.exit_code == EXIT_NETWORK_ERROR:
|
||
print(f"[ERROR] 网络异常，跳过当前应用: {ctx.app_name} ({result.error_reason})")
|
||
ctx.status_to_report = 'failed'
|
||
ctx.error_info = ErrorInfo.infra(InfraError.NETWORK_ERROR, result.error_reason)
|
||
elif result.exit_code == EXIT_APP_CRASH_ERROR:
|
||
if ctx.enable_app_block:
|
||
print(f"[WARN] Block任务应用闪退，视为正常结束: {ctx.app_name} ({result.error_reason})")
|
||
ctx.status_to_report = 'success'
|
||
ctx.error_info = ErrorInfo.success()
|
||
else:
|
||
print(f"[ERROR] 应用闪退，跳过当前应用: {ctx.app_name} ({result.error_reason})")
|
||
ctx.status_to_report = 'failed'
|
||
ctx.error_info = ErrorInfo.app(AppError.CRASH, result.error_reason)
|
||
elif result.exit_code == EXIT_APP_NEED_UPDATE:
|
||
print(f"[ERROR] 应用需更新，跳过当前应用: {ctx.app_name} ({result.error_reason})")
|
||
ctx.status_to_report = 'failed'
|
||
ctx.error_info = ErrorInfo.app(AppError.NEED_UPDATE, result.error_reason)
|
||
elif result.exit_code == EXIT_APP_LAUNCH_ERROR:
|
||
print(f"[ERROR] 启动异常，跳过当前应用: {ctx.app_name} ({result.error_reason})")
|
||
ctx.status_to_report = 'failed'
|
||
ctx.error_info = ErrorInfo.app(AppError.LAUNCH_ERROR, result.error_reason)
|
||
elif result.exit_code == EXIT_EXPLORATION_STUCK:
|
||
print(f"[ERROR] 探索停滞，跳过当前应用: {ctx.app_name} ({result.error_reason})")
|
||
ctx.status_to_report = 'failed'
|
||
ctx.error_info = ErrorInfo.from_stuck_reason(result.stuck_reason_code, ctx.final_result.guiagent_message or result.error_reason)
|
||
elif result.status != "SUCCESS":
|
||
print(f"[ERROR] 任务执行失败: {ctx.app_name} ({result.error_reason})")
|
||
ctx.status_to_report = 'failed'
|
||
ctx.error_info = ErrorInfo(ErrorCategory.BUSINESS_ERROR, BusinessError.OTHER, result.error_reason)
|
||
|
||
if ctx.error_info.category != ErrorCategory.SUCCESS:
|
||
return
|
||
|
||
if ctx.final_result.guiagent_message:
|
||
has_login_failed = "登录失败" in ctx.final_result.guiagent_message
|
||
has_register_failed = "注册失败" in ctx.final_result.guiagent_message
|
||
if has_login_failed and has_register_failed:
|
||
ctx.error_info = ErrorInfo.business(BusinessError.LOGIN_FAILED)
|
||
elif has_login_failed:
|
||
ctx.error_info = ErrorInfo.business(BusinessError.LOGIN_FAILED)
|
||
elif has_register_failed:
|
||
ctx.error_info = ErrorInfo.business(BusinessError.REGISTER_FAILED)
|
||
|
||
if ctx.final_result.login_count == 0 and ctx.final_result.register_count == 0:
|
||
ctx.error_info = ErrorInfo.business(BusinessError.NO_SCENARIO)
|
||
|
||
def _report_task_result(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker):
|
||
mon.finish_task(
|
||
ctx.status_to_report,
|
||
failed_stage=None if ctx.status_to_report == 'success' else 'collect',
|
||
error_type=None if ctx.status_to_report == 'success' else self._error_type_str(ctx.error_info),
|
||
error_message=None if ctx.status_to_report == 'success' else (ctx.error_info.reason or ctx.final_result.error_reason),
|
||
retry_count=ctx.collect_retry_count,
|
||
)
|
||
report_data = mon.build_report_payload(
|
||
ctx.status_to_report,
|
||
ctx.error_info if ctx.status_to_report != 'success' else None,
|
||
metrics={
|
||
"login_count": ctx.final_result.login_count,
|
||
"register_count": ctx.final_result.register_count,
|
||
"guiagent_message": ctx.final_result.guiagent_message,
|
||
"scenario_triggered": ctx.final_result.login_count > 0 or ctx.final_result.register_count > 0,
|
||
"stuck_reason_code": ctx.final_result.stuck_reason_code,
|
||
},
|
||
statistics=self.stats.get_current_statistics(
|
||
ctx.final_result,
|
||
self.current_round,
|
||
download_source=ctx.download_source,
|
||
),
|
||
)
|
||
print(f"任务完成，状态: {ctx.status_to_report}, 错误: {ctx.error_info.category.name}/{ctx.error_info.code}")
|
||
return worker.report(report_data)
|
||
|
||
def _run_one_task(self, worker: TaskWorker, ctx: TaskContext):
|
||
print(f"\n>>>> 开始处理任务: {ctx.app_name} <<<<")
|
||
print(f" 包名: {ctx.package_name}")
|
||
print(f" 任务键: {ctx.task_key}")
|
||
print(f" 国家码: {ctx.country_codes}")
|
||
print(f" 可用下载源: {ctx.available_sources}")
|
||
print(f" 应用阻塞: {ctx.enable_app_block} (mode={ctx.app_block_mode}, tag={ctx.app_block_tag})")
|
||
|
||
# 每个任务开始前：清除上一个任务遗留的 blocklist 并重置 PCAPManager
|
||
if self._pcap is not None:
|
||
try:
|
||
self._pcap.clear_app_blocklist()
|
||
except Exception as e:
|
||
logger.warning(f"清除上一个任务的 blocklist 失败: {e}")
|
||
self._pcap = None
|
||
|
||
from utils_android.TaskWorker.task_monitor import TaskMonitor # 延迟导入，避免单应用模式依赖 Redis
|
||
mon = TaskMonitor(worker, ctx.task_key)
|
||
mon.start_task()
|
||
self.network_watchdog.check_and_raise()
|
||
|
||
next_task = self._prepare_download(ctx, mon, worker)
|
||
if next_task is not None:
|
||
return next_task, mon
|
||
|
||
next_task = self._execute_download(ctx, mon, worker)
|
||
if next_task is not None:
|
||
return next_task, mon
|
||
|
||
next_task = self._prepare_collect(ctx, mon, worker)
|
||
if next_task is not None:
|
||
return next_task, mon
|
||
|
||
ctx.final_result = self._run_collect_task(ctx, mon)
|
||
self.stats.add_record(self.current_round, ctx.app_name, ctx.package_name, ctx.final_result)
|
||
self._resolve_collect_outcome(ctx, mon, worker)
|
||
mon.finish_stage(
|
||
"collect",
|
||
status="success" if ctx.status_to_report == 'success' else "failed",
|
||
duration_seconds=round(ctx.final_result.duration_seconds, 2),
|
||
)
|
||
stop_app(ctx.package_name)
|
||
start_app("com.netease.open.pocoservice")
|
||
self.pcap.stop_capture()
|
||
sleep(2)
|
||
return self._report_task_result(ctx, mon, worker), mon
|
||
|
||
def _stop_current_task(self, ctx: TaskContext, mon: TaskMonitor, worker: TaskWorker, error):
|
||
from utils_android.TaskWorker.task_monitor import TaskMonitor # noqa: F811
|
||
failed_stage = None
|
||
if mon and "collect" in mon.stage_started_at:
|
||
failed_stage = "collect"
|
||
mon.finish_stage("collect", status="failed")
|
||
elif mon and "download" in mon.stage_started_at:
|
||
failed_stage = "download"
|
||
mon.finish_stage("download", status="failed")
|
||
return self._report_terminal(
|
||
ctx,
|
||
mon or TaskMonitor(worker, ctx.task_key),
|
||
worker,
|
||
status="stop",
|
||
failed_stage=failed_stage,
|
||
error=error,
|
||
error_message=error.get('reason') if isinstance(error, dict) else error.reason,
|
||
)
|
||
|
||
def _finalize_task(self, ctx: TaskContext, worker: TaskWorker):
|
||
traffic_synced = False
|
||
log_synced = False
|
||
if ctx.package_name:
|
||
if ctx.keep_app_installed:
|
||
print(f"[INFO] 保留应用: {ctx.package_name}")
|
||
else:
|
||
print(f"[INFO] 卸载应用: {ctx.package_name}")
|
||
if self.adb_helper.uninstall(ctx.package_name):
|
||
print(f"[INFO] 应用 {ctx.package_name} 卸载成功")
|
||
else:
|
||
print(f"[WARNING] 卸载应用 {ctx.package_name} 失败")
|
||
traffic_synced = bool(self.data.sync_latest_traffic_file(ctx.package_name))
|
||
if log_filepath:
|
||
log_synced = bool(self.data.sync_log_file(log_filepath))
|
||
|
||
# 清除应用阻塞规则
|
||
if ctx.enable_app_block:
|
||
try:
|
||
print("[INFO] 正在清除应用阻塞规则...")
|
||
if self._pcap is not None:
|
||
self._pcap.clear_app_blocklist()
|
||
else:
|
||
from utils_android.Manager.pcapdroid_manager import AppBlocker
|
||
AppBlocker(self.adb_helper).clear_blocklist()
|
||
except Exception as e:
|
||
logger.warning(f"清除应用阻塞规则失败: {e}")
|
||
|
||
if not ctx.output_dir:
|
||
worker.event({
|
||
"event_type": "artifacts_synced",
|
||
"task_key": ctx.task_key,
|
||
"package_name": ctx.package_name,
|
||
"traffic_synced": traffic_synced,
|
||
"log_synced": log_synced,
|
||
})
|
||
return
|
||
worker.event({
|
||
"event_type": "artifacts_synced",
|
||
"task_key": ctx.task_key,
|
||
"package_name": ctx.package_name,
|
||
"traffic_synced": traffic_synced,
|
||
"log_synced": log_synced,
|
||
})
|
||
|
||
def _finalize_with_recovery(self, ctx: TaskContext, worker: TaskWorker, next_task):
|
||
try:
|
||
self._finalize_task(ctx, worker)
|
||
return next_task
|
||
except ADBException as e:
|
||
logger.error(f"finalize 阶段检测到 ADB 异常: {e}")
|
||
print(f"[CRITICAL] finalize 卸载阶段 ADB 断联 {ctx.app_name}，尝试重启{self.device_label}恢复...")
|
||
if self._restart_after_adb_fault():
|
||
print("[INFO] finalize 阶段环境恢复成功，继续执行后续任务")
|
||
return next_task
|
||
if next_task:
|
||
logger.warning("finalize 恢复失败，重新计时已领取的下一任务")
|
||
worker.retry()
|
||
raise FinalizeRecoveryFailed(f"finalize recovery failed: {e}")
|
||
|
||
def run(self, initial_task=None):
|
||
from utils_android.TaskWorker import TaskWorker # 延迟导入，避免单应用模式依赖 Redis 配置
|
||
worker = self._worker if self._worker is not None else TaskWorker()
|
||
|
||
logger.info("启动后台网络监控...")
|
||
self.network_watchdog.start()
|
||
|
||
print("\n[1] 初始化Worker...")
|
||
if initial_task is not None:
|
||
task = initial_task
|
||
print(f"[INFO] 使用注入的初始任务: {task.get('package_name', '?')}")
|
||
else:
|
||
task = worker.init()
|
||
if not task:
|
||
print("初始化失败，程序退出")
|
||
self.network_watchdog.stop()
|
||
return
|
||
|
||
print("\n[2] 开始处理任务...")
|
||
print("按 Ctrl+C 停止Worker\n")
|
||
|
||
fatal_error = None
|
||
interrupted = False
|
||
|
||
while task:
|
||
if task.get('package_name', '') == '-1':
|
||
logger.error(f"任务包名异常, 任务键: {task.get('task_key', '')}")
|
||
break
|
||
|
||
ctx = self._create_task_context(task)
|
||
if task.get('traffic_root'):
|
||
self.config['TRAFFIC_DATA_SHARE'] = task['traffic_root']
|
||
mon = None
|
||
next_task = None
|
||
interrupted_this_task = False
|
||
try:
|
||
next_task, mon = self._run_one_task(worker, ctx)
|
||
except NetworkException as e:
|
||
logger.error(f"后台监控检测到网络异常: {e}")
|
||
lan_info = f"{e.ip_address}局域网" if e.ip_address else "未知局域网"
|
||
mac_info = f"{e.mac_address}mac" if e.mac_address else "未知mac"
|
||
stop_error = ErrorInfo.infra(InfraError.NETWORK_ERROR, f"{lan_info}，{mac_info} 网络错误")
|
||
next_task = self._stop_current_task(ctx, mon, worker, stop_error)
|
||
logger.info("已停止当前任务，等待网络恢复...")
|
||
self.network_watchdog.stop()
|
||
except KeyboardInterrupt:
|
||
logger.info("\n收到中断信号,程序退出。")
|
||
next_task = self._stop_current_task(ctx, mon, worker, {"category": "USER_INTERRUPT", "code": 0, "reason": "User interrupted"})
|
||
interrupted_this_task = True
|
||
except Exception as e:
|
||
print(f"处理任务时出错: {e}")
|
||
traceback.print_exc()
|
||
failed_stage = None
|
||
if mon and "collect" in mon.stage_started_at:
|
||
failed_stage = "collect"
|
||
mon.finish_stage("collect", status="failed")
|
||
elif mon and "download" in mon.stage_started_at:
|
||
failed_stage = "download"
|
||
mon.finish_stage("download", status="failed")
|
||
next_task = self._report_terminal(
|
||
ctx,
|
||
mon or TaskMonitor(worker, ctx.task_key),
|
||
worker,
|
||
status="failed",
|
||
failed_stage=failed_stage,
|
||
error={"category": "UNEXPECTED", "code": 0, "reason": str(e)},
|
||
error_message=str(e),
|
||
)
|
||
try:
|
||
task = self._finalize_with_recovery(ctx, worker, next_task)
|
||
except FinalizeRecoveryFailed as e:
|
||
fatal_error = e
|
||
break
|
||
if interrupted_this_task:
|
||
interrupted = True
|
||
break
|
||
|
||
logger.info("任务循环结束，停止后台网络监控...")
|
||
self.network_watchdog.stop()
|
||
if fatal_error:
|
||
raise fatal_error
|
||
if interrupted:
|
||
raise KeyboardInterrupt
|
||
print("\n所有任务已完成或无更多任务")
|
||
|
||
|
||
|
||
if __name__ == '__main__':
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(
|
||
description="autool 批量/单应用测试运行器",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog="""示例:
|
||
# 标准批量模式（连接 Redis 中控端）
|
||
python batch_run.py
|
||
|
||
# 单应用本地测试模式
|
||
python batch_run.py --single --package com.google.android.youtube
|
||
python batch_run.py --single --package com.tencent.mm --app-name "微信" --country CN --keep-app
|
||
""",
|
||
)
|
||
parser.add_argument(
|
||
"--single", action="store_true",
|
||
help="启用单应用本地测试模式（不连接 Redis 中控端）",
|
||
)
|
||
parser.add_argument(
|
||
"--package", type=str, default=None,
|
||
help="应用包名（--single 模式下必需）",
|
||
)
|
||
parser.add_argument(
|
||
"--app-name", type=str, default=None,
|
||
help="应用显示名称（可选，默认取包名最后一段）",
|
||
)
|
||
parser.add_argument(
|
||
"--country", type=str, default="US",
|
||
help="国家码，默认 US",
|
||
)
|
||
parser.add_argument(
|
||
"--keep-app", action="store_true",
|
||
help="测试完成后保留应用（默认卸载）",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
# 清理自定义参数，避免 airtest 的 cli_setup 解析时报 unrecognized arguments
|
||
sys.argv = [sys.argv[0]]
|
||
|
||
if args.single:
|
||
if not args.package:
|
||
parser.error("--single 模式必须指定 --package")
|
||
|
||
app_name = args.app_name or args.package.split(".")[-1]
|
||
task_key = f"single-{args.package}-{datetime.now().strftime('%Y%m%d%H%M%S')}"
|
||
|
||
task = {
|
||
"app_name": app_name,
|
||
"package_name": args.package,
|
||
"task_key": task_key,
|
||
"country_codes": [args.country.upper()],
|
||
"available_sources": ["google_play", "local"],
|
||
"keep_app_installed": args.keep_app,
|
||
}
|
||
|
||
config = load_config()
|
||
print(f"\n{'='*50}")
|
||
print(f"[单应用模式]")
|
||
print(f" 应用: {app_name}")
|
||
print(f" 包名: {args.package}")
|
||
print(f" 国家码: {args.country.upper()}")
|
||
print(f" 保留应用: {args.keep_app}")
|
||
print(f"{'='*50}\n")
|
||
|
||
if not check_network():
|
||
logger.warning("初始网络检查失败，等待网络恢复...")
|
||
wait_for_network(check_interval=5)
|
||
|
||
executor = TestExecutor(worker=NullTaskWorker())
|
||
executor.setup()
|
||
executor.run(initial_task=task)
|
||
print("\n[单应用模式] 测试完成。")
|
||
|
||
else:
|
||
round_count = 0
|
||
stats = None
|
||
config = load_config()
|
||
|
||
while True:
|
||
try:
|
||
if not check_network():
|
||
logger.warning("初始网络检查失败，等待网络恢复...")
|
||
wait_for_network(check_interval=5)
|
||
|
||
if stats is None:
|
||
stats = BatchStatistics()
|
||
|
||
executor = TestExecutor(stats=stats)
|
||
|
||
executor.setup()
|
||
round_count += 1
|
||
executor.current_round = round_count
|
||
|
||
logger.info(f"========== 开始第 {round_count} 轮测试 ==========")
|
||
executor.run()
|
||
|
||
summary = stats.get_round_summary(round_count)
|
||
logger.info(f"========== 第 {round_count} 轮测试完成 ==========")
|
||
logger.info(f"本轮统计: 总计 {summary['total']} 个应用, 成功 {summary['success']}, 失败 {summary['failed']}, 重试 {summary['retry_count']} 次")
|
||
|
||
except KeyboardInterrupt:
|
||
logger.info(f"\n收到中断信号，共完成 {round_count} 轮测试，程序退出。")
|
||
raise
|
||
|
||
except Exception as e:
|
||
logger.error(f"测试过程中发生未预期错误: {e}")
|
||
logger.error(traceback.format_exc())
|
||
msg = f"测试过程中发生未预期错误: {e}"
|
||
if 'executor' in locals():
|
||
pass
|
||
|
||
finally:
|
||
if 'executor' in locals() and executor and hasattr(executor, 'network_watchdog'):
|
||
logger.info("正在清理网络监控资源...")
|
||
executor.network_watchdog.shutdown()
|