Airtest 入门及多设备管理总结

关注

发布于: 2021 年 04 月 08 日

本文首发于：行者AI

Airtest 是一款基于图像识别和 poco 控件识别的 UI 自动化测试工具，用于游戏和 App 测试，也广泛应用于设备群控，其特性和功能不亚于 appium 和 atx 等自动化框架。

说起 Airtest 就不得不提 AirtestIDE，一个强大的 GUI 工具，它整合了 Airtest 和 Poco 两大框架，内置 adb 工具、Poco-inspector、设备录屏、脚本编辑器、ui 截图等，也正是由于它集成许多了强大的工具，使得自动化测试变得更为方便，极大的提升了自动化测试效率，并且得到广泛的使用。

1. 简单入门

1.1 准备

从官网下载并安装 AirtestIDE。
准备一台移动设备，确保 USB 调试功能处于开启状态，也可使用模拟器代替。

1.2 启动 AirtestIDE

打开 AirtestIDE，会启动两个程序，一个是打印操作日志的控制台程序，如下：

一个是 AirtestIDE 的 UI 界面，如下：

1.3 连接设备

连接的时候要确保设备在线，通常需要点击刷新 ADB 来查看更新设备及设备状态，然后双击需要连接的设备即可连接，如果连接的设备是模拟器，需注意如下：

确保模拟器与 Airtest 中的 adb 版本一致，否则无法连接，命令行中使用 adb version 即可查看 adb 版本，Airtest 中的 adb 在 Install_path\airtest\core\android\static\adb\windows 目录下面。
确保勾选 Javacap 方式②连接，避免连接后出现黑屏。

1.4 UI 定位

在 Poco 辅助窗选择 Android①并且使能 Poco inspector②，然后将鼠标放到控件上面即可显示控件的 UI 名称③，也可在左侧双击 UI 名称将其写到脚本编辑窗中④。

1.5 脚本编辑

在脚本编辑窗编写操作脚本⑤，比如使用百度搜索去搜索 Airtest 关键词，输入关键字后点击百度一下控件即可完成搜索。

1.6 运行

运行脚本，并在 Log 查看窗查看运行日志⑥。以上操作只是简单入门，更多操作可参考官方文档。

2. 多线程中使用 Airtest

当项目中需要群控设备时，就会使用多进程或者多线程的方式来调度 Airtest，并将 Airtest 和 Poco 框架集成到项目中，以纯 Python 代码的方式来使用 Airtest，不过仍需 Airtest IDE 作为辅助工具帮助完成 UI 控件的定位，下面给大家分享一下使用 Airtest 控制多台设备的方法以及存在的问题。

2.1 安装

纯 python 环境中使用 Airtest，需在项目环境中安装 Airtest 和 Poco 两个模块，如下：pip install -U airtest pocoui

2.2 多设备连接

每台设备都需要单独绑定一个 Poco 对象，Poco 对象就是一个以 apk 的形式安装在设备内部的一个名为 com.netease.open.pocoservice 的服务（以下统称 pocoservice），这个服务可用于打印设备 UI 树以及模拟点击等，多设备连接的示例代码如下：

from airtest.core.api import *from poco.drivers.android.uiautomation import AndroidUiautomationPoco    
# 过滤日志air_logger = logging.getLogger("airtest")air_logger.setLevel(logging.ERROR)auto_setup(__file__)
dev1 = connect_device("Android:///127.0.0.1:21503")dev2 = connect_device("Android:///127.0.0.1:21503")dev3 = connect_device("Android:///127.0.0.1:21503")
poco1 = AndroidUiautomationPoco(device=dev1)poco2 = AndroidUiautomationPoco(device=dev2)poco3 = AndroidUiautomationPoco(device=dev3)

复制代码

2.3 Poco 管理

上面这个写法确实保证了每台设备都单独绑定了一个 Poco 对象，但是上面这种形式不利于 Poco 对象的管理，比如检测每个 Poco 的存活状态。因此需要一个容器去管理并创建 Poco 对象，这里套用源码里面一种方法作为参考，它使用单例模式去管理 Poco 的创建并将其存为字典，这样既保证了每台设备都有一个单独的 Poco，也方便通过设备串号去获取 Poco 对象，源码如下：

    class AndroidUiautomationHelper(object):        _nuis = {}            @classmethod        def get_instance(cls, device):            """            This is only a slot to store and get already initialized poco instance rather than initializing again. You can            simply pass the ``current device instance`` provided by ``airtest`` to get the AndroidUiautomationPoco instance.            If no such AndroidUiautomationPoco instance, a new instance will be created and stored.                 Args:                device (:py:obj:`airtest.core.device.Device`): more details refer to ``airtest doc``                Returns:                poco instance            """                if cls._nuis.get(device) is None:                cls._nuis[device] = AndroidUiautomationPoco(device)            return cls._nuis[device]

复制代码

AndroidUiautomationPoco 在初始化的时候，内部维护了一个线程 KeepRunningInstrumentationThread 监控 pocoservice，监控 pocoservice 的状态防止异常退出。

    class KeepRunningInstrumentationThread(threading.Thread):        """Keep pocoservice running"""            def __init__(self, poco, port_to_ping):            super(KeepRunningInstrumentationThread, self).__init__()            self._stop_event = threading.Event()            self.poco = poco            self.port_to_ping = port_to_ping            self.daemon = True            def stop(self):            self._stop_event.set()            def stopped(self):            return self._stop_event.is_set()            def run(self):            while not self.stopped():                if getattr(self.poco, "_instrument_proc", None) is not None:                    stdout, stderr = self.poco._instrument_proc.communicate()                    print('[pocoservice.apk] stdout: {}'.format(stdout))                    print('[pocoservice.apk] stderr: {}'.format(stderr))                if not self.stopped():                    self.poco._start_instrument(self.port_to_ping)  # 尝试重启                    time.sleep(1)

复制代码

这里存在的问题是，一旦 pocoservice 出了问题（不稳定），由于 KeepRunningInstrumentationThread 的存在，pocoservice 就会重启，但是由于 pocoservice 服务崩溃后，有时是无法重启的，就会循环抛出 raise RuntimeError("unable to launch AndroidUiautomationPoco")的异常，导致此设备无法正常运行，一般情况下，我们需要单独处理它，具体如下：

处理 Airtest 抛出的异常并确保 pocoservice 服务重启，一般情况下，需要重新安装 pocoservice，即重新初始化。但是如何才能检测 Poco 异常，并且捕获此异常呢？这里在介绍一种方式，在管理 Poco 时，使用定时任务的方法去检测 Poco 的状况，然后将异常 Poco 移除，等待其下次连接。

2.4 设备异常处理

一般情况下，设备异常主要表现为 AdbError、DeviceConnectionError，引起这类异常的原因多种多样，因为 Airtest 控制设备的核心就是通过 adb shell 命令去操作，只要执行 adb shell 命令，都有可能出现这类错误，你可以这样想，Airtest 中任何动作都是在执行 adb shell 命令，为确保项目能长期稳定运行，就要特别注意处理此类异常。

第一个问题

Airtest 的 adb shell 命令函数通过封装 subprocess.Popen 来实现，并且使用 communicate 接收 stdout 和 stderr，这种方式启动一个非阻塞的子进程是没有问题的，但是当使用 shell 命令去启动一个阻塞式的子进程时就会卡住，一直等待子进程结束或者主进程退出才能退出，而有时候我们不希望被子进程卡住，所以需单独封装一个不阻塞的 adb shell 函数，保证程序不会被卡住，这种情况下为确保进程启动成功，需自定义函数去检测该进程存在，如下：

    def rshell_nowait(self, command, proc_name):        """        调用远程设备的shell命令并立刻返回, 并杀死当前进程。        :param command: shell命令        :param proc_name: 命令启动的进程名, 用于停止进程        :return: 成功:启动进程的pid， 失败:None        """        if hasattr(self, "device"):            base_cmd_str = f"{self.device.adb.adb_path} -s {self.device.serialno} shell "            cmd_str = base_cmd_str + command            for _ in range(3):                proc = subprocess.Popen(cmd_str)                proc.kill()  # 此进程立即关闭，不会影响远程设备开启的子进程                pid = self.get_rpid(proc_name)                if pid:              return pid        def get_rpid(self, proc_name):        """        使用ps查询远程设备上proc_name对应的pid        :param proc_name: 进程名        :return: 成功:进程pid， 失败:None        """        if hasattr(self, "device"):            cmd = f'{self.device.adb.adb_path} -s {self.device.serialno} shell ps | findstr {proc_name}'            res = list(filter(lambda x: len(x) > 0, os.popen(cmd).read().split(' ')))            return res[1] if res else None

复制代码

注意：通过 subprocess.Popen 打开的进程记得使用完成后及时关闭，防止出现Too many open files的错误。

第二个问题

Airtest 中初始化 ADB 也是会经常报错，这直接导致设备连接失败，但是 Airtest 并没有直接捕获此类错误，所以我们需要在上层处理该错误并增加重试机制，如下面这样，也封装成装饰器或者使用 retrying.retry。

  def check_device(serialno, retries=3):      for _ in range(retries)          try:              adb = ADB(serialno)              adb.wait_for_device(timeout=timeout)              devices = [item[0] for item in adb.devices(state='device')]              return serialno in devices       except Exception as err:              pass

复制代码

一般情况下使用 try except 来捕可能的异常，这里推荐使用 funcy，funcy 是一款堪称瑞士军刀的 Python 库，其中有一个函数 silent 就是用来装饰可能引起异常的函数，silent 源码如下，它实现了一个名为 ignore 的装饰器来处理异常。当然 funcy 也封装许多 python 日常工作中常用的工具，感兴趣的话可以看看 funcy 的源码。

  def silent(func):        """忽略错误的调用"""        return ignore(Exception)(func)        def ignore(errors, default=None):        errors = _ensure_exceptable(errors)            def decorator(func):            @wraps(func)            def wrapper(*args, **kwargs):                try:                   return func(*args, **kwargs)                except errors as e:                  return default            return wrapper        return decorator                      def _ensure_exceptable(errors):        is_exception = isinstance(errors, type) and issubclass(errors, BaseException)        return errors if is_exception else tuple(errors)            #参考使用方法    import json        str1 = '{a: 1, 'b':2}'    json_str = silent(json.loads)(str1)

复制代码

第三个问题

Airtest 执行命令时会调用 G.DEVICE 获取当前设备（使用 Poco 对象底层会使用 G.DEVICE 而非自身初始化时传入的 device 对象），所以在多线程情况下，本该由这台设备执行的命令可能被切换另外一台设备执行从而导致一系列错误。解决办法就是维护一个队列，保证是主线程在执行 Airtest 的操作，并在使用 Airtest 的地方设置 G.DEVICE 确保 G.DEVICE 等于 Poco 的 device。

3.结语

Airtest 在稳定性、多设备控制尤其是多线程中存在很多坑。最好多看源码加深对 Airtest 的理解，然后再基于 Airtest 框架做一些高级的定制化扩展功能。

发布于: 2021 年 04 月 08 日阅读数: 34

原文链接:【http://xie.infoq.cn/article/79f2363c11abf85b4024e03c8】。

行者AI

关注

行者AI，为游戏插上人工智能的翅膀。 2020.12.18 加入

行者AI（成都潜在人工智能科技有限公司）专注于人工智能在游戏领域的研究和应用，凭借自研算法，推出游戏AI、智能内容审核、数据平台等产品服务。

发布

暂无评论

创作场景

Airtest 入门及多设备管理总结

1. 简单入门

1.1 准备

1.2 启动 AirtestIDE

1.3 连接设备

1.4 UI 定位

1.5 脚本编辑

1.6 运行

2. 多线程中使用 Airtest

2.1 安装

2.2 多设备连接

2.3 Poco 管理

2.4 设备异常处理

3.结语

行者AI

评论