# # Copyright (C) 2015 The Android Open Source Project # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # from __future__ import print_function import difflib import filecmp import glob import imp import multiprocessing import os import posixpath import re import shutil import subprocess import adb import ndk import util # pylint: disable=no-self-use def _get_jobs_arg(): return '-j{}'.format(multiprocessing.cpu_count() * 2) def _make_subtest_name(test, case): return '.'.join([test, case]) def _scan_test_suite(suite_dir, test_class, *args): tests = [] for dentry in os.listdir(suite_dir): path = os.path.join(suite_dir, dentry) if os.path.isdir(path): tests.append(test_class.from_dir(path, *args)) return tests class TestRunner(object): def __init__(self): self.tests = {} def add_suite(self, name, path, test_class, *args): if name in self.tests: raise KeyError('suite {} already exists'.format(name)) self.tests[name] = _scan_test_suite(path, test_class, *args) def _fixup_expected_failure(self, result, config, bug): if isinstance(result, Failure): return ExpectedFailure(result.test_name, config, bug) elif isinstance(result, Success): return UnexpectedSuccess(result.test_name, config, bug) else: # Skipped, UnexpectedSuccess, or ExpectedFailure. return result def _run_test(self, test, out_dir, test_filters): if not test_filters.filter(test.name): return [] config = test.check_unsupported() if config is not None: message = 'test unsupported for {}'.format(config) return [Skipped(test.name, message)] results = test.run(out_dir, test_filters) config, bug = test.check_broken() if config is None: return results # We need to check each individual test case for pass/fail and change # it to either an ExpectedFailure or an UnexpectedSuccess as necessary. return [self._fixup_expected_failure(r, config, bug) for r in results] def run(self, out_dir, test_filters): results = {suite: [] for suite in self.tests.keys()} for suite, tests in self.tests.items(): test_results = [] for test in tests: test_results.extend(self._run_test(test, out_dir, test_filters)) results[suite] = test_results return results def _maybe_color(text, color, do_color): return util.color_string(text, color) if do_color else text class TestResult(object): def __init__(self, test_name): self.test_name = test_name def __repr__(self): return self.to_string(colored=False) def passed(self): raise NotImplementedError def failed(self): raise NotImplementedError def to_string(self, colored=False): raise NotImplementedError class Failure(TestResult): def __init__(self, test_name, message): super(Failure, self).__init__(test_name) self.message = message def passed(self): return False def failed(self): return True def to_string(self, colored=False): label = _maybe_color('FAIL', 'red', colored) return '{} {}: {}'.format(label, self.test_name, self.message) class Success(TestResult): def passed(self): return True def failed(self): return False def to_string(self, colored=False): label = _maybe_color('PASS', 'green', colored) return '{} {}'.format(label, self.test_name) class Skipped(TestResult): def __init__(self, test_name, reason): super(Skipped, self).__init__(test_name) self.reason = reason def passed(self): return False def failed(self): return False def to_string(self, colored=False): label = _maybe_color('SKIP', 'yellow', colored) return '{} {}: {}'.format(label, self.test_name, self.reason) class ExpectedFailure(TestResult): def __init__(self, test_name, config, bug): super(ExpectedFailure, self).__init__(test_name) self.config = config self.bug = bug def passed(self): return True def failed(self): return False def to_string(self, colored=False): label = _maybe_color('KNOWN FAIL', 'yellow', colored) return '{} {}: known failure for {} ({})'.format( label, self.test_name, self.config, self.bug) class UnexpectedSuccess(TestResult): def __init__(self, test_name, config, bug): super(UnexpectedSuccess, self).__init__(test_name) self.config = config self.bug = bug def passed(self): return False def failed(self): return True def to_string(self, colored=False): label = _maybe_color('SHOULD FAIL', 'red', colored) return '{} {}: unexpected success for {} ({})'.format( label, self.test_name, self.config, self.bug) class Test(object): def __init__(self, name, test_dir): self.name = name self.test_dir = test_dir self.config = self.get_test_config() def get_test_config(self): return TestConfig.from_test_dir(self.test_dir) def run(self, out_dir, test_filters): raise NotImplementedError def check_broken(self): return self.config.match_broken(self.abi, self.platform, self.toolchain) def check_unsupported(self): return self.config.match_unsupported(self.abi, self.platform, self.toolchain) def check_subtest_broken(self, name): return self.config.match_broken(self.abi, self.platform, self.toolchain, subtest=name) def check_subtest_unsupported(self, name): return self.config.match_unsupported(self.abi, self.platform, self.toolchain, subtest=name) class AwkTest(Test): def __init__(self, name, test_dir, script): super(AwkTest, self).__init__(name, test_dir) self.script = script @classmethod def from_dir(cls, test_dir): test_name = os.path.basename(test_dir) script_name = test_name + '.awk' script = os.path.join(ndk.NDK_ROOT, 'build/awk', script_name) if not os.path.isfile(script): msg = '{} missing test script: {}'.format(test_name, script) raise RuntimeError(msg) # Check that all of our test cases are valid. for test_case in glob.glob(os.path.join(test_dir, '*.in')): golden_path = re.sub(r'\.in$', '.out', test_case) if not os.path.isfile(golden_path): msg = '{} missing output: {}'.format(test_name, golden_path) raise RuntimeError(msg) return cls(test_name, test_dir, script) # Awk tests only run in a single configuration. Disabling them per ABI, # platform, or toolchain has no meaning. Stub out the checks. def check_broken(self): return None, None def check_unsupported(self): return None def run(self, out_dir, test_filters): results = [] for test_case in glob.glob(os.path.join(self.test_dir, '*.in')): golden_path = re.sub(r'\.in$', '.out', test_case) result = self.run_case(out_dir, test_case, golden_path, test_filters) if result is not None: results.append(result) return results def run_case(self, out_dir, test_case, golden_out_path, test_filters): case_name = os.path.splitext(os.path.basename(test_case))[0] name = _make_subtest_name(self.name, case_name) if not test_filters.filter(name): return None out_path = os.path.join(out_dir, os.path.basename(golden_out_path)) with open(test_case, 'r') as test_in, open(out_path, 'w') as out_file: awk_path = ndk.get_tool('awk') print('{} -f {} < {} > {}'.format( awk_path, self.script, test_case, out_path)) rc = subprocess.call([awk_path, '-f', self.script], stdin=test_in, stdout=out_file) if rc != 0: return Failure(name, 'awk failed') if filecmp.cmp(out_path, golden_out_path): return Success(name) else: with open(out_path) as out_file: out_lines = out_file.readlines() with open(golden_out_path) as golden_out_file: golden_lines = golden_out_file.readlines() diff = ''.join(difflib.unified_diff( golden_lines, out_lines, fromfile='expected', tofile='actual')) message = 'output does not match expected:\n\n' + diff return Failure(name, message) def _prep_build_dir(src_dir, out_dir): if os.path.exists(out_dir): shutil.rmtree(out_dir) shutil.copytree(src_dir, out_dir) class TestConfig(object): """Describes the status of a test. Each test directory can contain a "test_config.py" file that describes the configurations a test is not expected to pass for. Previously this information could be captured in one of two places: the Application.mk file, or a BROKEN_BUILD/BROKEN_RUN file. Application.mk was used to state that a test was only to be run for a specific platform version, specific toolchain, or a set of ABIs. Unfortunately Application.mk could only specify a single toolchain or platform, not a set. BROKEN_BUILD/BROKEN_RUN files were too general. An empty file meant the test should always be skipped regardless of configuration. Any change that would put a test in that situation should be reverted immediately. These also didn't make it clear if the test was actually broken (and thus should be fixed) or just not applicable. A test_config.py file is more flexible. It is a Python module that defines at least one function by the same name as one in TestConfig.NullTestConfig. If a function is not defined the null implementation (not broken, supported), will be used. """ class NullTestConfig(object): def __init__(self): pass # pylint: disable=unused-argument @staticmethod def match_broken(abi, platform, toolchain, subtest=None): """Tests if a given configuration is known broken. A broken test is a known failing test that should be fixed. Any test with a non-empty broken section requires a "bug" entry with a link to either an internal bug (http://b/BUG_NUMBER) or a public bug (http://b.android.com/BUG_NUMBER). These tests will still be built and run. If the test succeeds, it will be reported as an error. Returns: A tuple of (broken_configuration, bug) or (None, None). """ return None, None @staticmethod def match_unsupported(abi, platform, toolchain, subtest=None): """Tests if a given configuration is unsupported. An unsupported test is a test that do not make sense to run for a given configuration. Testing x86 assembler on MIPS, for example. These tests will not be built or run. Returns: The string unsupported_configuration or None. """ return None # pylint: enable=unused-argument def __init__(self, file_path): # Note that this namespace isn't actually meaningful from our side; # it's only what the loaded module's __name__ gets set to. dirname = os.path.dirname(file_path) namespace = '.'.join([dirname, 'test_config']) try: self.module = imp.load_source(namespace, file_path) except IOError: self.module = None try: self.match_broken = self.module.match_broken except AttributeError: self.match_broken = self.NullTestConfig.match_broken try: self.match_unsupported = self.module.match_unsupported except AttributeError: self.match_unsupported = self.NullTestConfig.match_unsupported @classmethod def from_test_dir(cls, test_dir): path = os.path.join(test_dir, 'test_config.py') return cls(path) class DeviceTestConfig(TestConfig): """Specialization of test_config.py that includes device API level. We need to mark some tests as broken or unsupported based on what device they are running on, as opposed to just what they were built for. """ class NullTestConfig(object): def __init__(self): pass # pylint: disable=unused-argument @staticmethod def match_broken(abi, platform, device_platform, toolchain, subtest=None): return None, None @staticmethod def match_unsupported(abi, platform, device_platform, toolchain, subtest=None): return None # pylint: enable=unused-argument def _run_build_sh_test(test_name, build_dir, test_dir, build_flags, abi, platform, toolchain): _prep_build_dir(test_dir, build_dir) with util.cd(build_dir): build_cmd = ['sh', 'build.sh', _get_jobs_arg()] + build_flags test_env = dict(os.environ) if abi is not None: test_env['APP_ABI'] = abi if platform is not None: test_env['APP_PLATFORM'] = platform assert toolchain is not None test_env['NDK_TOOLCHAIN_VERSION'] = toolchain rc, out = util.call_output(build_cmd, env=test_env) if rc == 0: return Success(test_name) else: return Failure(test_name, out) def _run_ndk_build_test(test_name, build_dir, test_dir, build_flags, abi, platform, toolchain): _prep_build_dir(test_dir, build_dir) with util.cd(build_dir): args = [ 'APP_ABI=' + abi, 'NDK_TOOLCHAIN_VERSION=' + toolchain, _get_jobs_arg(), ] if platform is not None: args.append('APP_PLATFORM=' + platform) rc, out = ndk.build(build_flags + args) if rc == 0: return Success(test_name) else: return Failure(test_name, out) class PythonBuildTest(Test): """A test that is implemented by test.py. A test.py test has a test.py file in its root directory. This module contains a run_test function which returns a tuple of `(boolean_success, string_failure_message)` and takes the following kwargs (all of which default to None): abi: ABI to test as a string. platform: Platform to build against as a string. toolchain: Toolchain to use as a string. build_flags: Additional build flags that should be passed to ndk-build if invoked as a list of strings. """ def __init__(self, name, test_dir, abi, platform, toolchain, build_flags): super(PythonBuildTest, self).__init__(name, test_dir) self.abi = abi self.platform = platform self.toolchain = toolchain self.build_flags = build_flags def run(self, out_dir, _): build_dir = os.path.join(out_dir, self.name) print('Running build test: {}'.format(self.name)) _prep_build_dir(self.test_dir, build_dir) with util.cd(build_dir): module = imp.load_source('test', 'test.py') success, failure_message = module.run_test( abi=self.abi, platform=self.platform, toolchain=self.toolchain, build_flags=self.build_flags) if success: return [Success(self.name)] else: return [Failure(self.name, failure_message)] class ShellBuildTest(Test): def __init__(self, name, test_dir, abi, platform, toolchain, build_flags): super(ShellBuildTest, self).__init__(name, test_dir) self.abi = abi self.platform = platform self.toolchain = toolchain self.build_flags = build_flags def run(self, out_dir, _): build_dir = os.path.join(out_dir, self.name) print('Running build test: {}'.format(self.name)) if os.name == 'nt': reason = 'build.sh tests are not supported on Windows' return [Skipped(self.name, reason)] return [_run_build_sh_test(self.name, build_dir, self.test_dir, self.build_flags, self.abi, self.platform, self.toolchain)] class NdkBuildTest(Test): def __init__(self, name, test_dir, abi, platform, toolchain, build_flags): super(NdkBuildTest, self).__init__(name, test_dir) self.abi = abi self.platform = platform self.toolchain = toolchain self.build_flags = build_flags def run(self, out_dir, _): build_dir = os.path.join(out_dir, self.name) print('Running build test: {}'.format(self.name)) return [_run_ndk_build_test(self.name, build_dir, self.test_dir, self.build_flags, self.abi, self.platform, self.toolchain)] class BuildTest(object): @classmethod def from_dir(cls, test_dir, abi, platform, toolchain, build_flags): test_name = os.path.basename(test_dir) if os.path.isfile(os.path.join(test_dir, 'test.py')): return PythonBuildTest(test_name, test_dir, abi, platform, toolchain, build_flags) elif os.path.isfile(os.path.join(test_dir, 'build.sh')): return ShellBuildTest(test_name, test_dir, abi, platform, toolchain, build_flags) else: return NdkBuildTest(test_name, test_dir, abi, platform, toolchain, build_flags) def _copy_test_to_device(build_dir, device_dir, abi, test_filters, test_name): abi_dir = os.path.join(build_dir, 'libs', abi) if not os.path.isdir(abi_dir): raise RuntimeError('No libraries for {}'.format(abi)) test_cases = [] for test_file in os.listdir(abi_dir): if test_file in ('gdbserver', 'gdb.setup'): continue file_is_lib = False if not test_file.endswith('.so'): file_is_lib = True case_name = _make_subtest_name(test_name, test_file) if not test_filters.filter(case_name): continue test_cases.append(test_file) # TODO(danalbert): Libs with the same name will clobber each other. # This was the case with the old shell based script too. I'm trying not # to change too much in the translation. lib_path = os.path.join(abi_dir, test_file) print('\tPushing {} to {}...'.format(lib_path, device_dir)) adb.push(lib_path, device_dir) # Binaries pushed from Windows may not have execute permissions. if not file_is_lib: file_path = posixpath.join(device_dir, test_file) adb.shell('chmod +x ' + file_path) # TODO(danalbert): Sync data. # The libc++ tests contain a DATA file that lists test names and their # dependencies on file system data. These files need to be copied to # the device. if len(test_cases) == 0: raise RuntimeError('Could not find any test executables.') return test_cases class DeviceTest(Test): def __init__(self, name, test_dir, abi, platform, device_platform, toolchain, build_flags): super(DeviceTest, self).__init__(name, test_dir) self.abi = abi self.platform = platform self.device_platform = device_platform self.toolchain = toolchain self.build_flags = build_flags @classmethod def from_dir(cls, test_dir, abi, platform, device_platform, toolchain, build_flags): test_name = os.path.basename(test_dir) return cls(test_name, test_dir, abi, platform, device_platform, toolchain, build_flags) def get_test_config(self): return DeviceTestConfig.from_test_dir(self.test_dir) def check_broken(self): return self.config.match_broken(self.abi, self.platform, self.device_platform, self.toolchain) def check_unsupported(self): return self.config.match_unsupported(self.abi, self.platform, self.device_platform, self.toolchain) def check_subtest_broken(self, name): return self.config.match_broken(self.abi, self.platform, self.device_platform, self.toolchain, subtest=name) def check_subtest_unsupported(self, name): return self.config.match_unsupported(self.abi, self.platform, self.device_platform, self.toolchain, subtest=name) def run(self, out_dir, test_filters): print('Running device test: {}'.format(self.name)) build_dir = os.path.join(out_dir, self.name) build_result = _run_ndk_build_test(self.name, build_dir, self.test_dir, self.build_flags, self.abi, self.platform, self.toolchain) if not build_result.passed(): return [build_result] device_dir = posixpath.join('/data/local/tmp/ndk-tests', self.name) # We have to use `ls foo || mkdir foo` because Gingerbread was lacking # `mkdir -p`, the -d check for directory existence, stat, dirname, and # every other thing I could think of to implement this aside from ls. result, out = adb.shell('ls {0} || mkdir {0}'.format(device_dir)) if result != 0: raise RuntimeError('mkdir failed:\n' + '\n'.join(out)) results = [] try: test_cases = _copy_test_to_device( build_dir, device_dir, self.abi, test_filters, self.name) for case in test_cases: case_name = _make_subtest_name(self.name, case) if not test_filters.filter(case_name): continue config = self.check_subtest_unsupported(case) if config is not None: message = 'test unsupported for {}'.format(config) results.append(Skipped(case_name, message)) continue cmd = 'cd {} && LD_LIBRARY_PATH={} ./{}'.format( device_dir, device_dir, case) print('\tExecuting {}...'.format(case_name)) result, out = adb.shell(cmd) config, bug = self.check_subtest_broken(case) if config is None: if result == 0: results.append(Success(case_name)) else: results.append(Failure(case_name, '\n'.join(out))) else: if result == 0: results.append(UnexpectedSuccess(case_name, config, bug)) else: results.append(ExpectedFailure(case_name, config, bug)) return results finally: adb.shell('rm -r {}'.format(device_dir))