airtest实现多进程批量执行
在airtest源文件android.py中添加返回当前所有连接设备的方法get_all_devices()

def get_all_devices(self):
import numpy as np
"""
返回所有连接设备列表
Get all devices

    Returns:
        local devices list

    """
    if not ADB().devices(state="device"):
        raise IndexError("ADB devices not found")
    all_devices = np.array(ADB().devices(state="device"))
    return all_devices[:, 0]  # ADB().devices(state="device")

多进程实现：

-- encoding=utf8 --

author = "apple"

from airtest.core.api import *
from airtest.core.android import Android
import multiprocessing

def main(device_param):
auto_setup(file, devices=[device_param])
start_app('package_name')
from poco.drivers.android.uiautomation import AndroidUiautomationPoco
poco = AndroidUiautomationPoco(use_airtest_input=True, screenshot_each_action=False)
start_app('package_name')
poco(text="酒店").click()
assert_exists(Template(r"tpl1646186837975.png", record_pos=(-0.239, -0.705), resolution=(1080, 2340)), "进入酒店首页正常")

poco(text="查找酒店").click()
assert_exists(Template(r"tpl1646186934064.png", record_pos=(-0.379, -0.801), resolution=(1080, 2340)), "酒店列表展示正常")

poco(':id/hotel_list_item_image').click()
assert_exists(Template(r"tpl1646186994987.png", record_pos=(-0.404, -0.396), resolution=(1080, 2340)), "酒店详情页展示正常")

sleep(2.0)
stop_app('package_name')

def run(device_param_list):
"""
:param device_param_list:
:return:
"""
process_count = len(android.get_all_devices())
pool = multiprocessing.Pool(processes=process_count)
for dev in device_param_list:
device = "Android://127.0.0.1:5037/%s" % dev
pool.apply_async(main, args=(device,))
pool.close()
pool.join()

if name == "main":
android = Android()
device_param_list = android.get_all_devices() # 返回当前所有连接设备，非airtest原有方法
run(device_param_list)

标签：list,param,poco,devices,airtest,device,pycharm
来源： https://www.cnblogs.com/menglx/p/15978935.html