Сводка по входу в Airtest и управлению несколькими устройствами

искусственный интеллект контрольная работа

Эта статья была впервые опубликована на:Уокер ИИ

Airtest — это инструмент для автоматизации тестирования пользовательского интерфейса, основанный на распознавании изображений и распознавании элементов управления poco.Он используется для тестирования игр и приложений, а также широко используется для группового контроля устройств.Его возможности и функции не уступают фреймворкам автоматизации, таким как appium и atx. .

Говоря об Airtest, мы должны упомянуть AirtestIDE, мощный инструмент с графическим интерфейсом, который объединяет две среды Airtest и Poco, встроенный инструмент adb, Poco-инспектор, запись экрана устройства, редактор сценариев, снимок экрана пользовательского интерфейса и т. д. Именно благодаря этому многие мощные инструменты интегрированы, что делает автоматизированное тестирование более удобным, значительно повышает эффективность автоматизированного тестирования и широко используется.

1. Простое начало работы

1.1 Подготовка

  • Загрузите и установите AirtestIDE с официального сайта.
  • Подготовьте мобильное устройство, убедитесь, что функция отладки по USB включена, или используйте вместо него эмулятор.

1.2 Запустите AirtestIDE

При открытии AirtestIDE запускаются две программы, одна из которых представляет собой консольную программу, которая печатает журнал операций следующим образом:

Рисунок 1. Фоновый журнал AirtestIDE

Одним из них является пользовательский интерфейс AirtestIDE, а именно:

Рисунок 2. Пользовательский интерфейс AirtestIDE

1.3 Подключение устройства

При подключении убедитесь, что устройство находится в сети.Обычно вам нужно нажать «Обновить ADB», чтобы просмотреть обновленное устройство и статус устройства, а затем дважды щелкнуть устройство, которое нужно подключить, чтобы подключиться.Если подключенное устройство является эмулятором, оплатите внимание на следующее:

  • Убедитесь, что эмулятор такой же, как версия adb в Airtest, иначе он не может быть подключен.Используйте версию adb в командной строке, чтобы просмотреть версию adb.adb в Airtest находится в папке Install_path\airtest\core\android\static\ каталог adb\windows.

  • Обязательно проверьте подключение по методу Javacap ②, чтобы избежать появления черного экрана после подключения.

Рисунок 3. Подключение устройства AirtestIDE

1.4 Позиционирование пользовательского интерфейса

Выберите Android ① во вспомогательном окне Poco и включите инспектор Poco ②, затем наведите указатель мыши на элемент управления, чтобы отобразить имя пользовательского интерфейса элемента управления ③, или дважды щелкните имя пользовательского интерфейса слева, чтобы записать его в окне редактирования сценария ④.

Рисунок 4. Расположение пользовательского интерфейса AirtestIDE

1.5 Редактирование скрипта

Напишите сценарий операции ⑤ в окне редактирования сценария. Например, используйте поиск Baidu для поиска ключевых слов Airtest. После ввода ключевых слов щелкните элемент управления Baidu, чтобы завершить поиск.

1.6 Бег

Запустите сценарий и просмотрите текущий журнал в окне просмотра журнала ⑥. Вышеупомянутые операции — это просто введение, для получения дополнительной информации см. официальную документацию.

2. Использование Airtest в многопоточности

Когда в проекте требуется устройство группового управления, Airtest будет запланирован в многопроцессном или многопоточном режиме, а платформы Airtest и Poco будут интегрированы в проект для использования Airtest в чистом коде Python, но Airtest IDE еще нужен в качестве вспомогательного Инструмент помогает завершить позиционирование элементов управления пользовательского интерфейса.Позвольте мне поделиться с вами методами и проблемами использования Airtest для управления несколькими устройствами.

2.1 Установка

Чтобы использовать Airtest в чистой среде Python, вам необходимо установить модули Airtest и Poco в среде проекта следующим образом:pip install -U airtest pocoui

2.2 Подключение нескольких устройств

Каждое устройство должно привязываться к объекту Poco отдельно. Объект Poco — это служба с именем com.netease.open.pocoservice (далее совместно именуемая pocoservice), установленная на устройстве в виде apk. Эту службу можно использовать для печати Пользовательский интерфейс устройства Пример кода для подключения нескольких устройств выглядит следующим образом:

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 (нестабильно), pocoservice перезапустится из-за существования KeepRunningInstrumentationThread, но поскольку pocoservice дает сбой, иногда его невозможно перезапустить, он циклически выдает повышение RuntimeError("unable для запуска AndroidUiaautomationPoco"), из-за чего это устройство не может нормально работать, в общем, нам нужно разобраться с этим отдельно, как показано ниже:

Обработайте исключение, созданное Airtest, и убедитесь, что служба pocoservice перезапущена.В общем случае pocoservice необходимо переустановить, то есть повторно инициализировать. Но как обнаружить исключение Poco и поймать это исключение? Вот способ использовать метод временных задач для определения состояния Poco при управлении Poco, а затем удалить ненормальный Poco и дождаться его следующего подключения.

2.4 Обработка исключений устройства

В общем, исключениями устройства являются в основном AdbError и DeviceConnectionError.Существуют разные причины таких исключений, потому что ядром устройства управления Airtest является работа через команды оболочки adb.Пока выполняются команды оболочки adb, такие ошибки могут возникать., вы можно представить так, любое действие в Airtest — это выполнение команды adb shell, для обеспечения долговременной стабильной работы проекта следует уделить особое внимание обработке таких исключений.

  • Первый вопрос

Командная функция adb оболочки Airtest реализована путем инкапсуляции subprocess.Popen и использования сообщения для получения stdout и stderr.Таким образом запустить неблокирующий подпроцесс не проблема, но при использовании команд оболочки для запуска блокирующего подпроцесса это будет зависает, когда дочерний процесс завершается или основной процесс завершается до того, как он может завершиться Иногда мы не хотим, чтобы дочерний процесс застрял, поэтому нам нужно отдельно инкапсулировать неблокирующую функцию оболочки adb, чтобы гарантировать, что программа не будет Это В этом случае, чтобы гарантировать, что процесс запускается успешно, требуется пользовательская функция для обнаружения существования процесса, как показано ниже:

    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ошибка.

  • второй вопрос

Инициализация ADB в Airtest также часто сообщает об ошибках, что напрямую приводит к сбою подключения устройства, но 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 exclude используется для перехвата возможных исключений. Здесь рекомендуется Funcy. Funcy — это библиотека Python, которая называется Swiss Army Knife. следующим образом, который реализует декоратор, называемый 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 вместо объекта устройства, переданного при его инициализации), поэтому в случае мульти- threading, команда, которая должна быть выполнена этим устройством, может быть выполнена путем переключения на другое устройство, что приведет к серии ошибок. Решение состоит в том, чтобы поддерживать очередь, чтобы убедиться, что основной поток выполняет операции Airtest, и установить G.DEVICE, где используется Airtest, чтобы гарантировать, что G.DEVICE равен устройству Poco.

3. Заключение

Airtest имеет много ям в стабильности, управлении несколькими устройствами, особенно в многопоточности. Лучше всего просмотреть исходный код, чтобы углубить свое понимание Airtest, а затем сделать несколько расширенных настраиваемых расширений на основе платформы Airtest.


PS: Для получения дополнительной технической галантереи, пожалуйста, обратите внимание на [Публичный аккаунт | xingzhe_ai] и обсудите с ходоками!