[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 : 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 : 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 : 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 = []