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