app自动化测试---多台设备并行运行monkey（subprocss 子进程方式 && multiprocessing 多进程方式）

1.monkey运行命令

常用的adb命令

# monkey 执行命令
adb -s 127.0.0.1:62001 shell monkey -p org.cnodejs.android.md -v --throttle 500 20

-s 127.0.0.1:62001          # 执行的手机设备串号
-p org.cnodejs.android.md   # 执行运行的app包名
-v                          # 显示详细日志 日志级别分为三个基本  -vv  表示最详细的日志级别
--throttle 500              # 每个事件之间的 间隔时间 为500 毫秒
20                          # 执行20个事件

代码运行方式：

import subprocess

def test_run_monkey():
    monkey_cmd = f"adb -s 127.0.0.1:62001 shell monkey -p org.cnodejs.android.md -v --throttle 500 200"
    with open('./monkey_run.log',mode='w',encoding='utf8') as file:
        subprocess.Popen(monkey_cmd,stdout = file,shell=True)

# 调用方法
if __name__ == '__main__':
    test_run_monkey()

2.多台设备执行monkey命令（建议使用方案二）

要求：

  1.自动获取device的设备串号，比如设备连接 三台， 自动获取这三台手机的串号

  2.自动执行monkey ，运行monkey的时候，每台设备产生的日志信息应该保存到对应的设备的日志中。日志文件名以设备串号命名。

2.1 subprocss 子进程方式执行monkey命令

      缺点：执行完第一个设备之后才会去执行第二个设备，不能2个设备同时执行 

import subprocess

# 获取所有的连接设备
def get_connect_devices():
    devices = []
    proc = subprocess.Popen('adb devices',stdout=subprocess.PIPE,shell=True)
    for line in proc.stdout.readlines():
        # 将字节类型转换为字符串
        line_str = line.decode(encoding='utf8')
        if '	device' in line_str:
            # 字符串分割 提取 deviceid值
            device_id = line_str.strip().split('	device')[0]
            devices.append(device_id)
    print('devices----》',devices)    # ['127.0.0.1:62001', '127.0.0.1:62025']
    return devices


# 运行monkey命令
def run_monkey(device_id:str):
    print('device_id',device_id)
    # 字符串名字去掉特殊字符
    devicename = device_id.replace(':', '_')
    devicefile = devicename.replace('.', "_")

    monkey_cmd = f"adb -s {device_id} shell monkey -p org.cnodejs.android.md -vv --throttle 500 20"
    with open(f'./{devicefile}_monkey_run.log',mode='w',encoding='utf8') as file:
        subprocess.Popen(monkey_cmd,stdout = file,shell=True)

if __name__ == '__main__':
    devices = get_connect_devices()
    for device in devices:
        run_monkey(device)

2.2 multiprocessing 多进程的方式,并行运行monkey命令

     完美的满足了我们的需求

import subprocess
import multiprocessing

# 获取所有的连接设备
def get_connect_devices():
    devices = []
    proc = subprocess.Popen('adb devices',stdout=subprocess.PIPE,shell=True)
    for line in proc.stdout.readlines():
        # 将字节类型转换为字符串
        line_str = line.decode(encoding='utf8')
        if '	device' in line_str:
            # 字符串分割 提取 deviceid值
            device_id = line_str.strip().split('	device')[0]
            devices.append(device_id)
    print('devices----》',devices)    # ['127.0.0.1:62001', '127.0.0.1:62025']
    return devices


# 运行monkey命令
def run_monkey(device_id:str):
    print('device_id',device_id)
    # 字符串名字去掉特殊字符
    devicename = device_id.replace(':', '_')
    devicefile = devicename.replace('.', "_")

    monkey_cmd = f"adb -s {device_id} shell monkey -p org.cnodejs.android.md -vv --throttle 500 200"
    with open(f'./{devicefile}_monkey_run.log',mode='w',encoding='utf8') as file:
        subprocess.Popen(monkey_cmd,stdout = file,shell=True)

if __name__ == '__main__':
    devices = get_connect_devices()
    process = []
    for device_id in devices:
        p = multiprocessing.Process(target=run_monkey,args=(device_id,))
        process.append(p)
        p.start()

    for proc in process:
        proc.join()

运行代码之前，先手动的使用命令检查adb devices 检查连接设备，是否全部都连接上了，是否都安装了我们要测试的安装包，检查完毕，再运行代码

代码，运行之后，连接的多个设备会同时的执行monkey命令

 

3.multiprocessing 多进程解析

官方文档：https://docs.python.org/zh-cn/3/library/multiprocessing.html

import time

def run(name):
    time.sleep(2)
    print(f'helloworld--{name}')


if __name__ == '__main__':
    start= time.perf_counter()
    for i in range(4):
        run('张三')
    end = time.perf_counter()
    print(f'执行用时: {end-start}s')

使用多进程的方式执行：

import multiprocessing
import time

def run(name):
    time.sleep(2)
    print(f'helloworld--{name}')


if __name__ == '__main__':
    start= time.perf_counter()
    processes = []
    # 启动4个进程
    for i in range(4):
        # target=run     run是调用方法的方法名
        # args=(name,)   传参,name
        p = multiprocessing.Process(target=run,args=('张三',))
        processes.append(p)
        p.start()

    # 执行进程中的脚本
    for proce in processes:
        proce.join()   # 加入队列

    end = time.perf_counter()
    print(f'执行用时: {end-start}s')

使用多进程的方式，可以同时开启多个进程，大大缩短，代码执行所需要的时间