背景

  稳定性测试时，需要跑整机monkey，实际测试中发现：monkey打开应用时，无法点击“权限弹窗”，导致monkey无法进入应用首页。手动打开每个应用授权，非常耗时，基于此背景开发该工具。

一键停止Monkey批处理

@ECHO off  
@REM 无限循环的标签  

ECHO wait-for-device...
adb wait-for-device
ECHO Device connected.
echo.
:LOOP
@REM 拿到Monkey进程id
adb shell "kill -9 `ps -ef | grep com.android.commands.monkey | grep -v 'grep' | awk '{print $2}'`"
@REM 清除时钟的闹钟
adb shell pm clear com.android.deskclock
ECHO Monkey hat stopped.
echo.
PAUSE
GOTO LOOP  
@ECHO on

整机Monkey测试源码

# -*- coding: utf-8 -*-
# @Time    : 2026/4/9 14:00
# @FileName: 整机monkey.py
# @Author: 1159533975@qq.com
# @Software: PyCharm
import os
import re
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor

# 黑名单，monkey测试不跑的包
black_list = [
    # MTK log
    "com.debug.loggerui",
    "com.mediatek.engineermode",
    # 展讯 log
    "com.sprd.logmanager",
    "com.sprd.engineermode",
    # RK log
    "com.incar.rkylog",
    # 全志 log
    "com.softwinner.awlogsettings",
    # MTK非框架行动定位服务
    "com.mediatek.gnss.nonframeworklbs"
]

# 创建adb shell会话, 大大提升adb shell性能
class ADB_SHELL:
    def __init__(self):
        """
        简介：
            使用subprocess开启一个adb shell子线程并保持会话状态
            在频繁使用adb shell命令时，可节省启动shell的时间
        参数：
            :param device_no: 设备号
        返回：
            无
        """
        command = 'adb shell'
        process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=0, creationflags=subprocess.CREATE_NO_WINDOW)
        self.process = process

    def execute(self, command, output=False) -> list:
        """
        简介：
            执行adb shell命令
        参数：
            :param command: 命令
            :param output: 是否将输出结果打印在控制台
        返回：
            返回以行分隔、list类型的输出结果结果
        """
        # 防止在shell里再执行adb shell
        if command.startswith('adb shell'):
            command = command.replace('adb shell ', '', 1)

        process = self.process
        result = []

        process.stdin.write((command + '\n').encode('utf-8'))
        process.stdin.flush()

        # 发送标记命令，用于识别命令结束
        marker = "echo __COMMAND_FINISHED__"
        self.process.stdin.write((marker + '\n').encode('utf-8'))
        self.process.stdin.flush()

        while True:
            # 检查进程是否已结束
            if process.poll() is not None:
                break

            # 读取输出行
            line_bytes = process.stdout.readline()
            if not line_bytes:  # 无字节流，休眠
                time.sleep(0.05)
                continue
            # 容错解码：优先 UTF-8（安卓默认），失败则 GBK（Windows 兜底）
            try:
                line = line_bytes.decode('utf-8').strip()
            except UnicodeDecodeError:
                line = line_bytes.decode('gbk', errors='ignore').strip()

            if '__COMMAND_FINISHED__' in line:
                break

            if line:
                if output:
                    print(line)
                result.append(line)
        return result

    def exit(self):
        process = self.process
        process.stdin.close()
        process.terminate()
        process.wait()

# 等待设备连接
def wait_for_device():
    # 等待设备连接
    sys.stdout.write(f'\r等待设备连接...')
    sys.stdout.flush()
    # print(f'等待设备连接...')
    shell("adb wait-for-device")
    sys.stdout.write(f'\r')
    sys.stdout.flush()
    # 当前设备项目名和安卓版本号
    print(f"已连接设备: {get_device_name()}")

def shell(command, device=None, output: bool = False):
    """执行shell命令，返回 {'code': 状态码, 'output': 输出, 'error': 错误}"""
    if 'adb' == command[:3] and device is not None:
        command = 'adb -s ' + device + ' ' + command.split('adb ', 1)[1]
    sp = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
    result, error = sp.communicate(timeout=120)  # 基础超时控制（避免卡死）
    output_str = result.decode('utf-8', errors='ignore').strip()
    error_str = error.decode('utf-8', errors='ignore').strip()
    if output:
        print(output_str if sp.returncode == 0 else error_str)
    return {'code': sp.returncode, 'output': output_str, 'error': error_str}

# 获取设备名称
def get_device_name():
    return shell('adb shell getprop ro.product.model')['output'].strip()

# 设置息屏时长
def set_screen_off_timeout(duration=9999999):
    shell(f'adb shell settings put system screen_off_timeout {duration}')


# 获取所有可运行应用包名
def get_all_runnable_packages():
    """
    简介：
        获取所有可运行的应用包名
    返回：
        :return: 可运行的应用包名列表
    """
    runnable_packages = set()
    pattern = re.compile(r'packageName=(.+)')
    adb_shell = ADB_SHELL()

    output_list = adb_shell.execute('cmd package query-activities -a android.intent.action.MAIN -c android.intent.category.LAUNCHER | grep packageName=')
    for line in output_list:
        match = pattern.search(line)
        if match:
            runnable_packages.add(match.group(1))

    adb_shell.exit()
    return list(runnable_packages)

# 授予应用权限
def grant_authority(package):
    """
    简介：
        授予应用权限
    参数：
        :param package: 应用包名
    返回：
        :return: 本次授权的权限列表
    """
    adb_shell = ADB_SHELL()
    pattern = re.compile(r'(android\.permission\..+):')

    ungranted_permissions_output = adb_shell.execute(f'dumpsys package {package} | grep granted=false')

    ungranted_permissions_list = set()
    for permission in ungranted_permissions_output:
        match = pattern.search(permission)
        if match:
            ungranted_permissions_list.add(match.group(1))

    ungranted_permissions_list = list(ungranted_permissions_list)

    for permission in ungranted_permissions_list:
        adb_shell.execute(f'pm grant {package} {permission}')

    adb_shell.exit()

    return ungranted_permissions_list

# 授予应用权限
def grant_all_apps_permissions(thread_max_workers=5):
    """
    简介：
        授予所有应用权限
    参数：
        :param thread_max_workers: 线程池子线程数量（子线程过多可能导致电脑死机）
    返回：
        :return: 授权的应用包名列表
    """
    all_packages_list = get_all_runnable_packages()
    print(f"已获取所有可运行应用包名，数量为:{len(all_packages_list)}")
    with ThreadPoolExecutor(max_workers=thread_max_workers) as executor:
        for package in all_packages_list:
            executor.submit(grant_authority, package)
    return all_packages_list

# 写入黑名单
def write_to_file():
    file_path = f'{os.path.dirname(os.path.realpath(__file__))}\\blacklist.txt'
    with open(file_path, "w+", encoding="utf-8") as f:
        for package in black_list:
            f.write(f"{package}\n")
    return file_path

def run_all_apps_monkey(test_time=10000000):
    if shell(f"adb push {write_to_file()} /sdcard/")["code"] != 0:
        raise print("monkey test blacklist push fail!")
    else:
        monkey_command = f'adb shell "monkey ' \
                         f'--pkg-blacklist-file /sdcard/blacklist.txt ' \
                         f'--throttle 500 ' \
                         f'--randomize-throttle ' \
                         f'--ignore-crashes ' \
                         f'--ignore-timeouts ' \
                         f'--ignore-security-exceptions ' \
                         f'--ignore-native-crashes ' \
                         f'--pct-touch 40 ' \
                         f'--pct-motion 30 ' \
                         f'--pct-trackball 5 ' \
                         f'--pct-syskeys 5 ' \
                         f'--pct-appswitch 5 ' \
                         f'--pct-rotation 5 ' \
                         f'-v -v -v {test_time} ' \
                         f'1>/sdcard/monkeyInfo.txt 2>/sdcard/monkeyError.txt &"'
        shell(monkey_command)
        result = shell(monkey_command)
        if result['code'] == 0:
            print('Monkey命令已发送！')
        else:
            print('Monkey命令发送失败！')
            print(result['error'])

if __name__ == '__main__':
    # 等待设备连接
    wait_for_device()
    # 应用授权
    grant_all_apps_permissions()
    # 运行所有应用monkey
    run_all_apps_monkey()