本文首发于:行者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框架做一些高级的定制化扩大性能。