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。