This article was first published on:Walker AI

Airtest is a UI automation test tool based on image recognition and POCO control recognition. It is used for game and App testing, and is also widely used in device group control. Its features and functions are no less than those of appium and ATX automation frameworks.

Speaking of Airtest, I can’t help but mention the AirtestIDE, a powerful GUI tool which integrates Airtest and Poco, and has built-in ADB tools, Poco-Inspector, device screen recording, script editor, UI screenshot, etc. It makes automatic testing more convenient, greatly improves the efficiency of automatic testing, and is widely used.

1. Get started

1.1 to prepare

  • Download and install the AirtestIDE from the official website.
  • Get a mobile device and make sure USB debugging is enabled, or use an emulator instead.

1.2 start AirtestIDE

When you open the AirtestIDE, you can launch two programs, one is a console program to print operation logs, as follows:

Figure 1. Background log of the AirtestIDE

A UI interface of the AirtestIDE reads as follows:

Figure 2. AirtestIDE UI interface

1.3 Connecting Devices

Make sure the device is online when connecting. Usually, you need to click Refresh ADB to view the updated device and device status, and then double click the device to connect. If the connected device is an emulator, note the following:

  • Make sure that the emulator is the same as the ADB version in Airtest, otherwise it will not be able to connect. Adb in Airtest is under Install_path\ Airtest \core\ Android \static\adb\ Windows.

  • Ensure that the Javacap mode is selected to avoid black screen after connection.

Figure 3. The AirtestIDE device is connected

1.4 the UI localization

Select Android and enable the Poco Inspector in the Poco helper window, then place the mouse over the control to display the UI name of the control (3), or double-click the UI name on the left to write it to the script editing window (4).

Figure 4. Positioning of the AirtestIDE UI

1.5 Script Editing

Write operation script ⑤ in script editing window, for example, use Baidu search to search Airtest keyword, enter the keyword and click baidu control to complete the search.

1.6 run

Run the script and view the run Log in the Log view window. The above operations are just a simple introduction. For more operations, please refer to the official documentation.

2. Airtest is used in multithreading

When a teamwork control device is needed in a project, Airtest will be scheduled in multi-process or multi-thread mode, and Airtest and Poco framework will be integrated into the project to use Airtest in pure Python code, but Airtest IDE is still needed as an auxiliary tool to help complete the positioning of UI controls. Here’s how to use Airtest to control multiple devices and some of the problems.

2.1 installation

PIP install -u Airtest pocoui To use Airtest in a pure Python environment, install the Airtest and Poco modules as follows: PIP install -u Airtest pocoui

2.2 Multi-device Connection

Each device requires a separate binding a Poco object, Poco objects is an apk installed inside the equipment, in the form of a called com.net help ease. Open the pocoservice services (hereinafter generally referred to as the pocoservice), the service can be used in the UI tree and simulate click printing devices, etc., Example code for a multi-device connection is as follows:

from airtest.core.api import *
from poco.drivers.android.uiautomation import AndroidUiautomationPoco
    

# filter log
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)
Copy the code

2.3 Poco management

This does ensure that each device is bound to a separate Poco object, but it is not conducive to managing Poco objects, such as checking the survival status of each Poco. Therefore, a container is needed to manage and create Poco objects. Here, a method is used as a reference in the source code. It uses singleton mode to manage the creation of Poco and save it as a dictionary, which not only ensures that each device has a separate Poco, but also facilitates obtaining Poco objects by device string number.

    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]
Copy the code

AndroidUiautomationPoco within the initialization, maintains a thread pocoservice KeepRunningInstrumentationThread monitoring, monitoring pocoservice state to prevent abnormal exit.

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)Copy the code

Problems here is that once pocoservice problems (unstable), because of the existence of KeepRunningInstrumentationThread pocoservice will restart, but because of pocoservice service after the collapse, sometimes can’t restart, Raise RuntimeError(” Unable to launch AndroidUiautomationPoco”) will be raised in a loop, causing the device to fail to run normally. In general, we need to deal with it separately as follows:

Handle exceptions thrown by Airtest and ensure that the PocoService service is restarted. In general, pocoService needs to be reinstalled, that is, reinitialized. But how do you detect a Poco exception and catch it? This section describes a way to manage Poco. A scheduled task is used to detect the STATUS of the Poco, and then remove the abnormal Poco and wait for its next connection.

2.4 Device Exception Handling

In general, device exceptions are mainly AdbError and DeviceConnectionError. The causes of these exceptions are various. The core of Airtest control device is to operate through ADB shell commands. You can assume that any action in Airtest is executing adb shell commands, and special care should be taken to handle such exceptions to ensure long-term stability of the project.

  • First question

The Airtest adb shell function encapsulates subprocess.Popen and uses a communicate stdout and stderr to start a non-blocking child process. But when use the shell command to start a blocking type when the child process will be stuck, has been waiting for the child to end or exit to exit the main process, and sometimes we don’t want the quilt process gets stuck, so no need to separate packaging a blocking the adb shell functions, guarantee program won’t be stuck, this situation to ensure the success of the process started, A custom function is required to detect the existence of the process, as follows:

    def rshell_nowait(self, command, proc_name) :
        """ Calls the remote device's shell command and returns immediately, killing the current process. :param command: shell command: param proc_name: name of the process started by the command, which is used to stop the process :return: success: pid of the process started, failed :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()  # This process is closed immediately without affecting child processes started on the remote device
                pid = self.get_rpid(proc_name)
                if pid:
            	return pid
    
    def get_rpid(self, proc_name) :
        Run the ps command to query the PID corresponding to proc_name on the remote device :param proc_name: process name: return: successful: process PID, failed :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
Copy the code

Note: Open processes through subprocess.Popen must be closed immediately after use to prevent Too many open files errors.

  • Second question

However, Airtest does not catch this error directly, so we need to handle this error on top and add a retry mechanism, such as the following, as a decorator or using retrime.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
Copy the code

Funcy is a Swiss Army knife Python library. A silent function is used to dress up functions that may cause exceptions. The source code for silent is as follows: It implements a decorator named Ignore to handle exceptions. Funcy, of course, also encapsulates many of the tools used in python’s everyday work, so check out funcy’s source code if you’re interested.

def silent(func) :
      """ Ignore the wrong call """
      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)
      
  # Refer to the usage method
  import json
  
  str1 = '{a: 1, 'b':2}'
  json_str = silent(json.loads)(str1)    
Copy the code
  • Third question

Airtest calls g.content to get the current device when it executes commands (using Poco objects the underlying device uses G.content instead of the device object passed in when it initializations), so in multithreaded situations, commands that should be executed by this device may be switched to another device, resulting in a series of errors. The solution is to maintain a queue, ensure that the main thread is performing the Airtest operation, and set g.davis where Airtest is used to ensure that g.Davis is equal to the Poco device.

3. The conclusion

Airtest has many pitfalls in stability, multi-device control and especially multithreading. Better read the source code to deepen the understanding of Airtest, and then based on the Airtest framework to do some advanced custom extensions.

PS: more dry technology, pay attention to the public, | xingzhe_ai 】, and walker to discuss together!