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框架做一些高级的定制化扩展功能。

我们是行者AI，我们在“AI＋游戏”中不断前行。

快来【公众号 | xingzhe_ai】，和我们讨论更多技术问题吧！