Source code for poco.drivers.android.uiautomation

# coding=utf-8
__author__ = 'lxn3032'


import os
import requests
import time
import warnings
import threading
import atexit

from airtest.core.android.ime import YosemiteIme
from airtest.core.error import AdbShellError, AirtestError

from hrpc.client import RpcClient
from hrpc.transport.http import HttpTransport
from poco.pocofw import Poco
from poco.agent import PocoAgent
from poco.sdk.Attributor import Attributor
from poco.sdk.interfaces.screen import ScreenInterface
from poco.utils.hrpc.hierarchy import RemotePocoHierarchy
from poco.utils.airtest.input import AirtestInput
from poco.utils import six
from poco.utils.device import default_device
from poco.drivers.android.utils.installation import install, uninstall

__all__ = ['AndroidUiautomationPoco', 'AndroidUiautomationHelper']
this_dir = os.path.dirname(os.path.realpath(__file__))
PocoServicePackage = 'com.netease.open.pocoservice'
PocoServicePackageTest = 'com.netease.open.pocoservice.test'
UiAutomatorPackage = 'com.github.uiautomator'


class AndroidRpcClient(RpcClient):
    def __init__(self, endpoint):
        self.endpoint = endpoint
        super(AndroidRpcClient, self).__init__(HttpTransport)

    def initialize_transport(self):
        return HttpTransport(self.endpoint, self)


# deprecated
class AttributorWrapper(Attributor):
    """
    部分手机上仍不支持Accessibility.ACTION_SET_TEXT,使用YosemiteIme还是兼容性最好的方案
    这个class会hook住set_text,然后改用ime的text方法
    """

    def __init__(self, remote, ime):
        self.remote = remote
        self.ime = ime

    def getAttr(self, node, attrName):
        return self.remote.getAttr(node, attrName)

    def setAttr(self, node, attrName, attrVal):
        if attrName == 'text' and attrVal != '':
            # 先清除了再设置,虽然这样不如直接用ime的方法好,但是也能凑合用着
            current_val = self.remote.getAttr(node, 'text')
            if current_val:
                self.remote.setAttr(node, 'text', '')
            self.ime.text(attrVal)
        else:
            self.remote.setAttr(node, attrName, attrVal)


class ScreenWrapper(ScreenInterface):
    def __init__(self, screen):
        super(ScreenWrapper, self).__init__()
        self.screen = screen

    def getScreen(self, width):
        # Android上PocoService的实现为仅返回b64编码的图像,格式固定位jpg
        b64img = self.screen.getScreen(width)
        return b64img, 'jpg'

    def getPortSize(self):
        return self.screen.getPortSize()


class AndroidPocoAgent(PocoAgent):
    def __init__(self, endpoint, ime, use_airtest_input=False):
        self.client = AndroidRpcClient(endpoint)
        remote_poco = self.client.remote('poco-uiautomation-framework')
        dumper = remote_poco.dumper
        selector = remote_poco.selector
        attributor = remote_poco.attributor
        hierarchy = RemotePocoHierarchy(dumper, selector, attributor)

        if use_airtest_input:
            inputer = AirtestInput()
        else:
            inputer = remote_poco.inputer
        super(AndroidPocoAgent, self).__init__(hierarchy, inputer, ScreenWrapper(remote_poco.screen), None)


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 not self.stopped():
                self.poco._start_instrument(self.port_to_ping)  # 尝试重启
                time.sleep(1)


[docs]class AndroidUiautomationPoco(Poco): """ Poco Android implementation for testing **Android native apps**. Args: device (:py:obj:`Device`): :py:obj:`airtest.core.device.Device` instance provided by ``airtest``. leave the parameter default and the default device will be chosen. more details refer to ``airtest doc`` using_proxy (:py:obj:`bool`): whether use adb forward to connect the Android device or not force_restart (:py:obj:`bool`): whether always restart the poco-service-demo running on Android device or not options: see :py:class:`poco.pocofw.Poco` Examples: The simplest way to initialize AndroidUiautomationPoco instance and no matter your device network status:: from poco.drivers.android.uiautomation import AndroidUiautomationPoco poco = AndroidUiautomationPoco() poco('android:id/title').click() ... """ def __init__(self, device=None, using_proxy=True, force_restart=False, use_airtest_input=False, **options): # 加这个参数为了不在最新的pocounit方案中每步都截图 self.screenshot_each_action = True if options.get('screenshot_each_action') is False: self.screenshot_each_action = False self.device = device or default_device() self.adb_client = self.device.adb if using_proxy: self.device_ip = self.adb_client.host or "127.0.0.1" else: self.device_ip = self.device.get_ip_address() # save current top activity (@nullable) try: current_top_activity_package = self.device.get_top_activity_name() except AirtestError as e: # 在一些极端情况下,可能获取不到top activity的信息 print(e) current_top_activity_package = None if current_top_activity_package is not None: current_top_activity_package = current_top_activity_package.split('/')[0] # install ime self.ime = YosemiteIme(self.adb_client) # install self._instrument_proc = None self._install_service() # forward self.forward_list = [] if using_proxy: p0, _ = self.adb_client.setup_forward("tcp:10080") p1, _ = self.adb_client.setup_forward("tcp:10081") self.forward_list.extend(["tcp:%s" % p0, "tcp:%s" % p1]) else: p0 = 10080 p1 = 10081 # start ready = self._start_instrument(p0, force_restart=force_restart) if not ready: # 之前启动失败就卸载重装,现在改为尝试kill进程或卸载uiautomator self._kill_uiautomator() ready = self._start_instrument(p0) if current_top_activity_package is not None: current_top_activity2 = self.device.get_top_activity_name() if current_top_activity2 is None or current_top_activity_package not in current_top_activity2: self.device.start_app(current_top_activity_package, activity=True) if not ready: raise RuntimeError("unable to launch AndroidUiautomationPoco") if ready: # 首次启动成功后,在后台线程里监控这个进程的状态,保持让它不退出 self._keep_running_thread = KeepRunningInstrumentationThread(self, p0) self._keep_running_thread.start() endpoint = "http://{}:{}".format(self.device_ip, p1) agent = AndroidPocoAgent(endpoint, self.ime, use_airtest_input) super(AndroidUiautomationPoco, self).__init__(agent, **options) def _install_service(self): updated = install(self.adb_client, os.path.join(this_dir, 'lib', 'pocoservice-debug.apk')) return updated def _is_running(self, package_name): """ use ps |grep to check whether the process exists :param package_name: package name(e.g., com.github.uiautomator) or regular expression(e.g., poco\|airtest\|uiautomator\|airbase) :return: pid or None """ cmd = r' |echo $(grep -E {package_name})'.format(package_name=package_name) if self.device.sdk_version > 25: cmd = r'ps -A' + cmd else: cmd = r'ps' + cmd processes = self.adb_client.shell(cmd).splitlines() for ps in processes: if ps: ps = ps.split() return ps[1] return None def _start_instrument(self, port_to_ping, force_restart=False): if not force_restart: try: state = requests.get('http://{}:{}/uiautomation/connectionState'.format(self.device_ip, port_to_ping), timeout=10) state = state.json() if state.get('connected'): # skip starting instrumentation if UiAutomation Service already connected. return True except: pass if self._instrument_proc is not None: if self._instrument_proc.poll() is None: self._instrument_proc.kill() self._instrument_proc = None ready = False # self.adb_client.shell(['am', 'force-stop', PocoServicePackage]) # 启动instrument之前,先把主类activity启动起来,不然instrumentation可能失败 self.adb_client.shell('am start -n {}/.TestActivity'.format(PocoServicePackage)) instrumentation_cmd = [ 'am', 'instrument', '-w', '-e', 'debug', 'false', '-e', 'class', '{}.InstrumentedTestAsLauncher'.format(PocoServicePackage), '{}/androidx.test.runner.AndroidJUnitRunner'.format(PocoServicePackage)] self._instrument_proc = self.adb_client.start_shell(instrumentation_cmd) def cleanup_proc(proc): def wrapped(): try: proc.kill() except: pass return wrapped atexit.register(cleanup_proc(self._instrument_proc)) time.sleep(2) for i in range(10): try: requests.get('http://{}:{}'.format(self.device_ip, port_to_ping), timeout=10) ready = True break except requests.exceptions.Timeout: break except requests.exceptions.ConnectionError: if self._instrument_proc.poll() is not None: warnings.warn("[pocoservice.apk] instrumentation test server process is no longer alive") stdout = self._instrument_proc.stdout.read() stderr = self._instrument_proc.stderr.read() print('[pocoservice.apk] stdout: {}'.format(stdout)) print('[pocoservice.apk] stderr: {}'.format(stderr)) time.sleep(1) print("still waiting for uiautomation ready.") try: self.adb_client.shell( ['monkey', '-p', {PocoServicePackage}, '-c', 'android.intent.category.LAUNCHER', '1']) except Exception as e: pass self.adb_client.shell('am start -n {}/.TestActivity'.format(PocoServicePackage)) instrumentation_cmd = [ 'am', 'instrument', '-w', '-e', 'debug', 'false', '-e', 'class', '{}.InstrumentedTestAsLauncher'.format(PocoServicePackage), '{}/androidx.test.runner.AndroidJUnitRunner'.format(PocoServicePackage)] self._instrument_proc = self.adb_client.start_shell(instrumentation_cmd) continue return ready def _kill_uiautomator(self): """ poco-service无法与其他instrument启动的apk同时存在,因此在启动前,需要杀掉一些可能的进程: 比如 io.appium.uiautomator2.server, com.github.uiautomator, com.netease.open.pocoservice等 :return: """ pid = self._is_running("uiautomator") if pid: warnings.warn('{} should not run together with "uiautomator". "uiautomator" will be killed.' .format(self.__class__.__name__)) self.adb_client.shell(['am', 'force-stop', PocoServicePackage]) try: self.adb_client.shell(['kill', pid]) except AdbShellError: # 没有root权限 uninstall(self.adb_client, UiAutomatorPackage) def on_pre_action(self, action, ui, args): if self.screenshot_each_action: # airteset log用 from airtest.core.api import snapshot msg = repr(ui) if not isinstance(msg, six.text_type): msg = msg.decode('utf-8') snapshot(msg=msg) def stop_running(self): print('[pocoservice.apk] stopping PocoService') self._keep_running_thread.stop() self._keep_running_thread.join(3) self.remove_forwards() self.adb_client.shell(['am', 'force-stop', PocoServicePackage]) def remove_forwards(self): for p in self.forward_list: self.adb_client.remove_forward(p) self.forward_list = []
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]