背景

??穩(wěn)定性測(cè)試時(shí)，需要跑整機(jī)monkey，實(shí)際測(cè)試中發(fā)現(xiàn)：monkey打開應(yīng)用時(shí)，無法點(diǎn)擊“權(quán)限彈窗”，導(dǎo)致monkey無法進(jìn)入應(yīng)用首頁。手動(dòng)打開每個(gè)應(yīng)用授權(quán)，非常耗時(shí)，基于此背景開發(fā)該工具。

一鍵停止Monkey批處理

@ECHO off  
@REM 無限循環(huán)的標(biāo)簽  

ECHO wait-for-device...
adb wait-for-device
ECHO Device connected.
echo.
:LOOP
@REM 拿到Monkey進(jìn)程id
adb shell "kill -9 `ps -ef | grep com.android.commands.monkey | grep -v 'grep' | awk '{print $2}'`"
@REM 清除時(shí)鐘的鬧鐘
adb shell pm clear com.android.deskclock
ECHO Monkey hat stopped.
echo.
PAUSE
GOTO LOOP  
@ECHO on

整機(jī)Monkey測(cè)試源碼

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

# 黑名單，monkey測(cè)試不跑的包
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非框架行動(dòng)定位服務(wù)
    "com.mediatek.gnss.nonframeworklbs"
]

# 創(chuàng)建adb shell會(huì)話, 大大提升adb shell性能
class ADB_SHELL:
    def __init__(self):
        """
        簡(jiǎn)介：
            使用subprocess開啟一個(gè)adb shell子線程并保持會(huì)話狀態(tài)
            在頻繁使用adb shell命令時(shí)，可節(jié)省啟動(dòng)shell的時(shí)間
        參數(shù)：
            :param device_no: 設(shè)備號(hào)
        返回：
            無
        """
        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:
        """
        簡(jiǎn)介：
            執(zhí)行adb shell命令
        參數(shù)：
            :param command: 命令
            :param output: 是否將輸出結(jié)果打印在控制臺(tái)
        返回：
            返回以行分隔、list類型的輸出結(jié)果結(jié)果
        """
        # 防止在shell里再執(zhí)行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()

        # 發(fā)送標(biāo)記命令，用于識(shí)別命令結(jié)束
        marker = "echo __COMMAND_FINISHED__"
        self.process.stdin.write((marker + '\n').encode('utf-8'))
        self.process.stdin.flush()

        while True:
            # 檢查進(jìn)程是否已結(jié)束
            if process.poll() is not None:
                break

            # 讀取輸出行
            line_bytes = process.stdout.readline()
            if not line_bytes:  # 無字節(jié)流，休眠
                time.sleep(0.05)
                continue
            # 容錯(cuò)解碼：優(yōu)先 UTF-8（安卓默認(rèn)），失敗則 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()

# 等待設(shè)備連接
def wait_for_device():
    # 等待設(shè)備連接
    sys.stdout.write(f'\r等待設(shè)備連接...')
    sys.stdout.flush()
    # print(f'等待設(shè)備連接...')
    shell("adb wait-for-device")
    sys.stdout.write(f'\r')
    sys.stdout.flush()
    # 當(dāng)前設(shè)備項(xiàng)目名和安卓版本號(hào)
    print(f"已連接設(shè)備: {get_device_name()}")

def shell(command, device=None, output: bool = False):
    """執(zhí)行shell命令，返回 {'code': 狀態(tài)碼, 'output': 輸出, 'error': 錯(cuò)誤}"""
    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)  # 基礎(chǔ)超時(shí)控制（避免卡死）
    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}

# 獲取設(shè)備名稱
def get_device_name():
    return shell('adb shell getprop ro.product.model')['output'].strip()

# 設(shè)置息屏?xí)r長(zhǎng)
def set_screen_off_timeout(duration=9999999):
    shell(f'adb shell settings put system screen_off_timeout {duration}')


# 獲取所有可運(yùn)行應(yīng)用包名
def get_all_runnable_packages():
    """
    簡(jiǎn)介：
        獲取所有可運(yùn)行的應(yīng)用包名
    返回：
        :return: 可運(yùn)行的應(yīng)用包名列表
    """
    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)

# 授予應(yīng)用權(quán)限
def grant_authority(package):
    """
    簡(jiǎn)介：
        授予應(yīng)用權(quán)限
    參數(shù)：
        :param package: 應(yīng)用包名
    返回：
        :return: 本次授權(quán)的權(quán)限列表
    """
    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

# 授予應(yīng)用權(quán)限
def grant_all_apps_permissions(thread_max_workers=5):
    """
    簡(jiǎn)介：
        授予所有應(yīng)用權(quán)限
    參數(shù)：
        :param thread_max_workers: 線程池子線程數(shù)量（子線程過多可能導(dǎo)致電腦死機(jī)）
    返回：
        :return: 授權(quán)的應(yīng)用包名列表
    """
    all_packages_list = get_all_runnable_packages()
    print(f"已獲取所有可運(yùn)行應(yīng)用包名，數(shù)量為:{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命令已發(fā)送！')
        else:
            print('Monkey命令發(fā)送失敗！')
            print(result['error'])

if __name__ == '__main__':
    # 等待設(shè)備連接
    wait_for_device()
    # 應(yīng)用授權(quán)
    grant_all_apps_permissions()
    # 運(yùn)行所有應(yīng)用monkey
    run_all_apps_monkey()