引言

用 airtest 快一年了，一直没有好好总结过，前几天看到有人发帖问多设备并行测试的事，发现现有的文档都只提供了单设备串行脚本的案例，在脚本较多的情况下会比较耗时，于是自己简单做了一个 demo，希望能给新手一些帮助！

参考文档：xinufo https://testerhome.com/topics/16567

需求

1.多设备并行批量的脚本（适合做兼容测试）
2.多设备分配运行批量的脚本（适合做功能的回归测试）
3.每个脚本单独生成一个 html 报告并在父文件夹生成一个聚合报告

exe 使用步骤

下载点这里：懒人链接
（1）.双击打开 airtest 启动器.exe，进入程序主界面

（2）.左边选取安卓设备，右边点击 ‘选取脚本路径’ 按钮，选择脚本所在的根目录，选好后可在右边窗口选取要运行的脚本；

（3）.选取模式，模式分两种，在【config.ini 全局配置】中会有介绍；
（4）.点击启动按钮，通过控制台查看运行情况，静静地等待运行结果；
（5）.运行结束后会有一个弹窗提示，点击 ok 按钮查看该次测试的报告；

（6）.历史报告会自动生成在 logs_root 文件夹下

源码使用步骤

github 地址

（1）.把自己写的 air 脚本放置到 scripts 文件夹下
（2）.打开 config.ini，根据注释填写运行模式、脚本名称及设备序列号；
（3）.运行 main.py 文件，推荐用 pycharm 运行，全局环境下也可以直接用 运行.bat 来运行；
（4）.等待运行结果，自动生成的报告将在 logs_root 文件夹；注：报告依赖 airtest 的静态，这里不建议更改报告的文件结构

config.ini 全局配置

[baseconf]
scripts_root = scripts
scripts = 
devices = all
mode = 1
platform = Android

scripts_root # 脚本根目录，默认为工程目录下的 scripts 文件夹

scripts # 要运行的脚本名称列表，半角逗号连接，如：SzwyMobile1014-1036.air,hh.air，无内容则按顺序运行 scripts 目录下所有脚本

devices = all # 设备 id，半角逗号连接，格式为：PBV0216727000183，8TFDU18926001948，为空默认选取电脑上连的第一台设备，all 则运行所有设备

mode = 1 # 1:每台设备各自运行所有脚本，2：所有设备分配运行所有脚本

platform = Android # 平台为 Windows 时设备号需填窗口句柄

这里提供两种模式：

• mode = 1：每台设备各自运行所有要跑的脚本，即批量并行相同脚本，报告数量=脚本数量 x 设备数量，适合做兼容测试；
• mode = 2：采用消息队列，将所有要跑的脚本逐一以分配的方式给空闲的设备运行，报告数量=脚本数量，适合做功能的回归测试；

流程分析

1.利用 multiprocessing 根据设备数量生成进程池，每台设备占用一条子进程；

2-1.模式 1 每条进程里做一样的操作，将所有要运行的脚本实例化成 MyAirtestCase，再添加进 unittest.TestSuite() 中，并一起给 unittest.TextTestRunner 运行

for fPy in oScripts:
  oCase = NewCase(fPy, sLogDir, sDeviceNum)
  oTestSuite.addTest(oCase)
   ...
unittest.TextTestRunner(verbosity=0).run(oTestSuite)  # 运行脚本

2-2.模式 2 创建 multiprocessing.Queue() 队列，将所有脚本 put 进队列中

oAirQueue = multiprocessing.Queue()
for sAirScript in lSuite:
    oAirQueue.put(sAirScript)

3.模式 1 无特殊处理，静静等待每一条进程 TestSuite() 里面添加的 case 全都运行完即可；
模式 2 在每条子进程里通过 Queue() 的 get() 方法获取一个脚本，给 unittest 运行

if oScripts.empty():
    return
fPy = oScripts.get()
oCase = NewCase(fPy, sLogDir, sDeviceNum, oScripts)
oTestSuite.addTest(oCase)
unittest.TextTestRunner(verbosity=0).run(oTestSuite)  # 运行脚本

4.模式 2 在一个脚本运行结束时，即 tearDown 函数里，判定队列是否存在，如果队列里还有脚本没跑，则继续选取脚本继续运行，直到脚本队列为空；
模式 1 无队列，不执行以上逻辑，跑完一个脚本直接生成对应的报告，并执行下一个 case

def tearDown(self):
    try:
        output = os.path.join(self.logdir, "recording_0.mp4")
        print(output)
        self.m_oDev.stop_recording(output)
    except:
        traceback.print_exc()
    self.Report()
    if self.queue:
        RunScript(self.sDevice, self.m_LogRoot, oScripts=self.queue)

5.所有进程结束，代表单次测试结束，生成聚合报告

def Finish(sLogDir):
    print('test finish')
    sCombineLog = os.path.join(sLogDir, 'log.html')
    sCombineTxt = os.path.join(sLogDir, 'log.txt')
    with open(sCombineTxt, 'r') as f:
        lMsg = f.readlines()
    template_vars = {
        'patch_tag': os.path.basename(sLogDir),
        'files': [json.loads(line) for line in lMsg]
    }
    report.render('combine_log.html', sCombineLog, **template_vars)

其他模块作用

runner.py 利用 multiprocessing 根据设备数量生成进程池，单个进程里再利用 unittest 生成每一个脚本的测试用例

report.py 根据模板生成单个 airtest 脚本测试的报告，重写了 airtest 源码中若干源码，减少报告中的静态资源的路径依赖

utils.py 该模块提供了一些通用接口，其中还包括压缩本地报告上传至云平台的代码，上传地址需使用者自己填写

video.py 该模块利用 OpenCV 实现 Windows 端的录屏功能，弥补了 airtest 在 PC 运行时无法录屏的缺点。其中视频的帧率和录制时间间隔可以自己调整至一个合适的数值

file_lock.py 为了记录单次测试里每一个报告的聚合结果，这里采用将结果写入临时文件的方式。由于存在多条进程同时对一个文件进行读写操作的情况，我只是简单得用了文件锁来处理了一下。经测试在 windows 端进程较多的情况下仍会出现结果写入异常的情况，条件足够的话建议将结果保存在自己的数据库中。

最后

模块的具体作用还是直接看 github 和源码吧，废话不多说，show my code
github 地址