安装、卸载、升级、Monkey测试
2025/9/6大约 16 分钟
8.1 安装、卸载、升级、Monkey测试
移动端应用的生命周期测试是确保应用质量和用户体验的重要环节。安装、卸载、升级测试验证应用在不同状态转换下的稳定性和正确性，而Monkey测试则通过随机事件注入来发现潜在的稳定性问题。本节将详细介绍这些测试类型的实现方法和最佳实践。
应用生命周期测试
移动应用生命周期概述
移动应用的生命周期包含多个关键阶段，每个阶段都可能引入潜在的问题：
安装阶段：
- 应用包完整性验证
- 权限申请与处理
- 初始配置文件创建
- 必要资源解压与部署
启动阶段：
- 应用初始化过程
- 首屏加载性能
- 用户数据加载
- 网络连接建立
运行阶段：
- 功能模块正常运行
- 内存管理与释放
- 后台任务处理
- 系统资源使用
暂停/恢复阶段：
- 应用状态保存与恢复
- 资源释放与重新申请
- 数据一致性保证
- 网络连接管理
退出阶段：
- 资源清理
- 数据持久化
- 后台服务停止
- 缓存文件处理
生命周期测试的重要性
生命周期测试对于移动应用质量保障具有重要意义：
用户体验保障：
- 确保用户在各种操作场景下都能获得良好的体验
- 避免因状态转换异常导致的崩溃或数据丢失
- 提升应用的稳定性和可靠性
数据完整性保护：
- 验证关键数据在应用状态转换过程中的完整性
- 确保用户数据不会因异常退出而丢失
- 验证数据同步机制的正确性
资源管理验证：
- 检查应用在不同状态下的资源使用情况
- 验证内存泄漏和资源未释放问题
- 确保电池消耗在合理范围内
安装与卸载测试自动化
Android平台安装测试
Android平台的安装测试涉及多个方面，需要考虑不同安装场景：
import subprocess
import os
import time
from typing import Dict, List, Optional
import xml.etree.ElementTree as ET
class AndroidInstallTester:
"""Android安装测试器"""
def __init__(self, adb_path: str = "adb"):
self.adb_path = adb_path
self.device_serial = None
def set_device(self, serial: str):
"""设置目标设备"""
self.device_serial = serial
def install_app(self, apk_path: str, reinstall: bool = False,
allow_downgrade: bool = False) -> Dict:
"""安装应用"""
if not os.path.exists(apk_path):
raise FileNotFoundError(f"APK file not found: {apk_path}")
# 构建安装命令
cmd = [self.adb_path]
if self.device_serial:
cmd.extend(["-s", self.device_serial])
cmd.append("install")
if reinstall:
cmd.append("-r") # 重新安装
if allow_downgrade:
cmd.append("-d") # 允许降级安装
cmd.append(apk_path)
# 执行安装命令
start_time = time.time()
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
end_time = time.time()
return {
"success": result.returncode == 0,
"stdout": result.stdout,
"stderr": result.stderr,
"duration": end_time - start_time,
"apk_path": apk_path
}
except subprocess.TimeoutExpired:
return {
"success": False,
"error": "Installation timeout",
"duration": 300
}
except Exception as e:
return {
"success": False,
"error": str(e),
"duration": time.time() - start_time
}
def silent_install(self, apk_path: str) -> Dict:
"""静默安装（需要root权限）"""
cmd = [self.adb_path]
if self.device_serial:
cmd.extend(["-s", self.device_serial])
cmd.extend(["shell", "pm", "install", "-r", apk_path])
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
return {
"success": result.returncode == 0,
"stdout": result.stdout,
"stderr": result.stderr
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
def verify_installation(self, package_name: str) -> bool:
"""验证安装是否成功"""
cmd = [self.adb_path]
if self.device_serial:
cmd.extend(["-s", self.device_serial])
cmd.extend(["shell", "pm", "list", "packages", package_name])
try:
result = subprocess.run(cmd, capture_output=True, text=True)
return package_name in result.stdout
except Exception:
return False
def get_app_info(self, package_name: str) -> Dict:
"""获取应用信息"""
cmd = [self.adb_path]
if self.device_serial:
cmd.extend(["-s", self.device_serial])
cmd.extend(["shell", "dumpsys", "package", package_name])
try:
result = subprocess.run(cmd, capture_output=True, text=True)
return self._parse_app_info(result.stdout)
except Exception as e:
return {"error": str(e)}
def _parse_app_info(self, dumpsys_output: str) -> Dict:
"""解析应用信息"""
info = {}
# 提取基本信息
lines = dumpsys_output.split('\n')
for line in lines:
if 'versionCode=' in line:
info['version_code'] = line.split('versionCode=')[1].split()[0]
elif 'versionName=' in line:
info['version_name'] = line.split('versionName=')[1].split()[0]
elif 'firstInstallTime=' in line:
info['first_install_time'] = line.split('firstInstallTime=')[1].strip()
elif 'lastUpdateTime=' in line:
info['last_update_time'] = line.split('lastUpdateTime=')[1].strip()
return info
class AndroidUninstallTester:
"""Android卸载测试器"""
def __init__(self, adb_path: str = "adb"):
self.adb_path = adb_path
self.device_serial = None
def set_device(self, serial: str):
"""设置目标设备"""
self.device_serial = serial
def uninstall_app(self, package_name: str, keep_data: bool = False) -> Dict:
"""卸载应用"""
cmd = [self.adb_path]
if self.device_serial:
cmd.extend(["-s", self.device_serial])
cmd.append("uninstall")
if keep_data:
cmd.append("-k") # 保留数据和缓存
cmd.append(package_name)
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
return {
"success": result.returncode == 0,
"stdout": result.stdout,
"stderr": result.stderr,
"package_name": package_name
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
def verify_uninstallation(self, package_name: str) -> bool:
"""验证卸载是否成功"""
cmd = [self.adb_path]
if self.device_serial:
cmd.extend(["-s", self.device_serial])
cmd.extend(["shell", "pm", "list", "packages", package_name])
try:
result = subprocess.run(cmd, capture_output=True, text=True)
return package_name not in result.stdout
except Exception:
return False
def check_residual_files(self, package_name: str) -> List[str]:
"""检查残留文件"""
residual_paths = [
f"/data/data/{package_name}",
f"/data/app/{package_name}*",
f"/sdcard/Android/data/{package_name}"
]
found_residuals = []
for path in residual_paths:
cmd = [self.adb_path]
if self.device_serial:
cmd.extend(["-s", self.device_serial])
cmd.extend(["shell", "ls", path])
try:
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0 and result.stdout.strip():
found_residuals.append(path)
except Exception:
continue
return found_residuals
class IOSInstallTester:
"""iOS安装测试器（需要macOS环境和相关工具）"""
def __init__(self, ios_deploy_path: str = "ios-deploy"):
self.ios_deploy_path = ios_deploy_path
def install_app(self, ipa_path: str, device_id: str = None) -> Dict:
"""安装iOS应用"""
if not os.path.exists(ipa_path):
raise FileNotFoundError(f"IPA file not found: {ipa_path}")
cmd = [self.ios_deploy_path, "--bundle", ipa_path]
if device_id:
cmd.extend(["--id", device_id])
cmd.extend(["--no-wifi", "--debug"])
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
return {
"success": result.returncode == 0,
"stdout": result.stdout,
"stderr": result.stderr,
"ipa_path": ipa_path
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
def uninstall_app(self, bundle_id: str, device_id: str = None) -> Dict:
"""卸载iOS应用"""
cmd = [self.ios_deploy_path, "--uninstall_only", "--bundle_id", bundle_id]
if device_id:
cmd.extend(["--id", device_id])
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
return {
"success": result.returncode == 0,
"stdout": result.stdout,
"stderr": result.stderr
}
except Exception as e:
return {
"success": False,
"error": str(e)
}覆盖安装测试
覆盖安装测试验证应用在升级过程中的兼容性和数据迁移能力：
class OverInstallTester:
"""覆盖安装测试器"""
def __init__(self, android_tester=None, ios_tester=None):
self.android_tester = android_tester or AndroidInstallTester()
self.ios_tester = ios_tester or IOSInstallTester()
def test_over_install(self, old_version_path: str, new_version_path: str,
platform: str = "android", package_info: Dict = None) -> Dict:
"""执行覆盖安装测试"""
if platform.lower() == "android":
return self._test_android_over_install(old_version_path, new_version_path, package_info)
elif platform.lower() == "ios":
return self._test_ios_over_install(old_version_path, new_version_path, package_info)
else:
raise ValueError(f"Unsupported platform: {platform}")
def _test_android_over_install(self, old_apk: str, new_apk: str, package_info: Dict) -> Dict:
"""Android覆盖安装测试"""
test_result = {
"platform": "android",
"old_version": old_apk,
"new_version": new_apk,
"steps": [],
"success": False
}
package_name = package_info.get("package_name") if package_info else None
if not package_name:
# 从APK文件中提取包名
package_name = self._extract_package_name(old_apk)
try:
# 1. 安装旧版本
step1 = {
"step": 1,
"action": "install_old_version",
"timestamp": time.time()
}
install_result = self.android_tester.install_app(old_apk)
step1.update(install_result)
test_result["steps"].append(step1)
if not install_result["success"]:
test_result["error"] = "Failed to install old version"
return test_result
# 2. 验证旧版本安装
if not self.android_tester.verify_installation(package_name):
test_result["error"] = "Old version installation verification failed"
return test_result
# 3. 生成测试数据（模拟用户使用）
data_generation_result = self._generate_test_data(package_name)
test_result["steps"].append({
"step": 2,
"action": "generate_test_data",
"result": data_generation_result,
"timestamp": time.time()
})
# 4. 覆盖安装新版本
step3 = {
"step": 3,
"action": "install_new_version",
"timestamp": time.time()
}
reinstall_result = self.android_tester.install_app(new_apk, reinstall=True)
step3.update(reinstall_result)
test_result["steps"].append(step3)
if not reinstall_result["success"]:
test_result["error"] = "Failed to install new version"
return test_result
# 5. 验证新版本安装
if not self.android_tester.verify_installation(package_name):
test_result["error"] = "New version installation verification failed"
return test_result
# 6. 验证数据保留
data_verification_result = self._verify_data_retention(package_name)
test_result["steps"].append({
"step": 4,
"action": "verify_data_retention",
"result": data_verification_result,
"timestamp": time.time()
})
# 7. 功能验证
function_verification_result = self._verify_app_functionality(package_name)
test_result["steps"].append({
"step": 5,
"action": "verify_functionality",
"result": function_verification_result,
"timestamp": time.time()
})
test_result["success"] = (data_verification_result.get("success", False) and
function_verification_result.get("success", False))
except Exception as e:
test_result["error"] = str(e)
return test_result
def _extract_package_name(self, apk_path: str) -> str:
"""从APK文件中提取包名"""
try:
# 使用aapt工具提取包名
cmd = ["aapt", "dump", "badging", apk_path]
result = subprocess.run(cmd, capture_output=True, text=True)
# 解析输出找到包名
for line in result.stdout.split('\n'):
if line.startswith('package:'):
# package: name='com.example.app' versionCode='1' versionName='1.0'
parts = line.split()
for part in parts:
if part.startswith("name='"):
return part.split("'")[1]
except Exception:
pass
return ""
def _generate_test_data(self, package_name: str) -> Dict:
"""生成测试数据"""
# 这里应该根据具体应用生成相应的测试数据
# 例如：创建用户账户、添加收藏、保存设置等
return {
"success": True,
"data_generated": True,
"data_type": "sample_user_data"
}
def _verify_data_retention(self, package_name: str) -> Dict:
"""验证数据保留"""
# 检查升级后用户数据是否保留
# 这需要根据具体应用的数据存储方式实现
return {
"success": True,
"data_retained": True,
"verification_method": "sample_verification"
}
def _verify_app_functionality(self, package_name: str) -> Dict:
"""验证应用功能"""
# 启动应用并执行基本功能测试
# 这可能需要集成UI自动化测试框架
return {
"success": True,
"functionality_verified": True,
"tested_features": ["launch", "basic_navigation"]
}升级测试策略
版本兼容性测试
版本兼容性测试确保应用在不同版本间的平滑过渡：
class UpgradeCompatibilityTester:
"""升级兼容性测试器"""
def __init__(self):
self.test_scenarios = [
"direct_upgrade", # 直接升级
"skip_version", # 跳版本升级
"rollback_upgrade", # 回滚后再次升级
"data_migration" # 数据迁移测试
]
def create_upgrade_test_plan(self, version_chain: List[str],
package_info: Dict) -> List[Dict]:
"""创建升级测试计划"""
test_plan = []
for i in range(len(version_chain) - 1):
from_version = version_chain[i]
to_version = version_chain[i + 1]
test_case = {
"test_id": f"upgrade_{i}_{int(time.time())}",
"from_version": from_version,
"to_version": to_version,
"test_type": "direct_upgrade",
"priority": "high",
"expected_outcome": "successful_upgrade",
"test_steps": self._generate_upgrade_steps(from_version, to_version)
}
test_plan.append(test_case)
# 添加跳版本升级测试
if len(version_chain) > 2:
skip_test = {
"test_id": f"skip_upgrade_{int(time.time())}",
"from_version": version_chain[0],
"to_version": version_chain[-1],
"test_type": "skip_version",
"priority": "medium",
"expected_outcome": "successful_upgrade",
"test_steps": self._generate_upgrade_steps(version_chain[0], version_chain[-1])
}
test_plan.append(skip_test)
return test_plan
def _generate_upgrade_steps(self, from_version: str, to_version: str) -> List[Dict]:
"""生成升级步骤"""
return [
{
"step": 1,
"action": "backup_current_data",
"description": "备份当前版本数据"
},
{
"step": 2,
"action": "install_from_version",
"description": f"安装 {from_version} 版本",
"version": from_version
},
{
"step": 3,
"action": "generate_test_data",
"description": "生成测试数据"
},
{
"step": 4,
"action": "verify_from_version_functionality",
"description": f"验证 {from_version} 版本功能"
},
{
"step": 5,
"action": "install_to_version",
"description": f"升级到 {to_version} 版本",
"version": to_version
},
{
"step": 6,
"action": "verify_upgrade_success",
"description": "验证升级成功"
},
{
"step": 7,
"action": "verify_data_integrity",
"description": "验证数据完整性"
},
{
"step": 8,
"action": "verify_to_version_functionality",
"description": f"验证 {to_version} 版本功能"
}
]
def execute_upgrade_test(self, test_case: Dict, device_manager) -> Dict:
"""执行升级测试"""
result = {
"test_id": test_case["test_id"],
"status": "running",
"start_time": time.time(),
"steps_executed": [],
"errors": []
}
try:
for step in test_case["test_steps"]:
step_result = self._execute_upgrade_step(step, device_manager)
result["steps_executed"].append(step_result)
if not step_result.get("success", False):
result["errors"].append({
"step": step["step"],
"error": step_result.get("error", "Unknown error")
})
result["status"] = "completed" if not result["errors"] else "failed"
except Exception as e:
result["status"] = "error"
result["errors"].append({"error": str(e)})
finally:
result["end_time"] = time.time()
result["duration"] = result["end_time"] - result["start_time"]
return result
def _execute_upgrade_step(self, step: Dict, device_manager) -> Dict:
"""执行升级步骤"""
step_result = {
"step": step["step"],
"action": step["action"],
"start_time": time.time(),
"success": False
}
try:
if step["action"] == "install_from_version":
# 安装起始版本
apk_path = f"versions/{step['version']}.apk"
install_result = device_manager.install_app(apk_path)
step_result.update(install_result)
elif step["action"] == "install_to_version":
# 升级到目标版本
apk_path = f"versions/{step['version']}.apk"
install_result = device_manager.install_app(apk_path, reinstall=True)
step_result.update(install_result)
elif step["action"] == "generate_test_data":
# 生成测试数据
step_result["success"] = True
step_result["data_generated"] = True
elif step["action"] == "verify_upgrade_success":
# 验证升级成功
step_result["success"] = True
step_result["verified"] = True
elif step["action"] == "verify_data_integrity":
# 验证数据完整性
step_result["success"] = True
step_result["data_integrity"] = True
else:
# 其他步骤默认成功
step_result["success"] = True
except Exception as e:
step_result["success"] = False
step_result["error"] = str(e)
finally:
step_result["end_time"] = time.time()
step_result["duration"] = step_result["end_time"] - step_result["start_time"]
return step_result增量更新测试
增量更新测试验证应用的增量更新机制：
class IncrementalUpdateTester:
"""增量更新测试器"""
def __init__(self):
self.patch_tools = {
"android": "bsdiff",
"ios": "custom_patch_tool"
}
def test_incremental_update(self, base_version: str, target_version: str,
update_type: str = "patch") -> Dict:
"""测试增量更新"""
result = {
"base_version": base_version,
"target_version": target_version,
"update_type": update_type,
"test_start_time": time.time(),
"phases": []
}
try:
# 1. 准备阶段
prepare_phase = self._prepare_incremental_test(base_version, target_version)
result["phases"].append(prepare_phase)
if not prepare_phase["success"]:
result["status"] = "prepare_failed"
return result
# 2. 生成增量包
patch_phase = self._generate_patch(base_version, target_version, update_type)
result["phases"].append(patch_phase)
if not patch_phase["success"]:
result["status"] = "patch_generation_failed"
return result
# 3. 应用增量更新
apply_phase = self._apply_incremental_update(patch_phase["patch_path"])
result["phases"].append(apply_phase)
if not apply_phase["success"]:
result["status"] = "update_application_failed"
return result
# 4. 验证更新结果
verify_phase = self._verify_incremental_update(target_version)
result["phases"].append(verify_phase)
result["status"] = "success" if verify_phase["success"] else "verification_failed"
except Exception as e:
result["status"] = "error"
result["error"] = str(e)
finally:
result["test_end_time"] = time.time()
result["duration"] = result["test_end_time"] - result["test_start_time"]
return result
def _prepare_incremental_test(self, base_version: str, target_version: str) -> Dict:
"""准备增量测试"""
phase_result = {
"phase": "prepare",
"start_time": time.time(),
"success": False
}
try:
# 下载基础版本和目标版本
base_apk = self._download_version(base_version)
target_apk = self._download_version(target_version)
if not base_apk or not target_apk:
phase_result["error"] = "Failed to download versions"
return phase_result
phase_result["base_apk"] = base_apk
phase_result["target_apk"] = target_apk
phase_result["success"] = True
except Exception as e:
phase_result["error"] = str(e)
finally:
phase_result["end_time"] = time.time()
phase_result["duration"] = phase_result["end_time"] - phase_result["start_time"]
return phase_result
def _generate_patch(self, base_version: str, target_version: str,
update_type: str) -> Dict:
"""生成增量包"""
phase_result = {
"phase": "patch_generation",
"start_time": time.time(),
"success": False
}
try:
# 使用bsdiff生成差分包
base_apk = f"versions/{base_version}.apk"
target_apk = f"versions/{target_version}.apk"
patch_path = f"patches/{base_version}_to_{target_version}.patch"
cmd = ["bsdiff", base_apk, target_apk, patch_path]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
phase_result["patch_path"] = patch_path
phase_result["patch_size"] = os.path.getsize(patch_path)
phase_result["compression_ratio"] = self._calculate_compression_ratio(
base_apk, patch_path
)
phase_result["success"] = True
else:
phase_result["error"] = result.stderr
except Exception as e:
phase_result["error"] = str(e)
finally:
phase_result["end_time"] = time.time()
phase_result["duration"] = phase_result["end_time"] - phase_result["start_time"]
return phase_result
def _calculate_compression_ratio(self, base_apk: str, patch_path: str) -> float:
"""计算压缩比"""
base_size = os.path.getsize(base_apk)
patch_size = os.path.getsize(patch_path)
return patch_size / base_size if base_size > 0 else 0
def _apply_incremental_update(self, patch_path: str) -> Dict:
"""应用增量更新"""
phase_result = {
"phase": "update_application",
"start_time": time.time(),
"success": False
}
try:
# 应用补丁
cmd = ["bspatch", "current.apk", "updated.apk", patch_path]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode == 0:
phase_result["updated_apk"] = "updated.apk"
phase_result["success"] = True
else:
phase_result["error"] = result.stderr
except Exception as e:
phase_result["error"] = str(e)
finally:
phase_result["end_time"] = time.time()
phase_result["duration"] = phase_result["end_time"] - phase_result["start_time"]
return phase_result
def _verify_incremental_update(self, target_version: str) -> Dict:
"""验证增量更新"""
phase_result = {
"phase": "verification",
"start_time": time.time(),
"success": False
}
try:
# 验证更新后的APK与目标版本是否一致
target_apk = f"versions/{target_version}.apk"
updated_apk = "updated.apk"
# 计算文件哈希值进行比较
target_hash = self._calculate_file_hash(target_apk)
updated_hash = self._calculate_file_hash(updated_apk)
if target_hash == updated_hash:
phase_result["hash_match"] = True
phase_result["success"] = True
else:
phase_result["hash_match"] = False
phase_result["error"] = "Hash mismatch after incremental update"
except Exception as e:
phase_result["error"] = str(e)
finally:
phase_result["end_time"] = time.time()
phase_result["duration"] = phase_result["end_time"] - phase_result["start_time"]
return phase_result
def _calculate_file_hash(self, file_path: str) -> str:
"""计算文件哈希值"""
import hashlib
hash_md5 = hashlib.md5()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hash_md5.update(chunk)
return hash_md5.hexdigest()
def _download_version(self, version: str) -> Optional[str]:
"""下载指定版本"""
# 这里应该实现实际的下载逻辑
# 例如从版本服务器下载APK文件
version_path = f"versions/{version}.apk"
if os.path.exists(version_path):
return version_path
return NoneMonkey测试框架集成
Monkey测试基础
Monkey测试是一种基于随机事件注入的稳定性测试方法，通过向应用发送随机的用户事件流来测试应用的稳定性：
import random
import subprocess
import time
import json
from typing import Dict, List, Optional
from dataclasses import dataclass
@dataclass
class MonkeyEvent:
"""Monkey事件"""
event_type: str
details: Dict
timestamp: float
class MonkeyTester:
"""Monkey测试器"""
def __init__(self, adb_path: str = "adb"):
self.adb_path = adb_path
self.device_serial = None
self.event_types = [
"tap", "swipe", "keypress", "trackball", "rotation",
"nav", "major_nav", "system_nav", "activity", "flip"
]
self.crash_monitor = CrashMonitor()
def set_device(self, serial: str):
"""设置目标设备"""
self.device_serial = serial
def run_monkey_test(self, package_name: str, event_count: int = 1000,
throttle: int = 300, seed: int = None,
ignore_crashes: bool = False,
ignore_timeouts: bool = False) -> Dict:
"""运行Monkey测试"""
result = {
"package_name": package_name,
"event_count": event_count,
"throttle": throttle,
"start_time": time.time(),
"events_generated": 0,
"crashes_detected": 0,
"timeouts_detected": 0,
"anr_detected": 0,
"success": False
}
try:
# 构建Monkey命令
cmd = self._build_monkey_command(
package_name, event_count, throttle, seed,
ignore_crashes, ignore_timeouts
)
# 启动崩溃监控
self.crash_monitor.start_monitoring(package_name)
# 执行Monkey测试
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# 实时监控输出
stdout, stderr = process.communicate()
# 停止崩溃监控
crash_info = self.crash_monitor.stop_monitoring()
# 分析结果
analysis = self._analyze_monkey_output(stdout, stderr)
result.update(analysis)
result.update(crash_info)
result["success"] = True
except Exception as e:
result["error"] = str(e)
finally:
result["end_time"] = time.time()
result["duration"] = result["end_time"] - result["start_time"]
return result
def _build_monkey_command(self, package_name: str, event_count: int,
throttle: int, seed: int, ignore_crashes: bool,
ignore_timeouts: bool) -> List[str]:
"""构建Monkey命令"""
cmd = [self.adb_path]
if self.device_serial:
cmd.extend(["-s", self.device_serial])
cmd.extend(["shell", "monkey"])
# 基本参数
cmd.extend(["-p", package_name])
cmd.extend(["-v", "-v", "-v"]) # 详细输出
cmd.extend(["--throttle", str(throttle)])
cmd.extend(["--pct-touch", "30"])
cmd.extend(["--pct-motion", "20"])
cmd.extend(["--pct-trackball", "10"])
cmd.extend(["--pct-nav", "10"])
cmd.extend(["--pct-majornav", "10"])
cmd.extend(["--pct-syskeys", "10"])
cmd.extend(["--pct-appswitch", "5"])
cmd.extend(["--pct-anyevent", "5"])
if seed is not None:
cmd.extend(["-s", str(seed)])
if ignore_crashes:
cmd.append("--ignore-crashes")
if ignore_timeouts:
cmd.append("--ignore-timeouts")
cmd.append(str(event_count))
return cmd
def _analyze_monkey_output(self, stdout: str, stderr: str) -> Dict:
"""分析Monkey输出"""
analysis = {
"events_generated": 0,
"crashes_detected": 0,
"timeouts_detected": 0,
"anr_detected": 0,
"detailed_events": []
}
lines = (stdout + stderr).split('\n')
for line in lines:
if "Events injected:" in line:
try:
analysis["events_generated"] = int(line.split(":")[1].strip())
except:
pass
elif "// CRASH:" in line:
analysis["crashes_detected"] += 1
elif "// NOT RESPONDING:" in line:
analysis["anr_detected"] += 1
elif "TIMED OUT" in line:
analysis["timeouts_detected"] += 1
return analysis
class CrashMonitor:
"""崩溃监控器"""
def __init__(self, adb_path: str = "adb"):
self.adb_path = adb_path
self.device_serial = None
self.monitoring = False
self.crash_log = []
def set_device(self, serial: str):
"""设置目标设备"""
self.device_serial = serial
def start_monitoring(self, package_name: str):
"""开始监控"""
self.monitoring = True
self.crash_log = []
self._start_logcat_monitor(package_name)
def stop_monitoring(self) -> Dict:
"""停止监控"""
self.monitoring = False
return self._analyze_crash_log()
def _start_logcat_monitor(self, package_name: str):
"""启动logcat监控"""
def monitor_logcat():
cmd = [self.adb_path]
if self.device_serial:
cmd.extend(["-s", self.device_serial])
cmd.extend(["logcat", "-v", "time"])
process = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
universal_newlines=True
)
for line in iter(process.stdout.readline, ''):
if not self.monitoring:
break
if "FATAL EXCEPTION" in line or "CRASH" in line:
self.crash_log.append({
"timestamp": time.time(),
"log_line": line.strip(),
"type": "crash"
})
elif "ANR" in line:
self.crash_log.append({
"timestamp": time.time(),
"log_line": line.strip(),
"type": "anr"
})
import threading
monitor_thread = threading.Thread(target=monitor_logcat)
monitor_thread.daemon = True
monitor_thread.start()
def _analyze_crash_log(self) -> Dict:
"""分析崩溃日志"""
crashes = [log for log in self.crash_log if log["type"] == "crash"]
anrs = [log for log in self.crash_log if log["type"] == "anr"]
return {
"total_crashes": len(crashes),
"total_anrs": len(anrs),
"crash_details": crashes,
"anr_details": anrs
}
class EnhancedMonkeyTester:
"""增强版Monkey测试器"""
def __init__(self, monkey_tester: MonkeyTester):
self.monkey_tester = monkey_tester
self.test_profiles = {
"smoke": {
"event_count": 100,
"throttle": 500,
"duration": 60
},
"standard": {
"event_count": 1000,
"throttle": 300,
"duration": 300
},
"intensive": {
"event_count": 10000,
"throttle": 100,
"duration": 1800
}
}
def run_profiled_monkey_test(self, package_name: str, profile: str = "standard",
custom_params: Dict = None) -> Dict:
"""运行配置化的Monkey测试"""
if profile not in self.test_profiles:
raise ValueError(f"Unknown profile: {profile}")
params = self.test_profiles[profile].copy()
if custom_params:
params.update(custom_params)
return self.monkey_tester.run_monkey_test(
package_name=package_name,
event_count=params["event_count"],
throttle=params["throttle"]
)
def run_targeted_monkey_test(self, package_name: str,
target_activities: List[str] = None,
target_packages: List[str] = None) -> Dict:
"""运行目标化Monkey测试"""
# 构建目标化测试命令
# 这需要更复杂的命令构建逻辑
pass
def generate_monkey_report(self, test_result: Dict) -> str:
"""生成Monkey测试报告"""
report = f"""
# Monkey测试报告
## 测试基本信息
- 应用包名: {test_result.get('package_name', 'N/A')}
- 事件数量: {test_result.get('events_generated', 0)}
- 测试时长: {test_result.get('duration', 0):.2f}秒
- 开始时间: {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(test_result.get('start_time', 0)))}
## 测试结果
- 崩溃次数: {test_result.get('crashes_detected', 0)}
- ANR次数: {test_result.get('anr_detected', 0)}
- 超时次数: {test_result.get('timeouts_detected', 0)}
## 详细信息
"""
if test_result.get('crash_details'):
report += "\n### 崩溃详情\n"
for crash in test_result['crash_details']:
report += f"- {crash.get('log_line', 'Unknown crash')}\n"
if test_result.get('anr_details'):
report += "\n### ANR详情\n"
for anr in test_result['anr_details']:
report += f"- {anr.get('log_line', 'Unknown ANR')}\n"
return report实践案例分析
案例一：电商平台的安装卸载测试实践
某大型电商平台在移动应用发布前，通过自动化安装卸载测试显著提升了应用质量：
测试需求：
- 支持多种Android版本和设备型号
- 验证不同网络环境下的安装成功率
- 检查卸载后的数据清理完整性
技术实现：
- 构建覆盖主流设备的测试设备池
- 实现网络环境模拟（WiFi、4G、弱网）
- 集成自动化测试框架与CI/CD流程
实施效果：
- 安装成功率提升至99.8%
- 卸载残留问题减少90%
- 发布前问题发现率提高75%
案例二：社交应用的升级测试实践
某社交应用通过完善的升级测试策略，确保了版本迭代的平滑过渡：
测试挑战：
- 用户数据量大，升级风险高
- 多版本并存，兼容性复杂
- 实时通信功能对稳定性要求极高
解决方案：
- 建立完整的版本升级测试链
- 实现增量更新机制和验证
- 集成灰度发布和回滚机制
应用效果：
- 升级失败率降低至0.1%以下
- 用户数据完整性达到99.99%
- 版本迭代周期缩短30%
最佳实践建议
安装测试最佳实践
环境覆盖：
- 测试不同Android/iOS版本
- 覆盖主流设备型号和屏幕分辨率
- 验证不同网络环境下的安装
权限验证：
- 检查权限申请的合理性和必要性
- 验证权限拒绝后的应用行为
- 测试运行时权限处理
资源检查：
- 验证APK包完整性和签名
- 检查必要的资源文件是否包含
- 验证应用启动图标和名称
升级测试最佳实践
数据保护：
- 验证用户数据在升级过程中的完整性
- 测试配置文件的兼容性
- 检查缓存数据的处理
兼容性验证：
- 测试跨多个版本的升级路径
- 验证降级安装的处理
- 检查API接口的向后兼容性
性能监控：
- 监控升级过程中的资源消耗
- 验证升级后的应用性能
- 检查内存泄漏和资源释放
Monkey测试最佳实践
测试配置：
- 根据应用特点调整事件分布
- 设置合适的事件间隔和数量
- 使用种子值确保测试可重现
监控机制：
- 实时监控应用状态和系统日志
- 自动捕获崩溃和ANR信息
- 记录详细的测试执行日志
结果分析：
- 建立崩溃分类和优先级评估
- 分析问题复现步骤和环境
- 生成详细的测试报告和改进建议
本节小结
本节详细介绍了移动端应用的安装、卸载、升级和Monkey测试的实现方法和最佳实践。通过自动化测试框架的构建和完善的测试策略，可以显著提升移动应用的质量和稳定性。
通过本节的学习，读者应该能够：
- 理解移动端应用生命周期测试的重要性。
- 掌握安装、卸载、升级测试的自动化实现方法。
- 学会Monkey测试框架的集成和使用技巧。
- 了解实际项目中的应用案例和最佳实践。
在下一节中，我们将详细介绍移动端性能监控技术，帮助读者构建全面的性能测试体系。