Airtest自定义启动器支持批量运行脚本，并兼容在AirtestIDE中使用

小编注：上期详细讲了Airtest启动器的原理，以及在最后给出了2个实现方案。本次是第2个方案的另一个实现案例，供大家学习参考。

Python v3.7.0 / Airtest: 1.1.1 / PocoUI: 1.0.78

自定义的启动器主要实现了以下功能：

将一些公共参数和方法添加到全局变量中，在各业务脚本中无需声明，可直接使用，如语句超时时间 TIMEOUT，就在此进行统一设置；

设置Airtest全局属性值，对所有脚本生效，如下所示：

# 图像识别精确度阈值 [0,1]
 ST.THRESHOLD = 0.80
# assert语句里图像识别时使用的高要求阈值 [0,1]
 ST.THRESHOLD_STRICT = 0.85
# 每一步操作后等待多长时间再进行下一步操作
 ST.OPDELAY = 1
# 图像查找超时时间，默认为20s
 ST.FIND_TIMEOUT = 10
# 修改图像识别算法顺序，只要成功匹配任意一个符合设定阙值的结果，程序就会认为识别成功
 ST.CVSTRATEGY = ["tpl", "sift", "brisk"]
1
2
3
4
5
6
7
8
9
10

在正式脚本运行前后，添加子脚本的运行，使得运行各个业务脚本时，初始页面都将为小程序首页，不需要再添加额外的环境判断代码；

setup.air:在每个脚本运行前，都会首先运行此脚本。此脚本会判断当前页面是否位于小程序首页，若不是，则会重启微信，进入小程序首页；

teardown.air:在每个脚本运行结束后，都会运行此脚本。此脚本会点击页面顶部的HOME图标，返回到小程序首页。若成功返回到小程序首页，则会尝试关闭首页遮罩广告（注意：遮罩广告样式不能改变），若未能成功返回首页，则会重启微信，再进入小程序首页。

通过配置config.csv中的内容，实现批量运行脚本；

在运行脚本时，会先迭代查找suite/目录下所有以.air结尾的目录，自动忽略setup.air和teardown.air，然后读取config.csv中所配置的 Label 为 "Y"的脚本名称，二者做交集运算，其结果作为本次实际要运行的脚本集合。

批量运行脚本结束后，支持生成聚合报告，与Jenkins进行持续集成时，此报告会作为附件进行发送。点击聚合报告中模块名称，会跳转到Airtest Project自带的测试报告页面。

Airtest Project自带的测试报告中，引用了很多静态资源，如js/css文件、页面截图等，这里通过使用Nginx来搭建静态资源服务器，使测试报告可以正常加载。

兼容在AirtestIDE中使用自定义启动器

持续集成时，必然要通过命令行来运行业务脚本，需要手动传入一些参数如设备号，脚本名等，而使用AirtesIDE运行脚本时，这些参数会自动传入。源码中使用argparse的方法来解析命令行传入的参数，为了兼容在AirtestIDE中使用自定义的启动器，此处不能使用和源码相同的方法来解析命令行参数，否则AirtestIDE中会无法识别参数，而导致报错;

import sys

if len(sys.argv) == 1:  # 如果没有传入任何命令行参数，则批量执行脚本
    main()
    quit()
else:  # 在AirtestIDE中单个运行脚本时，会传入script、log、device等参数，则调用以下原始方法；
    ap = runner_parser()
    args = ap.parse_args()
    run_script(args, McCustomAirtestCase)
1
2
3
4
5
6
7
8
9

直接贴项目原始代码，注释已经写得很详细了。

#!/usr/bin/env python
# -*- coding:utf8 -*-
"""
@File  : mc_launcher.py
@Author: Rethink
@Date  : 2019/12/20 10:49
@Desc  :
"""
import csv
import datetime
import os
import shutil
import sys
import time
from argparse import *
from pathlib import Path
from typing import List

import airtest.report.report as report
import jinja2
from airtest.cli.parser import runner_parser
from airtest.cli.runner import run_script
from airtest.core.settings import Settings as ST

# 若脚本在IDE中运行，IDE会自动帮忙加载AirtestCase；若用命令行运行脚本，则需要导入 AirtestCase
if not globals().get("AirtestCase"):
    from airtest.cli.runner import AirtestCase

AIRTEST_EXPORT_DIR = os.getenv('AIRTEST_EXPORT_DIR')  # 测试报告相关资源打包后的导出路径，目录不存在会自动创建
AUTOLINE_HOST = os.getenv('AUTOLINE_HOST')  # 静态资源文件服务器 格式：Scheme://IP:Port

if Path(AIRTEST_EXPORT_DIR).is_dir():
    pass
else:
    os.makedirs(AIRTEST_EXPORT_DIR)

class McCustomAirtestCase(AirtestCase):
    """
    Aietest Project自定义启动器，参考文档：http://airtest.netease.com/docs/cn/7_settings/3_script_record_features.html
    """

    PROJECT_ROOT = os.getenv("AIRTEST_PROJECT_ROOT", r"E:\treasure\Airtest\suite")  # 设置子脚本存放的根目录

    def setUp(self):
        print("----------Custom Setup [Hook method]----------")
        # 将自定义的公共变量加入到`self.scope`中，在脚本代码中就可以直接使用
        self.scope["SLEEPTIME"] = 1  # 睡眠时间
        self.scope["TIMEOUT"] = 5  # 超时时间

        # 设置`Airtest`全局属性值
        ST.THRESHOLD = 0.80  # 图像识别精确度阈值 [0,1]
        ST.THRESHOLD_STRICT = 0.85  # assert语句里图像识别时使用的高要求阈值 [0,1]
        ST.OPDELAY = 2  # 每一步操作后等待多长时间进行下一步操作, 只针对Airtest语句有效, 默认0.1s
        ST.FIND_TIMEOUT = 10  # 图像查找超时时间，默认为20s
        ST.CVSTRATEGY = ["tpl", "sift", "brisk"]  # 修改图像识别算法顺序，只要成功匹配任意一个符合设定阙值的结果，程序就会认为识别成功

        # 可以将一些通用的操作进行封装，然后在其他脚本中 import;
        # Airtest 提供了 using 接口，能够将需要引用的脚本加入 sys.path 里，其中包含的图片文件也会被加入 Template 的搜索路径中
        # using("common.air")    # 相对于PROJECT_ROOT的路径
        self.exec_other_script("setup.air")
        super(McCustomAirtestCase, self).setUp()

    def tearDown(self):
        print("----------Custom Teardown [Hook method]----------")
        self.exec_other_script("teardown.air")
        super(McCustomAirtestCase, self).tearDown()


def find_all_scripts(suite_dir: str = "") -> list:
    """
    遍历suite目录，取出所有的测试脚本
    """
    suite = []

    if not suite_dir:
        suite_dir = McCustomAirtestCase.PROJECT_ROOT

    for fpath in Path(suite_dir).iterdir():
        tmp = Path(suite_dir, fpath)
        if not tmp.is_dir():
            pass
        else:
            if fpath.suffix == '.air' and fpath.stem not in ["setup", "teardown"]:  # 这里会排除掉初始化脚本
                suite.append(fpath.name)
            else:
                deep_scripts = find_all_scripts(tmp)  # 递归遍历
                suite += deep_scripts

    return suite


def allow_run_scripts() -> List[str]:
    """
    读取配置文件，返回允许运行的脚本名称
    """
    config_allow_run = []

    with open("config.csv", "r") as f:
        f_csv = csv.DictReader(f)
        for row in f_csv:
            if row["Label"].upper() == "Y":
                config_allow_run.append(row["Script"].strip())

    return config_allow_run


def run_airtest(script, log_root, device=""):
    """
    运行单个脚本，并生成测试报告，返回运行结果
    :param script:  *.air, 要运行的脚本
    :param device:  设备字符串
    :param log_root:  脚本日志存放目录
    """
    if log_root.is_dir():
        shutil.rmtree(log_root)
    else:
        os.makedirs(log_root)
        print(str(log_root) + '>>> is created')

    # 组装运行参数
    args = Namespace(device=device,  # 设备字符串
                     log=log_root,  # log目录
                     recording=None,  # 禁止录屏
                     script=script  # *.air
                     )
    run_script(args, McCustomAirtestCase)


def generate_report(script, *, log_root, export_root):
    """
    生成测试报告
    :param script:  运行名称
    :param log_root:  脚本log目录
    :return: export_root  测试报告输出目录
    """
    # 测试报告导出目录
    if not export_root.is_dir():
        os.makedirs(export_root)
        print(str(export_root) + '>>> is created')

    output_file = Path(export_root, script.replace('.air', '.log'), 'log.html')  # 测试报告`log.html`存放路径

    # 生成测试报告
    rpt = report.LogToHtml(script_root=script,  # *.air
                           log_root=log_root,  # log目录
                           export_dir=export_root,  # 设置此参数后，生成的报告内资源引用均使用相对路径
                           lang='zh',  # 设置语言, 默认"en"
                           script_name=script.replace(".air", ".py"),  # *.air/*.py
                           static_root=AUTOLINE_HOST + '/static',  # 设置此参数后，打包后的资源目录中不会包含样式资源
                           plugins=['poco.utils.airtest.report']  # 使报告支持poco语句
                           )
    rpt.report(template_name="log_template.html", output_file=output_file)

    # 提取脚本运行结果
    result = rpt.test_result  # True or False

    return result


def summary_html(results: list, output_dir: str, template_dir: str, elapsed_time):
    """
    生成自定义的聚合报告
    :param results:  用例执行结果
    :param output_dir:  html输出目录
    :param template_parent:  jinja2模板所在目录
    """
    env = jinja2.Environment(
        loader=jinja2.FileSystemLoader(template_dir),
        extensions=(),
        autoescape=True,
    )
    template = env.get_template("summary_template.html", template_dir)
    html = template.render({"results": results,  # 运行结果
                            "now": datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),  # 当前时间
                            "elapsed_time": elapsed_time,  # 脚本运行时长
                            })
    output_file = Path(output_dir, "summary.html")  # 聚合报告路径
    with open(output_file, 'w', encoding="utf-8") as f:
        f.write(html)


def copy_lastest_report(source):
    """
    复制最新生成的测试报告到 lastest/
    """
    latest = Path(AIRTEST_EXPORT_DIR, 'latest')
    if latest.is_dir():
        shutil.rmtree(latest)  # dst 目录必须不能存在，否则copytree报错

    shutil.copytree(os.path.abspath(source),
                    os.path.abspath(latest))  # 目录内文件不能正在使用 否则无法复制成功PermissionError: [WinError 5] 拒绝访问


def print_message(*message):
    print("*" * 50)
    mes = " ".join(map(str, message))
    print(mes)
    print("*" * 50)


def clear_history_report(report_dir, critical_day):
    """
    清除生成的历史测试报告
    :param report_dir:  测试报告目录
    :param critical_day:
    """
    root = Path(report_dir)
    reports = [report for report in root.iterdir() if report.is_dir() and report.name != 'latest']
    for report in reports:
        dt = datetime.datetime.strptime(report.name, '%Y%m%d%H%M%S')
        delta = datetime.datetime.now() - dt
        if delta.days > int(critical_day):
            shutil.rmtree(os.path.abspath(report))


def main():
    start_time = int(time.time())
    all_scripts = find_all_scripts()
    config_allow_run = allow_run_scripts()
    suite: set = set(all_scripts).intersection(config_allow_run)
    print_message("本次要运行的用例集合为：", suite)

    results = []  # 脚本运行结果汇总
    dt = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
    export_root = Path(AIRTEST_EXPORT_DIR, dt)  # 测试报告导出目录

    for script in suite:
        print_message("正在运行的用例名称：" + script)

        log_root = Path(McCustomAirtestCase.PROJECT_ROOT, 'log', script.replace('.air', '.log'))  # 脚本日志目录
        run_airtest(script, log_root)  # 运行脚本
        result = generate_report(script, log_root=log_root, export_root=export_root)  # 生成测试报告
        results.append((script, result))

        if not result:
            print_message("用例执行失败: " + script)
        else:
            print_message("用例执行成功:" + script)

    print_message("用例运行结果汇总:", results)
    end_time = int(time.time())
    elapsed_time = end_time - start_time
    summary_html(results, output_dir=export_root, template_dir=McCustomAirtestCase.PROJECT_ROOT,
                 elapsed_time=elapsed_time)
    copy_lastest_report(source=export_root)
    clear_history_report(AIRTEST_EXPORT_DIR, 3)


if __name__ == '__main__':
    """
    通过命令行来启动air脚本时，需要传入一些参数如设备号，脚本名等，而使用AirtesIDE运行脚本时，这些参数会自动传入；
    源码中使用argparse的方法来解析命令行参数，此处不能使用和源码相同的方法，否则AirtestIDE中会无法识别参数，而导致报错;
    具体源码，详见runner_parser、parse_args和run_script三个方法
    """
    # parser = argparse.ArgumentParser()
    # parser.add_argument("-b", "--Batch", type=str, help="是否启用批量运行脚本功能，为了支持在AirtestIDE中使用自定义的启动器，此项默认不启用")
    # args = parser.parse_args()
    #
    # # 如果传入Batch参数且其值为true, 则表示开启批量运行脚本功能
    # if args.Batch and args.Batch.lower() == "true":
    #     main()
    # else:
    #     # AirtestIDE中运行脚本时, 会自动带入script, --device, --log, --recording参数；
    #     ap = runner_parser()
    #     args = ap.parse_args()
    #     run_script(args, McCustomAirtestCase)

    if len(sys.argv) == 1:  # 如果没有传入任何命令行参数，则批量执行脚本
        main()
        quit()
    else:  # 在AirtestIDE中运行脚本时，会传入script、log、device等参数，则调用以下方法；
        ap = runner_parser()
        args = ap.parse_args()
        run_script(args, McCustomAirtestCase)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274

如果你不想一个人野蛮生长，找不到系统的资料，问题得不到帮助，坚持几天便放弃的感受的话，可以加入我们的QQ群：746506216，大家可以一起讨论交流，里面会有各种软件测试资料和技术交流。

---

资源分享

下方这份完整的软件测试视频学习教程已经上传CSDN官方认证的二维码，朋友们如果需要可以自行免费领取 【保证100%免费】