# Copyright 2015 The Chromium Authors # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. import copy import logging import os import re from devil.android import apk_helper from pylib import constants from pylib.base import base_test_result from pylib.base import test_exception from pylib.base import test_instance from pylib.base import test_run from pylib.constants import host_paths from pylib.instrumentation import instrumentation_parser from pylib.instrumentation import test_result from pylib.symbols import deobfuscator from pylib.symbols import stack_symbolizer from pylib.utils import dexdump from pylib.utils import gold_utils from pylib.utils import test_filter with host_paths.SysPath(host_paths.BUILD_UTIL_PATH): from lib.common import unittest_util # Ref: http://developer.android.com/reference/android/app/Activity.html _ACTIVITY_RESULT_CANCELED = 0 _ACTIVITY_RESULT_OK = -1 _COMMAND_LINE_PARAMETER = 'cmdlinearg-parameter' _DEFAULT_ANNOTATIONS = [ 'SmallTest', 'MediumTest', 'LargeTest', 'EnormousTest', 'IntegrationTest', ] # This annotation is for disabled tests that should not be run in Test Reviver. _DO_NOT_REVIVE_ANNOTATIONS = ['DoNotRevive', 'Manual'] _EXCLUDE_UNLESS_REQUESTED_ANNOTATIONS = ['DisabledTest', 'FlakyTest', 'Manual'] _VALID_ANNOTATIONS = set( _DEFAULT_ANNOTATIONS + _DO_NOT_REVIVE_ANNOTATIONS + _EXCLUDE_UNLESS_REQUESTED_ANNOTATIONS ) _BASE_INSTRUMENTATION_CLASS_NAME = ( 'org.chromium.base.test.BaseChromiumAndroidJUnitRunner' ) _SKIP_PARAMETERIZATION = 'SkipCommandLineParameterization' _PARAMETERIZED_COMMAND_LINE_FLAGS = 'ParameterizedCommandLineFlags' _PARAMETERIZED_COMMAND_LINE_FLAGS_SWITCHES = ( 'ParameterizedCommandLineFlags$Switches' ) _NATIVE_CRASH_RE = re.compile('(process|native) crash', re.IGNORECASE) # The ID of the bundle value Instrumentation uses to report which test index the # results are for in a collection of tests. Note that this index is 1-based. _BUNDLE_CURRENT_ID = 'current' # The ID of the bundle value Instrumentation uses to report the test class. _BUNDLE_CLASS_ID = 'class' # The ID of the bundle value Instrumentation uses to report the test name. _BUNDLE_TEST_ID = 'test' # The ID of the bundle value Instrumentation uses to report the crash stack, if # the test crashed. _BUNDLE_STACK_ID = 'stack' # The ID of the bundle value Chrome uses to report the test duration. _BUNDLE_DURATION_ID = 'duration_ms' # The following error messages are too general to be useful in failure # clustering. The runner doesn't report failure reason when such failure # reason is parsed from test logs. _BANNED_FAILURE_REASONS = [ # Default error message from org.chromium.base.test.util.CallbackHelper # when timeout at expecting call back. 'java.util.concurrent.TimeoutException: waitForCallback timed out!', ] class MissingSizeAnnotationError(test_exception.TestException): def __init__(self, class_name): super().__init__( class_name + ': Test method is missing required size annotation. Add one of: ' + ', '.join('@' + a for a in _VALID_ANNOTATIONS) ) class CommandLineParameterizationException(test_exception.TestException): pass def GenerateTestResults( result_code, result_bundle, statuses, duration_ms, device_abi, symbolizer ): """Generate test results from |statuses|. Args: result_code: The overall status code as an integer. result_bundle: The summary bundle dump as a dict. statuses: A list of 2-tuples containing: - the status code as an integer - the bundle dump as a dict mapping string keys to string values Note that this is the same as the third item in the 3-tuple returned by |_ParseAmInstrumentRawOutput|. duration_ms: The duration of the test in milliseconds. device_abi: The device_abi, which is needed for symbolization. symbolizer: The symbolizer used to symbolize stack. Returns: A list containing an instance of InstrumentationTestResult for each test parsed. """ results = [] # Results from synthetic ClassName#null tests, which occur from exceptions in # @BeforeClass / @AfterClass. class_failure_results = [] def add_result(result): if result.GetName().endswith('#null'): if result.GetType() == base_test_result.ResultType.UNKNOWN: # Not sure how this happens. http://crbug.com/397924393 # It presumably means two START events happen in a row. pass else: assert result.GetType() == base_test_result.ResultType.FAIL, ( 'test result of %r should be %r, but got %r' % ( result.GetName(), base_test_result.ResultType.FAIL, result.GetType(), ) ) class_failure_results.append(result) else: results.append(result) current_result = None cumulative_duration = 0 for status_code, bundle in statuses: # If the last test was a failure already, don't override that failure with # post-test failures that could be caused by the original failure. if ( status_code == instrumentation_parser.STATUS_CODE_BATCH_FAILURE and current_result.GetType() != base_test_result.ResultType.FAIL ): current_result.SetType(base_test_result.ResultType.FAIL) _MaybeSetLog(bundle, current_result, symbolizer, device_abi) continue if status_code == instrumentation_parser.STATUS_CODE_TEST_DURATION: # For the first result, duration will be set below to the difference # between the reported and actual durations to account for overhead like # starting instrumentation. if results: current_duration = int( bundle.get(_BUNDLE_DURATION_ID, duration_ms) ) current_result.SetDuration(current_duration) cumulative_duration += current_duration continue test_class = bundle.get(_BUNDLE_CLASS_ID, '') test_method = bundle.get(_BUNDLE_TEST_ID, '') if test_class and test_method: test_name = '%s#%s' % (test_class, test_method) else: continue if status_code == instrumentation_parser.STATUS_CODE_START: if current_result: add_result(current_result) current_result = test_result.InstrumentationTestResult( test_name, base_test_result.ResultType.UNKNOWN, duration_ms ) else: if status_code == instrumentation_parser.STATUS_CODE_OK: if ( current_result.GetType() == base_test_result.ResultType.UNKNOWN ): current_result.SetType(base_test_result.ResultType.PASS) elif status_code == instrumentation_parser.STATUS_CODE_SKIP: current_result.SetType(base_test_result.ResultType.SKIP) elif ( status_code == instrumentation_parser.STATUS_CODE_ASSUMPTION_FAILURE ): current_result.SetType(base_test_result.ResultType.SKIP) else: if status_code not in ( instrumentation_parser.STATUS_CODE_ERROR, instrumentation_parser.STATUS_CODE_FAILURE, ): logging.error( 'Unrecognized status code %d. Handling as an error.', status_code, ) current_result.SetType(base_test_result.ResultType.FAIL) _MaybeSetLog(bundle, current_result, symbolizer, device_abi) if current_result: if current_result.GetType() == base_test_result.ResultType.UNKNOWN: crashed = result_code == _ACTIVITY_RESULT_CANCELED and any( _NATIVE_CRASH_RE.search(l) for l in result_bundle.values() ) if crashed: current_result.SetType(base_test_result.ResultType.CRASH) add_result(current_result) if results: logging.info( 'Adding cumulative overhead to test %s: %dms', results[0].GetName(), duration_ms - cumulative_duration, ) results[0].SetDuration(duration_ms - cumulative_duration) # Copy failures from @BeforeClass / @AfterClass into all tests that are # marked as passing. for class_result in class_failure_results: prefix = class_result.GetName()[: -len('null')] for result in results: if ( result.GetName().startswith(prefix) and result.GetType() == base_test_result.ResultType.PASS ): result.SetType(base_test_result.ResultType.FAIL) result.SetLog(class_result.GetLog()) result.SetFailureReason(class_result.GetFailureReason()) return results def _MaybeSetLog(bundle, current_result, symbolizer, device_abi): if _BUNDLE_STACK_ID in bundle: stack = bundle[_BUNDLE_STACK_ID] if symbolizer and device_abi: current_result.SetLog( '%s\n%s' % ( stack, '\n'.join( symbolizer.ExtractAndResolveNativeStackTraces( stack, device_abi ) ), ) ) else: current_result.SetLog(stack) parsed_failure_reason = _ParseExceptionMessage(stack) if parsed_failure_reason not in _BANNED_FAILURE_REASONS: current_result.SetFailureReason(parsed_failure_reason) def _ParseExceptionMessage(stack): """Extracts the exception message from the given stack trace.""" # This interprets stack traces reported via InstrumentationResultPrinter: # https://source.chromium.org/chromium/chromium/src/+/main:third_party/android_support_test_runner/runner/src/main/java/android/support/test/internal/runner/listener/InstrumentationResultPrinter.java;l=181?q=InstrumentationResultPrinter&type=cs # This is a standard Java stack trace, of the form: # # at SomeClass.SomeMethod(...) # at ... lines = stack.split('\n') for i, line in enumerate(lines): if line.startswith('\tat'): return '\n'.join(lines[0:i]) # No call stack found, so assume everything is the exception message. return stack def FilterTests( tests, filter_strs=None, annotations=None, excluded_annotations=None ): """Filter a list of tests Args: tests: a list of tests. e.g. [ {'annotations": {}, 'class': 'com.example.TestA', 'method':'test1'}, {'annotations": {}, 'class': 'com.example.TestB', 'method':'test2'}] filter_strs: list of googletest-style filter string. annotations: a dict of wanted annotations for test methods. excluded_annotations: a dict of annotations to exclude. Return: A list of filtered tests """ def test_names_from_pattern(combined_pattern, test_names): patterns = combined_pattern.split(':') hashable_patterns = set() filename_patterns = [] for pattern in patterns: if '*' in pattern or '?' in pattern or '[' in pattern: filename_patterns.append(pattern) else: hashable_patterns.add(pattern) filter_test_names = ( set( unittest_util.FilterTestNames( test_names, ':'.join(filename_patterns) ) ) if len(filename_patterns) > 0 else set() ) for test_name in test_names: if test_name in hashable_patterns: filter_test_names.add(test_name) return filter_test_names def get_test_names(test): test_names = set() # Allow fully-qualified name as well as an omitted package. unqualified_class_test = { 'class': test['class'].split('.')[-1], 'method': test['method'], } test_name = GetTestName(test, sep='.') test_names.add(test_name) unqualified_class_test_name = GetTestName( unqualified_class_test, sep='.' ) test_names.add(unqualified_class_test_name) unique_test_name = GetUniqueTestName(test, sep='.') test_names.add(unique_test_name) junit4_test_name = GetTestNameWithoutParameterSuffix(test, sep='.') test_names.add(junit4_test_name) unqualified_junit4_test_name = GetTestNameWithoutParameterSuffix( unqualified_class_test, sep='.' ) test_names.add(unqualified_junit4_test_name) return test_names def get_tests_from_names(tests, test_names, tests_to_names): '''Returns the tests for which the given names apply Args: tests: a list of tests. e.g. [ {'annotations": {}, 'class': 'com.example.TestA', 'method':'test1'}, {'annotations": {}, 'class': 'com.example.TestB', 'method':'test2'}] test_names: a collection of names determining tests to return. Return: A list of tests that match the given test names ''' filtered_tests = [] for t in tests: current_test_names = tests_to_names[id(t)] for current_test_name in current_test_names: if current_test_name in test_names: filtered_tests.append(t) break return filtered_tests def remove_tests_from_names(tests, remove_test_names, tests_to_names): '''Returns the tests from the given list with given names removed Args: tests: a list of tests. e.g. [ {'annotations": {}, 'class': 'com.example.TestA', 'method':'test1'}, {'annotations": {}, 'class': 'com.example.TestB', 'method':'test2'}] remove_test_names: a collection of names determining tests to remove. tests_to_names: a dcitionary of test ids to a collection of applicable names for that test Return: A list of tests that don't match the given test names ''' filtered_tests = [] for t in tests: for name in tests_to_names[id(t)]: if name in remove_test_names: break else: filtered_tests.append(t) return filtered_tests def gtests_filter(tests, combined_filters): '''Returns the tests after the combined_filters have been applied Args: tests: a list of tests. e.g. [ {'annotations": {}, 'class': 'com.example.TestA', 'method':'test1'}, {'annotations": {}, 'class': 'com.example.TestB', 'method':'test2'}] combined_filters: the filter string representing tests to exclude Return: A list of tests that should still be included after the combined_filters are applied to their names ''' if not combined_filters: return tests # Collect all test names all_test_names = set() tests_to_names = {} for t in tests: tests_to_names[id(t)] = get_test_names(t) for name in tests_to_names[id(t)]: all_test_names.add(name) for combined_filter in combined_filters: pattern_groups = combined_filter.split('-') negative_pattern = ( pattern_groups[1] if len(pattern_groups) > 1 else None ) positive_pattern = pattern_groups[0] if positive_pattern: # Only use the test names that match the positive pattern positive_test_names = test_names_from_pattern( positive_pattern, all_test_names ) tests = get_tests_from_names( tests, positive_test_names, tests_to_names ) if negative_pattern: # Remove any test the negative filter matches remove_names = test_names_from_pattern( negative_pattern, all_test_names ) tests = remove_tests_from_names( tests, remove_names, tests_to_names ) return tests def annotation_filter(all_annotations): if not annotations: return True return any_annotation_matches(annotations, all_annotations) def excluded_annotation_filter(all_annotations): if not excluded_annotations: return True return not any_annotation_matches(excluded_annotations, all_annotations) def any_annotation_matches(filter_annotations, all_annotations): return any( ak in all_annotations and annotation_value_matches(av, all_annotations[ak]) for ak, av in filter_annotations ) def annotation_value_matches(filter_av, av): if filter_av is None: return True if isinstance(av, dict): tav_from_dict = av['value'] # If tav_from_dict is an int, the 'in' operator breaks, so convert # filter_av and manually compare. See https://crbug.com/1019707 if isinstance(tav_from_dict, int): return int(filter_av) == tav_from_dict return filter_av in tav_from_dict if isinstance(av, list): return filter_av in av return filter_av == av return_tests = [] for t in gtests_filter(tests, filter_strs): # Enforce that all tests declare their size. if not any(a in _VALID_ANNOTATIONS for a in t['annotations']): raise MissingSizeAnnotationError(GetTestName(t)) if not annotation_filter( t['annotations'] ) or not excluded_annotation_filter(t['annotations']): continue return_tests.append(t) return return_tests def GetTestsFromDexdump(test_apk): dex_dumps = dexdump.Dump(test_apk) tests = [] def get_test_methods(methods, annotations): test_methods = [] for method in methods: if method.startswith('test'): method_annotations = annotations.get(method, {}) # Dexdump used to not return any annotation info # So MediumTest annotation was added to all methods # Preserving this behaviour by adding MediumTest if none of the # size annotations are included in these annotations if not any( valid in method_annotations for valid in _VALID_ANNOTATIONS ): method_annotations.update({'MediumTest': None}) test_methods.append( {'method': method, 'annotations': method_annotations} ) return test_methods for dump in dex_dumps: for package_name, package_info in dump.items(): for class_name, class_info in package_info['classes'].items(): if ( class_name.endswith('Test') and not class_info['is_abstract'] ): classAnnotations, methodsAnnotations = class_info[ 'annotations' ] tests.append( { 'class': '%s.%s' % (package_name, class_name), 'annotations': classAnnotations, 'methods': get_test_methods( class_info['methods'], methodsAnnotations ), } ) return tests def GetTestName(test, sep='#'): """Gets the name of the given test. Note that this may return the same name for more than one test, e.g. if a test is being run multiple times with different parameters. Args: test: the instrumentation test dict. sep: the character(s) that should join the class name and the method name. Returns: The test name as a string. """ test_name = '%s%s%s' % (test['class'], sep, test['method']) assert not any(char in test_name for char in ' *-:'), ( 'The test name must not contain any of the characters in " *-:". See ' 'https://crbug.com/912199' ) return test_name def GetTestNameWithoutParameterSuffix(test, sep='#', parameterization_sep='__'): """Gets the name of the given JUnit4 test without parameter suffix. For most WebView JUnit4 javatests, each test is parameterizatized with "__sandboxed_mode" to run in both non-sandboxed mode and sandboxed mode. This function returns the name of the test without parameterization so test filters can match both parameterized and non-parameterized tests. Args: test: the instrumentation test dict. sep: the character(s) that should join the class name and the method name. parameterization_sep: the character(s) that separate method name and method parameterization suffix. Returns: The test name without parameter suffix as a string. """ name = GetTestName(test, sep=sep) return name.split(parameterization_sep)[0] def GetUniqueTestName(test, sep='#'): """Gets the unique name of the given test. This will include text to disambiguate between tests for which GetTestName would return the same name. Args: test: the instrumentation test dict. sep: the character(s) that should join the class name and the method name. Returns: The unique test name as a string. """ display_name = GetTestName(test, sep=sep) if test.get('flags', [None])[0]: sanitized_flags = [x.replace('-', '_') for x in test['flags']] display_name = '%s_with_%s' % (display_name, '_'.join(sanitized_flags)) assert not any(char in display_name for char in ' *-:'), ( 'The test name must not contain any of the characters in " *-:". See ' 'https://crbug.com/912199' ) return display_name class InstrumentationTestInstance(test_instance.TestInstance): def __init__(self, args, data_deps_delegate, error_func): super().__init__() self._additional_apks = [] self._additional_apexs = [] self._forced_queryable_additional_apks = [] self._instant_additional_apks = [] self._apk_under_test = None self._apk_under_test_incremental_install_json = None self._modules = None self._fake_modules = None self._additional_locales = None self._package_info = None self._suite = None self._test_apk = None self._test_apk_as_instant = False self._test_apk_incremental_install_json = None self._test_package = None self._junit4_runner_class = None self._uses_base_instrumentation = None self._has_chromium_test_listener = None self._use_native_coverage_listener = None self._test_support_apk = None self._initializeApkAttributes(args, error_func) self._data_deps = None self._data_deps_delegate = None self._runtime_deps_path = None self._variations_test_seed_path = args.variations_test_seed_path self._webview_variations_test_seed_path = ( args.webview_variations_test_seed_path ) self._store_data_dependencies_in_temp = False self._initializeDataDependencyAttributes(args, data_deps_delegate) self._annotations = None self._excluded_annotations = None self._has_external_annotation_filters = None self._test_filters = None self._initializeTestFilterAttributes(args) self._run_setup_commands = [] self._run_teardown_commands = [] self._initializeSetupTeardownCommandAttributes(args) self._flags = None self._use_apk_under_test_flags_file = False self._webview_flags = args.webview_command_line_arg self._initializeFlagAttributes(args) self._screenshot_dir = None self._timeout_scale = None self._wait_for_java_debugger = None self._initializeTestControlAttributes(args) self._coverage_directory = None self._initializeTestCoverageAttributes(args) self._store_tombstones = False self._symbolizer = None self._enable_breakpad_dump = False self._proguard_mapping_path = None self._deobfuscator = None self._initializeLogAttributes(args) self._replace_system_package = None self._initializeReplaceSystemPackageAttributes(args) self._system_packages_to_remove = None self._initializeSystemPackagesToRemoveAttributes(args) self._use_voice_interaction_service = None self._initializeUseVoiceInteractionService(args) self._use_webview_provider = None self._initializeUseWebviewProviderAttributes(args) self._skia_gold_properties = None self._initializeSkiaGoldAttributes(args) self._test_launcher_batch_limit = None self._initializeTestLauncherAttributes(args) self._approve_app_links_domain = None self._approve_app_links_package = None self._initializeApproveAppLinksAttributes(args) self._webview_process_mode = args.webview_process_mode self._webview_rebaseline_mode = args.webview_rebaseline_mode self._wpr_enable_record = args.wpr_enable_record self._external_shard_index = args.test_launcher_shard_index self._total_external_shards = args.test_launcher_total_shards self._is_unit_test = False self._initializeUnitTestFlag(args) self._run_disabled = args.run_disabled def _initializeApkAttributes(self, args, error_func): if args.apk_under_test: apk_under_test_path = args.apk_under_test if not args.apk_under_test.endswith( '.apk' ) and not args.apk_under_test.endswith('.apks'): apk_under_test_path = os.path.join( constants.GetOutDirectory(), constants.SDK_BUILD_APKS_DIR, '%s.apk' % args.apk_under_test, ) # TODO(jbudorick): Move the realpath up to the argument parser once # APK-by-name is no longer supported. apk_under_test_path = os.path.realpath(apk_under_test_path) if not os.path.exists(apk_under_test_path): error_func( 'Unable to find APK under test: %s' % apk_under_test_path ) self._apk_under_test = apk_helper.ToHelper(apk_under_test_path) test_apk_path = args.test_apk if not args.test_apk.endswith('.apk') and not args.test_apk.endswith( '.apks' ): test_apk_path = os.path.join( constants.GetOutDirectory(), constants.SDK_BUILD_APKS_DIR, '%s.apk' % args.test_apk, ) # TODO(jbudorick): Move the realpath up to the argument parser once # APK-by-name is no longer supported. test_apk_path = os.path.realpath(test_apk_path) if not os.path.exists(test_apk_path): error_func('Unable to find test APK: %s' % test_apk_path) self._test_apk = apk_helper.ToHelper(test_apk_path) self._suite = os.path.splitext(os.path.basename(args.test_apk))[0] self._test_apk_as_instant = args.test_apk_as_instant self._apk_under_test_incremental_install_json = ( args.apk_under_test_incremental_install_json ) self._test_apk_incremental_install_json = ( args.test_apk_incremental_install_json ) if self._test_apk_incremental_install_json: assert self._suite.endswith('_incremental') self._suite = self._suite[: -len('_incremental')] self._modules = args.modules self._fake_modules = args.fake_modules self._additional_locales = args.additional_locales self._test_support_apk = apk_helper.ToHelper( os.path.join( constants.GetOutDirectory(), constants.SDK_BUILD_TEST_JAVALIB_DIR, '%sSupport.apk' % self._suite, ) ) self._test_package = self._test_apk.GetPackageName() all_instrumentations = self._test_apk.GetAllInstrumentations() if len(all_instrumentations) > 1: logging.warning('This test apk has more than one instrumentation') self._junit4_runner_class = ( all_instrumentations[0]['android:name'] if all_instrumentations else None ) test_apk_metadata = dict(self._test_apk.GetAllMetadata()) self._has_chromium_test_listener = bool( test_apk_metadata.get('org.chromium.hasTestRunListener') ) self._use_native_coverage_listener = bool( test_apk_metadata.get('org.chromium.useNativeCoverageListener') ) if self._junit4_runner_class: if self._test_apk_incremental_install_json: for name, value in test_apk_metadata.items(): if ( name.startswith('incremental-install-instrumentation-') and value == _BASE_INSTRUMENTATION_CLASS_NAME ): self._uses_base_instrumentation = True break else: self._uses_base_instrumentation = ( self._junit4_runner_class == _BASE_INSTRUMENTATION_CLASS_NAME ) self._package_info = None if self._apk_under_test: package_under_test = self._apk_under_test.GetPackageName() for package_info in constants.PACKAGE_INFO.values(): if package_under_test == package_info.package: self._package_info = package_info break if not self._package_info: logging.warning( 'Unable to find package info for %s. ' '(This may just mean that the test package is ' 'currently being installed.)', self._test_package, ) for x in set( args.additional_apks + args.forced_queryable_additional_apks + args.instant_additional_apks ): if not os.path.exists(x): error_func('Unable to find additional APK: %s' % x) apk = apk_helper.ToHelper(x) self._additional_apks.append(apk) if x in args.forced_queryable_additional_apks: self._forced_queryable_additional_apks.append(apk) if x in args.instant_additional_apks: self._instant_additional_apks.append(apk) self._additional_apexs = args.additional_apexs def _initializeDataDependencyAttributes(self, args, data_deps_delegate): self._data_deps = [] self._data_deps_delegate = data_deps_delegate self._store_data_dependencies_in_temp = ( args.store_data_dependencies_in_temp ) self._runtime_deps_path = args.runtime_deps_path self._device_data_filters = getattr(args, 'device_data_filters', []) if not self._runtime_deps_path: logging.warning('No data dependencies will be pushed.') def _initializeTestFilterAttributes(self, args): self._test_filters = test_filter.InitializeFiltersFromArgs(args) self._has_external_annotation_filters = bool( args.annotation_str or args.exclude_annotation_str ) def annotation_element(a): a = a.split('=', 1) return (a[0], a[1] if len(a) == 2 else None) if args.annotation_str: self._annotations = [ annotation_element(a) for a in args.annotation_str.split(',') ] elif not self._test_filters: self._annotations = [ annotation_element(a) for a in _DEFAULT_ANNOTATIONS ] else: self._annotations = [] if args.exclude_annotation_str: self._excluded_annotations = [ annotation_element(a) for a in args.exclude_annotation_str.split(',') ] else: self._excluded_annotations = [] requested_annotations = set(a[0] for a in self._annotations) if args.run_disabled: self._excluded_annotations.extend( annotation_element(a) for a in _DO_NOT_REVIVE_ANNOTATIONS if a not in requested_annotations ) else: self._excluded_annotations.extend( annotation_element(a) for a in _EXCLUDE_UNLESS_REQUESTED_ANNOTATIONS if a not in requested_annotations ) def _initializeSetupTeardownCommandAttributes(self, args): self._run_setup_commands = args.run_setup_commands self._run_teardown_commands = args.run_teardown_commands def _initializeFlagAttributes(self, args): self._use_apk_under_test_flags_file = args.use_apk_under_test_flags_file self._flags = ['--enable-test-intents'] if args.command_line_flags: self._flags.extend(args.command_line_flags) if args.device_flags_file: with open(args.device_flags_file) as device_flags_file: stripped_lines = (l.strip() for l in device_flags_file) self._flags.extend(flag for flag in stripped_lines if flag) if ( args.strict_mode and args.strict_mode != 'off' and ( # TODO(yliuyliu): Turn on strict mode for coverage once # crbug/1006397 is fixed. not args.coverage_dir ) ): self._flags.append('--strict-mode=' + args.strict_mode) def _initializeTestControlAttributes(self, args): self._screenshot_dir = args.screenshot_dir self._timeout_scale = args.timeout_scale or 1 self._wait_for_java_debugger = args.wait_for_java_debugger def _initializeTestCoverageAttributes(self, args): self._coverage_directory = args.coverage_dir def _initializeLogAttributes(self, args): self._enable_breakpad_dump = args.enable_breakpad_dump self._proguard_mapping_path = args.proguard_mapping_path self._store_tombstones = args.store_tombstones self._symbolizer = stack_symbolizer.Symbolizer( self.apk_under_test.path if self.apk_under_test else None ) def _initializeReplaceSystemPackageAttributes(self, args): if ( not hasattr(args, 'replace_system_package') or not args.replace_system_package ): return self._replace_system_package = args.replace_system_package def _initializeSystemPackagesToRemoveAttributes(self, args): if ( not hasattr(args, 'system_packages_to_remove') or not args.system_packages_to_remove ): return self._system_packages_to_remove = args.system_packages_to_remove def _initializeUseVoiceInteractionService(self, args): if ( not hasattr(args, 'use_voice_interaction_service') or not args.use_voice_interaction_service ): return self._use_voice_interaction_service = args.use_voice_interaction_service def _initializeUseWebviewProviderAttributes(self, args): if ( not hasattr(args, 'use_webview_provider') or not args.use_webview_provider ): return self._use_webview_provider = args.use_webview_provider def _initializeSkiaGoldAttributes(self, args): self._skia_gold_properties = gold_utils.AndroidSkiaGoldProperties(args) def _initializeTestLauncherAttributes(self, args): if hasattr(args, 'test_launcher_batch_limit'): self._test_launcher_batch_limit = args.test_launcher_batch_limit def _initializeApproveAppLinksAttributes(self, args): if not hasattr(args, 'approve_app_links') or not args.approve_app_links: return # The argument will be formatted as com.android.thing:www.example.com . app_links = args.approve_app_links.split(':') if len(app_links) != 2 or not app_links[0] or not app_links[1]: logging.warning( '--approve_app_links option provided, but malformed.' ) return self._approve_app_links_package = app_links[0] self._approve_app_links_domain = app_links[1] def _initializeUnitTestFlag(self, args): self._is_unit_test = args.is_unit_test @property def additional_apks(self): return self._additional_apks @property def additional_apexs(self): return self._additional_apexs @property def apk_under_test(self): return self._apk_under_test @property def apk_under_test_incremental_install_json(self): return self._apk_under_test_incremental_install_json @property def approve_app_links_package(self): return self._approve_app_links_package @property def approve_app_links_domain(self): return self._approve_app_links_domain @property def modules(self): return self._modules @property def fake_modules(self): return self._fake_modules @property def additional_locales(self): return self._additional_locales @property def coverage_directory(self): return self._coverage_directory @property def enable_breakpad_dump(self): return self._enable_breakpad_dump @property def external_shard_index(self): return self._external_shard_index @property def flags(self): return self._flags[:] @property def is_unit_test(self): return self._is_unit_test @property def junit4_runner_class(self): return self._junit4_runner_class @property def has_chromium_test_listener(self): return self._has_chromium_test_listener @property def has_external_annotation_filters(self): return self._has_external_annotation_filters @property def uses_base_instrumentation(self): return self._uses_base_instrumentation @property def use_native_coverage_listener(self): return self._use_native_coverage_listener @property def package_info(self): return self._package_info @property def replace_system_package(self): return self._replace_system_package @property def run_setup_commands(self): return self._run_setup_commands @property def run_teardown_commands(self): return self._run_teardown_commands @property def use_voice_interaction_service(self): return self._use_voice_interaction_service @property def use_webview_provider(self): return self._use_webview_provider @property def webview_flags(self): return self._webview_flags[:] @property def screenshot_dir(self): return self._screenshot_dir @property def skia_gold_properties(self): return self._skia_gold_properties @property def store_data_dependencies_in_temp(self): return self._store_data_dependencies_in_temp @property def store_tombstones(self): return self._store_tombstones @property def suite(self): return self._suite @property def symbolizer(self): return self._symbolizer @property def system_packages_to_remove(self): return self._system_packages_to_remove @property def target_package(self): return self._test_apk.GetAllInstrumentations()[0][ 'android:targetPackage' ] @property def test_apk(self): return self._test_apk @property def test_apk_as_instant(self): return self._test_apk_as_instant @property def test_apk_incremental_install_json(self): return self._test_apk_incremental_install_json @property def test_filters(self): return self._test_filters @property def test_launcher_batch_limit(self): return self._test_launcher_batch_limit @property def test_support_apk(self): return self._test_support_apk @property def test_package(self): return self._test_package @property def timeout_scale(self): return self._timeout_scale @property def total_external_shards(self): return self._total_external_shards @property def use_apk_under_test_flags_file(self): return self._use_apk_under_test_flags_file @property def variations_test_seed_path(self): return self._variations_test_seed_path @property def webview_variations_test_seed_path(self): return self._webview_variations_test_seed_path @property def wait_for_java_debugger(self): return self._wait_for_java_debugger @property def wpr_record_mode(self): return self._wpr_enable_record @property def webview_process_mode(self): return self._webview_process_mode @property def webview_rebaseline_mode(self): return self._webview_rebaseline_mode @property def wpr_replay_mode(self): return not self._wpr_enable_record # override def TestType(self): return 'instrumentation' # override def GetPreferredAbis(self): # We could alternatively take the intersection of what they all support, # but it should never be the case that they support different things. apks = [self._test_apk, self._apk_under_test] + self._additional_apks for apk in apks: if apk: ret = apk.GetAbis() if ret: return ret return [] # override def SetUp(self): if self.wait_for_java_debugger and not self._test_apk.GetIsDebuggable(): raise Exception( 'Passed --wait-for-java-debugger flag but did not set ' 'debuggable_apks = true in GN args' ) deps = self._data_deps_delegate( self._runtime_deps_path, device_data_filters=self._device_data_filters, ) self._data_deps.extend(deps) if self._proguard_mapping_path: self._deobfuscator = deobfuscator.DeobfuscatorPool( self._proguard_mapping_path ) def GetDataDependencies(self): return self._data_deps def GetRunDisabledFlag(self): return self._run_disabled def GetLogcatPackageNames(self): ret = {self.target_package, self._test_package} if self._apk_under_test: ret.add(self._apk_under_test.GetPackageName()) ret.update(x.GetPackageName() for x in self._additional_apks) return ','.join(sorted(ret)) def MaybeDeobfuscateLines(self, lines): if not self._deobfuscator: return lines return self._deobfuscator.TransformLines(lines) def ProcessRawTests(self, raw_tests): inflated_tests = self._ParameterizeTestsWithFlags( self._InflateTests(raw_tests) ) filtered_tests = FilterTests( inflated_tests, self._test_filters, self._annotations, self._excluded_annotations, ) if self._test_filters and not filtered_tests: for t in inflated_tests: logging.debug(' %s', GetUniqueTestName(t)) logging.warning('Unmatched Filters: %s', self._test_filters) filtered_tests_with_excluded = FilterTests( inflated_tests, self._test_filters, self._annotations, [] ) num_disabled = len(filtered_tests_with_excluded) - len(filtered_tests) if num_disabled: test_run.ShowDisabledTestsHint(count=num_disabled) return filtered_tests def IsApkForceQueryable(self, apk): return apk in self._forced_queryable_additional_apks def IsApkInstant(self, apk): return apk in self._instant_additional_apks # pylint: disable=no-self-use def _InflateTests(self, tests): inflated_tests = [] for clazz in tests: for method in clazz['methods']: annotations = dict(clazz['annotations']) annotations.update(method['annotations']) # Preserve historic default. if not self._uses_base_instrumentation and not any( a in _VALID_ANNOTATIONS for a in annotations ): annotations['MediumTest'] = {} inflated_tests.append( { 'class': clazz['class'], 'method': method['method'], 'annotations': annotations, } ) return inflated_tests def _ParameterizeTestsWithFlags(self, tests): def _checkParameterization(annotations): types = [ _PARAMETERIZED_COMMAND_LINE_FLAGS_SWITCHES, _PARAMETERIZED_COMMAND_LINE_FLAGS, ] if types[0] in annotations and types[1] in annotations: raise CommandLineParameterizationException( 'Multiple command-line parameterization types: {}.'.format( ', '.join(types) ) ) def _switchesToFlags(switches): return ['--{}'.format(s) for s in switches if s] def _annotationToSwitches(clazz, methods): if clazz == _PARAMETERIZED_COMMAND_LINE_FLAGS_SWITCHES: return [methods['value']] if clazz == _PARAMETERIZED_COMMAND_LINE_FLAGS: list_of_switches = [] for annotation in methods['value']: for c, m in annotation.items(): list_of_switches += _annotationToSwitches(c, m) return list_of_switches return [] def _setTestFlags(test, flags): if flags: test['flags'] = flags elif 'flags' in test: del test['flags'] new_tests = [] for t in tests: annotations = t['annotations'] list_of_switches = [] _checkParameterization(annotations) if _SKIP_PARAMETERIZATION not in annotations: for clazz, methods in annotations.items(): list_of_switches += _annotationToSwitches(clazz, methods) if list_of_switches: _setTestFlags(t, _switchesToFlags(list_of_switches[0])) for p in list_of_switches[1:]: parameterized_t = copy.copy(t) _setTestFlags(parameterized_t, _switchesToFlags(p)) new_tests.append(parameterized_t) return tests + new_tests @staticmethod def GenerateTestResults( result_code, result_bundle, statuses, duration_ms, device_abi, symbolizer, ): return GenerateTestResults( result_code, result_bundle, statuses, duration_ms, device_abi, symbolizer, ) # override def TearDown(self): self.symbolizer.CleanUp() if self._deobfuscator: self._deobfuscator.Close() self._deobfuscator = None