Airtest入門及多設備管理總結

本文首發於：行者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框架做一些高級的定制化擴展功能。

---

PS：更多技術干貨，快關注【公眾號 | xingzhe_ai】，與行者一起討論吧！