<s> import os <EOL> import sys <EOL> import time <EOL> from wlauto . core . extension import Parameter <EOL> from wlauto . core . workload import Workload <EOL> from wlauto . core . resource import NO_ONE <EOL> from wlauto . common . resources import ExtensionAsset , Executable <EOL> from wlauto . exceptions import WorkloadError , ResourceError , ConfigError <EOL> from wlauto . utils . android import ApkInfo , ANDROID_NORMAL_PERMISSIONS <EOL> from wlauto . utils . types import boolean <EOL> import wlauto . common . android . resources <EOL> DELAY = <NUM_LIT:5> <EOL> class UiAutomatorWorkload ( Workload ) : <EOL> """<STR_LIT>""" <EOL> supported_platforms = [ '<STR_LIT>' ] <EOL> uiauto_package = '<STR_LIT>' <EOL> uiauto_class = '<STR_LIT>' <EOL> uiauto_method = '<STR_LIT>' <EOL> run_timeout = <NUM_LIT:4> * <NUM_LIT> <EOL> def __init__ ( self , device , _call_super = True , ** kwargs ) : <EOL> if _call_super : <EOL> super ( UiAutomatorWorkload , self ) . __init__ ( device , ** kwargs ) <EOL> self . uiauto_file = None <EOL> self . device_uiauto_file = None <EOL> self . command = None <EOL> self . uiauto_params = { } <EOL> def init_resources ( self , context ) : <EOL> self . uiauto_file = context . resolver . get ( wlauto . common . android . resources . JarFile ( self ) ) <EOL> if not self . uiauto_file : <EOL> raise ResourceError ( '<STR_LIT>' . format ( self . name ) ) <EOL> self . device_uiauto_file = self . device . path . join ( self . device . working_directory , <EOL> os . path . basename ( self . uiauto_file ) ) <EOL> if not self . uiauto_package : <EOL> self . uiauto_package = os . path . splitext ( os . path . basename ( self . uiauto_file ) ) [ <NUM_LIT:0> ] <EOL> def setup ( self , context ) : <EOL> method_string = '<STR_LIT>' . format ( self . uiauto_package , self . uiauto_class , self . uiauto_method ) <EOL> params_dict = self . uiauto_params <EOL> params_dict [ '<STR_LIT>' ] = self . device . working_directory <EOL> params = '<STR_LIT>' <EOL> for k , v in self . uiauto_params . iteritems ( ) : <EOL> params += '<STR_LIT>' . format ( k , v ) <EOL> self . command = '<STR_LIT>' . format ( self . device_uiauto_file , params , method_string ) <EOL> self . device . push_file ( self . uiauto_file , self . device_uiauto_file ) <EOL> self . device . killall ( '<STR_LIT>' ) <EOL> def run ( self , context ) : <EOL> result = self . device . execute ( self . command , self . run_timeout ) <EOL> if '<STR_LIT>' in result : <EOL> raise WorkloadError ( result ) <EOL> else : <EOL> self . logger . debug ( result ) <EOL> time . sleep ( DELAY ) <EOL> def update_result ( self , context ) : <EOL> pass <EOL> def teardown ( self , context ) : <EOL> self . device . delete_file ( self . device_uiauto_file ) <EOL> def validate ( self ) : <EOL> if not self . uiauto_file : <EOL> raise WorkloadError ( '<STR_LIT>' . format ( self . name ) ) <EOL> if not self . uiauto_package : <EOL> raise WorkloadError ( '<STR_LIT>' . format ( self . name ) ) <EOL> class ApkWorkload ( Workload ) : <EOL> """<STR_LIT>""" <EOL> package = None <EOL> activity = None <EOL> view = None <EOL> supported_platforms = [ '<STR_LIT>' ] <EOL> parameters = [ <EOL> Parameter ( '<STR_LIT>' , kind = int , default = <NUM_LIT> , <EOL> description = '<STR_LIT>' ) , <EOL> Parameter ( '<STR_LIT>' , kind = boolean , default = True , <EOL> description = '''<STR_LIT>''' ) , <EOL> Parameter ( '<STR_LIT>' , kind = boolean , default = False , <EOL> description = '''<STR_LIT>''' ) , <EOL> Parameter ( '<STR_LIT>' , kind = boolean , default = False , <EOL> description = '<STR_LIT>' ) , <EOL> ] <EOL> def __init__ ( self , device , _call_super = True , ** kwargs ) : <EOL> if _call_super : <EOL> super ( ApkWorkload , self ) . __init__ ( device , ** kwargs ) <EOL> self . apk_file = None <EOL> self . apk_version = None <EOL> self . logcat_log = None <EOL> def init_resources ( self , context ) : <EOL> self . apk_file = context . resolver . get ( wlauto . common . android . resources . ApkFile ( self ) , <EOL> version = getattr ( self , '<STR_LIT:version>' , None ) , <EOL> strict = self . check_apk ) <EOL> def validate ( self ) : <EOL> if self . check_apk : <EOL> if not self . apk_file : <EOL> raise WorkloadError ( '<STR_LIT>' . format ( self . name ) ) <EOL> else : <EOL> if self . force_install : <EOL> raise ConfigError ( '<STR_LIT>' ) <EOL> def setup ( self , context ) : <EOL> self . initialize_package ( context ) <EOL> self . launch_package ( ) <EOL> self . device . execute ( '<STR_LIT>' ) <EOL> self . device . clear_logcat ( ) <EOL> def initialize_package ( self , context ) : <EOL> installed_version = self . device . get_installed_package_version ( self . package ) <EOL> if self . check_apk : <EOL> self . initialize_with_host_apk ( context , installed_version ) <EOL> else : <EOL> if not installed_version : <EOL> message = '''<STR_LIT>''' <EOL> raise WorkloadError ( message . format ( self . package ) ) <EOL> message = '<STR_LIT>' <EOL> self . logger . debug ( message . format ( installed_version ) ) <EOL> self . reset ( context ) <EOL> self . apk_version = installed_version <EOL> context . add_classifiers ( apk_version = self . apk_version ) <EOL> def initialize_with_host_apk ( self , context , installed_version ) : <EOL> host_version = ApkInfo ( self . apk_file ) . version_name <EOL> if installed_version != host_version : <EOL> if installed_version : <EOL> message = '<STR_LIT>' <EOL> self . logger . debug ( message . format ( os . path . basename ( self . apk_file ) , <EOL> host_version , installed_version ) ) <EOL> else : <EOL> message = '<STR_LIT>' <EOL> self . logger . debug ( message . format ( os . path . basename ( self . apk_file ) , <EOL> host_version ) ) <EOL> self . force_install = True <EOL> else : <EOL> message = '<STR_LIT>' <EOL> self . logger . debug ( message . format ( os . path . basename ( self . apk_file ) , <EOL> host_version ) ) <EOL> if self . force_install : <EOL> if installed_version : <EOL> self . device . uninstall ( self . package ) <EOL> if self . install_apk ( context ) : <EOL> self . apk_version = host_version <EOL> else : <EOL> self . apk_version = installed_version <EOL> else : <EOL> self . apk_version = installed_version <EOL> self . reset ( context ) <EOL> def launch_package ( self ) : <EOL> if not self . activity : <EOL> output = self . device . execute ( '<STR_LIT>' . format ( self . package ) ) <EOL> else : <EOL> output = self . device . execute ( '<STR_LIT>' . format ( self . package , self . activity ) ) <EOL> if '<STR_LIT>' in output : <EOL> self . device . execute ( '<STR_LIT>' . format ( self . package ) ) <EOL> raise WorkloadError ( output ) <EOL> self . logger . debug ( output ) <EOL> def reset ( self , context ) : <EOL> self . device . execute ( '<STR_LIT>' . format ( self . package ) ) <EOL> self . device . execute ( '<STR_LIT>' . format ( self . package ) ) <EOL> if self . device . get_sdk_version ( ) >= <NUM_LIT> : <EOL> self . _grant_requested_permissions ( ) <EOL> def install_apk ( self , context ) : <EOL> success = False <EOL> output = self . device . install ( self . apk_file , self . install_timeout ) <EOL> if '<STR_LIT>' in output : <EOL> if '<STR_LIT>' in output : <EOL> self . logger . warn ( '<STR_LIT>' ) <EOL> self . reset ( context ) <EOL> else : <EOL> raise WorkloadError ( output ) <EOL> else : <EOL> self . logger . debug ( output ) <EOL> success = True <EOL> self . do_post_install ( context ) <EOL> return success <EOL> def _grant_requested_permissions ( self ) : <EOL> dumpsys_output = self . device . execute ( command = "<STR_LIT>" . format ( self . package ) ) <EOL> permissions = [ ] <EOL> lines = iter ( dumpsys_output . splitlines ( ) ) <EOL> for line in lines : <EOL> if "<STR_LIT>" in line : <EOL> break <EOL> for line in lines : <EOL> if "<STR_LIT>" in line : <EOL> permissions . append ( line . split ( "<STR_LIT::>" ) [ <NUM_LIT:0> ] . strip ( ) ) <EOL> else : <EOL> break <EOL> for permission in permissions : <EOL> permission_name = permission . rsplit ( '<STR_LIT:.>' , <NUM_LIT:1> ) [ <NUM_LIT:1> ] <EOL> if permission_name not in ANDROID_NORMAL_PERMISSIONS : <EOL> self . device . execute ( "<STR_LIT>" . format ( self . package , permission ) ) <EOL> def do_post_install ( self , context ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> def run ( self , context ) : <EOL> pass <EOL> def update_result ( self , context ) : <EOL> self . logcat_log = os . path . join ( context . output_directory , '<STR_LIT>' ) <EOL> self . device . dump_logcat ( self . logcat_log ) <EOL> context . add_iteration_artifact ( name = '<STR_LIT>' , <EOL> path = '<STR_LIT>' , <EOL> kind = '<STR_LIT>' , <EOL> description = '<STR_LIT>' ) <EOL> def teardown ( self , context ) : <EOL> self . device . execute ( '<STR_LIT>' . format ( self . package ) ) <EOL> if self . uninstall_apk : <EOL> self . device . uninstall ( self . package ) <EOL> AndroidBenchmark = ApkWorkload <EOL> class ReventWorkload ( Workload ) : <EOL> default_setup_timeout = <NUM_LIT:5> * <NUM_LIT> <EOL> default_run_timeout = <NUM_LIT:10> * <NUM_LIT> <EOL> def __init__ ( self , device , _call_super = True , ** kwargs ) : <EOL> if _call_super : <EOL> super ( ReventWorkload , self ) . __init__ ( device , ** kwargs ) <EOL> devpath = self . device . path <EOL> self . on_device_revent_binary = devpath . join ( self . device . binaries_directory , '<STR_LIT>' ) <EOL> self . on_device_setup_revent = devpath . join ( self . device . working_directory , '<STR_LIT>' . format ( self . device . name ) ) <EOL> self . on_device_run_revent = devpath . join ( self . device . working_directory , '<STR_LIT>' . format ( self . device . name ) ) <EOL> self . setup_timeout = kwargs . get ( '<STR_LIT>' , self . default_setup_timeout ) <EOL> self . run_timeout = kwargs . get ( '<STR_LIT>' , self . default_run_timeout ) <EOL> self . revent_setup_file = None <EOL> self . revent_run_file = None <EOL> def init_resources ( self , context ) : <EOL> self . revent_setup_file = context . resolver . get ( wlauto . common . android . resources . ReventFile ( self , '<STR_LIT>' ) ) <EOL> self . revent_run_file = context . resolver . get ( wlauto . common . android . resources . ReventFile ( self , '<STR_LIT>' ) ) <EOL> def setup ( self , context ) : <EOL> self . _check_revent_files ( context ) <EOL> self . device . killall ( '<STR_LIT>' ) <EOL> command = '<STR_LIT>' . format ( self . on_device_revent_binary , self . on_device_setup_revent ) <EOL> self . device . execute ( command , timeout = self . setup_timeout ) <EOL> def run ( self , context ) : <EOL> command = '<STR_LIT>' . format ( self . on_device_revent_binary , self . on_device_run_revent ) <EOL> self . logger . debug ( '<STR_LIT>' . format ( os . path . basename ( self . on_device_run_revent ) ) ) <EOL> self . device . execute ( command , timeout = self . run_timeout ) <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> def update_result ( self , context ) : <EOL> pass <EOL> def teardown ( self , context ) : <EOL> self . device . delete_file ( self . on_device_setup_revent ) <EOL> self . device . delete_file ( self . on_device_run_revent ) <EOL> def _check_revent_files ( self , context ) : <EOL> revent_binary = context . resolver . get ( Executable ( NO_ONE , self . device . abi , '<STR_LIT>' ) ) <EOL> if not os . path . isfile ( revent_binary ) : <EOL> message = '<STR_LIT>' . format ( revent_binary ) <EOL> message += '<STR_LIT>' <EOL> raise WorkloadError ( message ) <EOL> if not self . revent_setup_file : <EOL> message = '<STR_LIT>' . format ( self . device . name ) <EOL> raise WorkloadError ( message ) <EOL> if not self . revent_run_file : <EOL> message = '<STR_LIT>' . format ( self . device . name ) <EOL> raise WorkloadError ( message ) <EOL> self . on_device_revent_binary = self . device . install_executable ( revent_binary ) <EOL> self . device . push_file ( self . revent_run_file , self . on_device_run_revent ) <EOL> self . device . push_file ( self . revent_setup_file , self . on_device_setup_revent ) <EOL> class AndroidUiAutoBenchmark ( UiAutomatorWorkload , AndroidBenchmark ) : <EOL> supported_platforms = [ '<STR_LIT>' ] <EOL> def __init__ ( self , device , ** kwargs ) : <EOL> UiAutomatorWorkload . __init__ ( self , device , ** kwargs ) <EOL> AndroidBenchmark . __init__ ( self , device , _call_super = False , ** kwargs ) <EOL> def init_resources ( self , context ) : <EOL> UiAutomatorWorkload . init_resources ( self , context ) <EOL> AndroidBenchmark . init_resources ( self , context ) <EOL> def setup ( self , context ) : <EOL> UiAutomatorWorkload . setup ( self , context ) <EOL> AndroidBenchmark . setup ( self , context ) <EOL> def update_result ( self , context ) : <EOL> UiAutomatorWorkload . update_result ( self , context ) <EOL> AndroidBenchmark . update_result ( self , context ) <EOL> def teardown ( self , context ) : <EOL> UiAutomatorWorkload . teardown ( self , context ) <EOL> AndroidBenchmark . teardown ( self , context ) <EOL> class GameWorkload ( ApkWorkload , ReventWorkload ) : <EOL> """<STR_LIT>""" <EOL> asset_file = None <EOL> saved_state_file = None <EOL> view = '<STR_LIT>' <EOL> loading_time = <NUM_LIT:10> <EOL> supported_platforms = [ '<STR_LIT>' ] <EOL> parameters = [ <EOL> Parameter ( '<STR_LIT>' , default = <NUM_LIT> , override = True ) , <EOL> Parameter ( '<STR_LIT>' , kind = int , default = <NUM_LIT> , <EOL> description = '<STR_LIT>' ) , <EOL> Parameter ( '<STR_LIT>' , kind = bool , default = True , <EOL> description = """<STR_LIT>""" ) , <EOL> ] <EOL> def __init__ ( self , device , ** kwargs ) : <EOL> ApkWorkload . __init__ ( self , device , ** kwargs ) <EOL> ReventWorkload . __init__ ( self , device , _call_super = False , ** kwargs ) <EOL> self . logcat_process = None <EOL> self . module_dir = os . path . dirname ( sys . modules [ self . __module__ ] . __file__ ) <EOL> self . revent_dir = os . path . join ( self . module_dir , '<STR_LIT>' ) <EOL> def init_resources ( self , context ) : <EOL> ApkWorkload . init_resources ( self , context ) <EOL> ReventWorkload . init_resources ( self , context ) <EOL> def setup ( self , context ) : <EOL> ApkWorkload . setup ( self , context ) <EOL> self . logger . debug ( '<STR_LIT>' ) <EOL> time . sleep ( self . loading_time ) <EOL> ReventWorkload . setup ( self , context ) <EOL> def do_post_install ( self , context ) : <EOL> ApkWorkload . do_post_install ( self , context ) <EOL> self . _deploy_assets ( context , self . assets_push_timeout ) <EOL> def reset ( self , context ) : <EOL> self . device . execute ( '<STR_LIT>' . format ( self . package ) ) <EOL> if self . saved_state_file : <EOL> self . _deploy_resource_tarball ( context , self . saved_state_file ) <EOL> else : <EOL> if self . clear_data_on_reset : <EOL> self . device . execute ( '<STR_LIT>' . format ( self . package ) ) <EOL> self . _deploy_assets ( context ) <EOL> def run ( self , context ) : <EOL> ReventWorkload . run ( self , context ) <EOL> def teardown ( self , context ) : <EOL> if not self . saved_state_file : <EOL> ApkWorkload . teardown ( self , context ) <EOL> else : <EOL> self . device . execute ( '<STR_LIT>' . format ( self . package ) ) <EOL> ReventWorkload . teardown ( self , context ) <EOL> def _deploy_assets ( self , context , timeout = <NUM_LIT> ) : <EOL> if self . asset_file : <EOL> self . _deploy_resource_tarball ( context , self . asset_file , timeout ) <EOL> if self . saved_state_file : <EOL> self . _deploy_resource_tarball ( context , self . saved_state_file , timeout ) <EOL> def _deploy_resource_tarball ( self , context , resource_file , timeout = <NUM_LIT> ) : <EOL> kind = '<STR_LIT:data>' <EOL> if '<STR_LIT::>' in resource_file : <EOL> kind , resource_file = resource_file . split ( '<STR_LIT::>' , <NUM_LIT:1> ) <EOL> ondevice_cache = self . device . path . join ( self . device . resource_cache , self . name , resource_file ) <EOL> if not self . device . file_exists ( ondevice_cache ) : <EOL> asset_tarball = context . resolver . get ( ExtensionAsset ( self , resource_file ) ) <EOL> if not asset_tarball : <EOL> message = '<STR_LIT>' <EOL> raise WorkloadError ( message . format ( resource_file , self . name ) ) <EOL> self . device . push_file ( asset_tarball , ondevice_cache , timeout = timeout ) <EOL> device_asset_directory = self . device . path . join ( self . device . external_storage_directory , '<STR_LIT>' , kind ) <EOL> deploy_command = '<STR_LIT>' . format ( device_asset_directory , <EOL> self . device . busybox , <EOL> ondevice_cache ) <EOL> self . device . execute ( deploy_command , timeout = timeout , as_root = True ) </s>
