Airtest入门及多设备管理总结

时间 2021-05-09

标签 python android shell json windows api 多线程 app 框架编辑器栏目 Python 繁體版

原文原文链接

本文首发于：行者AI

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

提及Airtest就不得不提AirtestIDE，一个强大的GUI工具，它整合了Airtest和Poco两大框架，内置adb工具、Poco-inspector、设备录屏、脚本编辑器、ui截图等，也正是因为它集成许多了强大的工具，使得自动化测试变得更为方便，极大的提高了自动化测试效率，而且获得普遍的使用。android

1. 简单入门

1.1 准备

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

1.2 启动AirtestIDE

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

一个是AirtestIDE的UI界面，以下：json

1.3 链接设备

链接的时候要确保设备在线，一般须要点击刷新ADB来查看更新设备及设备状态，而后双击须要链接的设备便可链接，若是链接的设备是模拟器，需注意以下：windows

确保模拟器与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名称将其写到脚本编辑窗中④。api

1.5 脚本编辑

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

1.6 运行

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

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