#!/usr/bin/env python
# Copyright (C) 2016 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.
'''Main test suite execution script.'''
import argparse
import inspect
import logging
import os
import signal
import subprocess
import sys
import time
import collections
import xml.etree.ElementTree as ET
from config import Config
from tests.harness import util_constants
from tests.harness.exception import TestSuiteException, FailFastException
from tests.harness import UtilAndroid
from tests.harness import UtilBundle
from tests.harness import util_log
from tests.harness.util_functions import load_py_module
from tests.harness.decorators import deprecated
# For some reason pylint is not able to understand the class returned by
# from util_log.get_logger() and generates a lot of false warnings
#pylint: disable=maybe-no-member
EMU_PROC = None
def _parse_args():
'''Parse the command line arguments.
Returns:
A namespace object that contains the options specified to run_tests on
the command line.
'''
parser = argparse.ArgumentParser(description='Run the test suite.')
parser.add_argument('--config', '-c',
metavar='path',
help='Path to a custom config file.')
parser.add_argument('--device', '-d',
help='Specify the device id of the device to test on.')
parser.add_argument('--test', '-t',
metavar='path',
help='Specify a specific test to run.')
group = parser.add_mutually_exclusive_group()
group.add_argument('--wimpy', '-w',
action='store_true',
default=None,
help='Test only a core subset of features.')
group.add_argument('--app-types',
default=['java', 'cpp', 'jni'],
nargs='*',
help='Specify a list of Android app types against which'
' to run the tests',
dest='bundle_types')
parser.add_argument('--install-only',
action='store_true',
default=False,
help='It only runs the pre-run stage of the test suite.'
' It installs the required APKs but does not '
'execute the tests.',
dest='install_only')
parser.add_argument('--no-install', '-n',
action='store_true',
default=False,
help='Stop the test suite installing apks to device.',
dest='noinstall')
parser.add_argument('--no-uninstall',
action='store_true',
default=False,
help='Stop the test suite uninstalling apks after '
'completion.',
dest='nouninstall')
parser.add_argument('--print-to-stdout',
action='store_true',
default=False,
help='Print all logging information to standard out.',
dest='print_to_stdout')
parser.add_argument('--verbose', '-v',
action='store_true',
default=None,
help='Store extra info in the log.')
parser.add_argument('--fail-fast',
action='store_true',
default=False,
help='Exit the test suite immediately on the first failure.')
parser.add_argument('--run-emu',
action='store_true',
default=None,
help='Spawn an emulator and run the test suite on that.'
' Specify the emulator command line in the config'
' file or with -emu-cmd.',
dest='run_emu')
# Get the properties of the Config class and add a command line argument
# for each.
this_module = sys.modules[__name__]
for member_name, member_obj in inspect.getmembers(Config):
if (inspect.isdatadescriptor(member_obj) and
member_name not in ['__weakref__', 'device', 'verbose']):
# List type properties can take one or more arguments
num_args = None
if (isinstance(member_obj, property)
and isinstance(member_obj.fget(Config), list)):
num_args = '+'
opt_name = member_name.replace('_', '-')
setattr(this_module, opt_name, '')
parser.add_argument('--' + opt_name,
nargs=num_args,
help=member_obj.__doc__,
dest=member_name)
return parser.parse_args()
def _choice(first_choice, second_choice):
'''Return first_choice if it is not None otherwise return second_choice.
Args:
first_choice: The first choice value.
second_choice: The alternative value.
Returns:
The first argument if it is not None, and the second otherwise.
'''
return first_choice if first_choice else second_choice
class State(object):
'''This class manages all objects required by the test suite.'''
# pylint: disable=too-many-instance-attributes
# Since this is a state class many attributes are expected.
def __init__(self):
'''State constructor.
Raises:
TestSuiteException: When unable to load config file.
AssertionError: When assertions fail.
'''
# Parse the command line options
args = _parse_args()
# create a config instance
if args.config:
# use the user supplied
config = State.load_user_configuration(args.config)
else:
# use the default configuration
config = Config()
# save the test blacklist
self.blacklist = _choice(args.blacklist, config.blacklist)
# Allow any of the command line arguments to override the
# values in the config file.
self.adb_path = _choice(args.adb_path, config.adb_path)
self.host_port = int(_choice(args.host_port, config.host_port))
self.device = _choice(args.device, config.device)
self.user_specified_device = self.device
self.device_port = int(_choice(args.device_port, config.device_port))
self.lldb_server_path_device = _choice(args.lldb_server_path_device,
config.lldb_server_path_device)
self.lldb_server_path_host = _choice(args.lldb_server_path_host,
config.lldb_server_path_host)
self.aosp_product_path = _choice(args.aosp_product_path,
config.aosp_product_path)
self.log_file_path = _choice(args.log_file_path, config.log_file_path)
self.results_file_path = _choice(args.results_file_path,
config.results_file_path)
self.lldb_path = _choice(args.lldb_path, config.lldb_path)
self.print_to_stdout = args.print_to_stdout
self.verbose = _choice(args.verbose, config.verbose)
self.timeout = int(_choice(args.timeout, config.timeout))
self.emu_cmd = _choice(args.emu_cmd, config.emu_cmd)
self.run_emu = args.run_emu
self.wimpy = args.wimpy
self.bundle_types = args.bundle_types if not self.wimpy else ['java']
self.fail_fast = args.fail_fast
# validate the param "verbose"
if not isinstance(self.verbose, bool):
raise TestSuiteException('The parameter "verbose" should be a '
'boolean: {0}'.format(self.verbose))
# create result array
self.results = dict()
self.single_test = args.test
# initialise the logging facility
log_level = logging.INFO if not self.verbose else logging.DEBUG
util_log.initialise("driver",
print_to_stdout=self.print_to_stdout,
level=log_level,
file_mode='w', # open for write
file_path=self.log_file_path
)
log = util_log.get_logger()
if self.run_emu and not self.emu_cmd:
log.TestSuiteException('Need to specify --emu-cmd (or specify a'
' value in the config file) if using --run-emu.')
# create a results file
self.results_file = open(self.results_file_path, 'w')
# create an android helper object
self.android = UtilAndroid(self.adb_path,
self.lldb_server_path_device,
self.device)
assert self.android
# create a test bundle
self.bundle = UtilBundle(self.android,
self.aosp_product_path)
assert self.bundle
# save the no pushing option
assert isinstance(args.noinstall, bool)
self.noinstall = args.noinstall
assert isinstance(args.nouninstall, bool)
self.nouninstall = args.nouninstall
# install only option
assert type(args.install_only) is bool
self.install_only = args.install_only
if self.install_only:
log.log_and_print('Option --install-only set. The test APKs will '
'be installed on the device but the tests will '
'not be executed.')
if self.noinstall:
raise TestSuiteException('Conflicting options given: '
'--install-only and --no-install')
# TCP port modifier which is used to increment the port number used for
# each test case to avoid collisions.
self.port_mod = 0
# total number of test files that have been executed
self.test_count = 0
def get_android(self):
'''Return the android ADB helper instance.
Returns:
The android ADB helper, instance of UtilAndroid.
'''
assert self.android
return self.android
def get_bundle(self):
'''Return the test executable bundle.
Returns:
The test exectable collection, instance of UtilBundle.
'''
return self.bundle
def add_result(self, name, app_type, result):
'''Add a test result to the collection.
Args:
name: String name of the test that has executed.
app_type: type of app i.e. java, jni, or cpp
result: String result of the test, "pass", "fail", "error".
'''
key = (name, app_type)
assert key not in self.results
self.results[key] = result
def get_single_test(self):
'''Get the name of the single test to run.
Returns:
A string that is the name of the python file containing the test to
be run. If all tests are to be run this returns None.
'''
return self.single_test
@staticmethod
def load_user_configuration(path):
'''Load the test suite config from the give path.
Instantiate the Config class found in the module at the given path.
If no suitable class is available, it raises a TestSuiteException.
Args:
path: String location of the module.
Returns:
an instance of the Config class, defined in the module.
Raises:
TestSuiteException: when unable to import the module or when a
subclass of Config is not found inside it.
'''
# load the module
config_module = load_py_module(path)
if not config_module:
raise TestSuiteException('Unable to import the module from "%s"'
% (path))
# look for a subclass of Config
for name, value in inspect.getmembers(config_module):
if (inspect.isclass(value)
and name != 'Config'
and issubclass(value, Config)):
# that's our candidate
return value()
# otherwise there are no valid candidates
raise TestSuiteException('The provided user configuration is not '
'valid. The module must define a subclass '
'of Config')
def _kill_emulator():
''' Kill the emulator process. '''
global EMU_PROC
if EMU_PROC:
try:
EMU_PROC.terminate()
except OSError:
# can't kill a dead proc
log = util_log.get_logger()
log.debug('Trying to kill an emulator but it is already dead.')
def _check_emulator_terminated():
''' Throw an exception if the emulator process has ended.
Raises:
TestSuiteException: If the emulator process has ended.
'''
global EMU_PROC
assert EMU_PROC
if EMU_PROC.poll():
stdout, stderr = EMU_PROC.communicate()
raise TestSuiteException('The emulator terminated with output:'
'\nstderr: {0}\nstdout: {1}.'.format(stderr, stdout))
@deprecated()
def _launch_emulator(state):
'''Launch the emulator and wait for it to boot.
Args:
emu_cmd: The command line to run the emulator.
Raises:
TestSuiteException: If an emulator already exists or the emulator
process terminated before we could connect to it, or
we failed to copy lldb-server to the emulator.
'''
global EMU_PROC
android = state.android
if state.user_specified_device:
if android.device_with_substring_exists(state.user_specified_device):
raise TestSuiteException(
'A device with name {0} already exists.',
state.user_specified_device)
else:
if android.device_with_substring_exists('emulator'):
raise TestSuiteException('An emulator already exists.')
assert state.emu_cmd
EMU_PROC = subprocess.Popen(state.emu_cmd.split(),
stdout=None,
stderr=subprocess.STDOUT)
log = util_log.get_logger()
log.info('Launching emulator with command line {0}'.format(state.emu_cmd))
tries_number = 180
tries = tries_number
found_device = False
while not found_device:
try:
android.validate_device(False, 'emulator')
found_device = True
except TestSuiteException as ex:
tries -= 1
if tries == 0:
# Avoid infinitely looping if the emulator won't boot
log.warning(
'Giving up trying to validate device after {0} tries.'
.format(tries_number))
raise ex
_check_emulator_terminated()
# wait a bit and try again, maybe it has now booted
time.sleep(10)
tries = 500
while not android.is_booted():
tries -= 1
if tries == 0:
# Avoid infinitely looping if the emulator won't boot
raise TestSuiteException('The emulator has failed to boot.')
_check_emulator_terminated()
time.sleep(5)
# Need to be root before we can push lldb-server
android.adb_root()
android.wait_for_device()
# Push the lldb-server executable to the device.
output = android.adb('push {0} {1}'.format(state.lldb_server_path_host,
state.lldb_server_path_device))
if 'failed to copy' in output or 'No such file or directory' in output:
raise TestSuiteException(
'unable to push lldb-server to the emulator: {0}.'
.format(output))
output = android.shell('chmod a+x {0}'
.format(state.lldb_server_path_device))
if 'No such file or directory' in output:
raise TestSuiteException('Failed to copy lldb-server to the emulator.')
def _restart_emulator(state):
'''Kill the emulator and start a new instance.
Args:
state: Test suite state collection, instance of State.
'''
_kill_emulator()
_launch_emulator(state)
def _run_test(state, name, bundle_type):
'''Execute a single test case.
Args:
state: Test suite state collection, instance of State.
name: String file name of the test to execute.
bundle_type: string for the installed app type (cpp|jni|java)
Raises:
AssertionError: When assertion fails.
'''
assert isinstance(name, str)
try:
state.android.check_adb_alive()
except TestSuiteException as expt:
global EMU_PROC
if EMU_PROC:
_restart_emulator(state)
else:
raise expt
log = util_log.get_logger()
sys.stdout.write('Running {0}\r'.format(name))
sys.stdout.flush()
log.info('Running {0}'.format(name))
run_tests_dir = os.path.dirname(os.path.realpath(__file__))
run_test_path = os.path.join(run_tests_dir, 'tests', 'run_test.py')
# Forward port for lldb-server on the device to our host
hport = int(state.host_port) + state.port_mod
dport = int(state.device_port) + state.port_mod
state.android.forward_port(hport, dport)
state.port_mod += 1
log.debug('Giving up control to {0}...'.format(name))
params = map(str, [
sys.executable,
run_test_path,
name,
state.log_file_path,
state.adb_path,
state.lldb_server_path_device,
state.aosp_product_path,
dport,
state.android.get_device_id(),
state.print_to_stdout,
state.verbose,
state.wimpy,
state.timeout,
bundle_type
])
return_code = subprocess.call(params)
state.test_count += 1
state.android.remove_port_forwarding()
log.seek_to_end()
# report in sys.stdout the result
success = return_code == util_constants.RC_TEST_OK
status_handlers = collections.defaultdict(lambda: ('error', log.error), (
(util_constants.RC_TEST_OK, ('pass', log.info)),
(util_constants.RC_TEST_TIMEOUT, ('timeout', log.error)),
(util_constants.RC_TEST_IGNORED, ('ignored', log.info)),
(util_constants.RC_TEST_FAIL, ('fail', log.critical))
)
)
status_name, status_logger = status_handlers[return_code]
log.info('Running %s: %s', name, status_name.upper())
status_logger("Test %r: %s", name, status_name)
# Special case for ignored tests - just return now
if return_code == util_constants.RC_TEST_IGNORED:
return
state.add_result(name, bundle_type, status_name)
if state.fail_fast and not success:
raise FailFastException(name)
# print a running total pass rate
passes = sum(1 for key, value in state.results.items() if value == 'pass')
log.info('Current pass rate: %s of %s executed.', passes, len(state.results))
def _check_lldbserver_exists(state):
'''Check lldb-server exists on the target device and it is executable.
Raises:
TestSuiteError: If lldb-server does not exist on the target.
'''
assert state
message = 'Unable to verify valid lldb-server on target'
android = state.get_android()
assert android
cmd = state.lldb_server_path_device
out = android.shell(cmd, False)
if not isinstance(out, str):
raise TestSuiteException(message)
if out.find('Usage:') < 0:
raise TestSuiteException(message)
def _suite_pre_run(state):
'''This function is executed before the test cases are run (setup).
Args:
state: Test suite state collection, instance of State.
Return:
True if the pre_run step completes without error.
Checks made:
- Validating that adb exists and runs.
- Validating that a device is attached.
- We have root access to the device.
- All test binaries were pushed to the device.
- The port for lldb-server was forwarded correctly.
Raises:
AssertionError: When assertions fail.
'''
assert state
log = util_log.get_logger()
try:
android = state.get_android()
bundle = state.get_bundle()
assert android
assert bundle
# validate ADB helper class
android.validate_adb()
log.log_and_print('Located ADB')
if state.run_emu:
log.log_and_print('Launching emulator...')
_launch_emulator(state)
log.log_and_print('Started emulator ' + android.device)
else:
android.validate_device()
log.log_and_print('Located device ' + android.device)
if state.noinstall and not state.single_test:
bundle.check_apps_installed(state.wimpy)
# elevate to root user
android.adb_root()
android.wait_for_device()
# check that lldb-server exists on device
android.kill_servers()
_check_lldbserver_exists(state)
if not state.noinstall:
# push all tests to the device
log.log_and_print('Pushing all tests...')
bundle.push_all()
log.log_and_print('Pushed all tests')
log.log_and_print('Pre run complete')
except TestSuiteException as expt:
log.exception('Test suite pre run failure')
# Even if we are logging the error, it may be helpful and more
# immediate to find out the error into the terminal
log.log_and_print('ERROR: Unable to set up the test suite: %s\n'
% expt.message, logging.ERROR)
return False
return True
def _suite_post_run(state):
'''This function is executed after the test cases have run (teardown).
Args:
state: Test suite state collection, instance of State.
Returns:
Number of failures
'''
log = util_log.get_logger()
if not state.noinstall and not state.nouninstall:
if state.wimpy:
state.bundle.uninstall_all_apk()
else:
state.bundle.uninstall_all()
log.log_and_print('Uninstalled/Deleted all tests')
total = 0
passes = 0
failures = 0
results = ET.Element('testsuite')
results.attrib['name'] = 'LLDB RS Test Suite'
for key, value in state.results.items():
total += 1
if value == 'pass':
passes += 1
else:
failures += 1
# test case name, followed by pass, failure or error elements
testcase = ET.Element('testcase')
testcase.attrib['name'] = "%s:%s" % key
result_element = ET.Element(value)
result_element.text = "%s:%s" % key
testcase.append(result_element)
results.append(testcase)
assert passes + failures == total, 'Invalid test results status'
if failures:
log.log_and_print(
'The following failures occurred:\n%s\n' %
'\n'.join('failed: %s:%s' % test_spec
for test_spec, result in state.results.items() if result != 'pass'
))
log.log_and_print('{0} of {1} passed'.format(passes, total))
if total:
log.log_and_print('{0}% rate'.format((passes*100)/total))
results.attrib['tests'] = str(total)
state.results_file.write(ET.tostring(results, encoding='iso-8859-1'))
return failures
def _discover_tests(state):
'''Discover all tests in the tests directory.
Returns:
List of strings, test file names from the 'tests' directory.
'''
tests = []
single_test = state.get_single_test()
if single_test is None:
file_dir = os.path.dirname(os.path.realpath(__file__))
tests_dir = os.path.join(file_dir, 'tests')
for sub_dir in os.listdir(tests_dir):
current_test_dir = os.path.join(tests_dir, sub_dir)
if os.path.isdir(current_test_dir):
dir_name = os.path.basename(current_test_dir)
if dir_name == 'harness':
continue
for item in os.listdir(current_test_dir):
if (item.startswith('test')
and item.endswith('.py')
and not item in state.blacklist):
tests.append(item)
else:
if single_test.endswith('.py'):
tests.append(single_test)
else:
tests.append(single_test + '.py')
return tests
def _deduce_python_path(state):
'''Try to deduce the PYTHONPATH environment variable via the LLDB binary.
Args:
state: Test suite state collection, instance of State.
Returns:
True if PYTHONPATH has been updated, False otherwise.
Raises:
TestSuiteException: If lldb path provided in the config or command line
is incorrect.
AssertionError: If an assertion fails.
'''
lldb_path = state.lldb_path
if not lldb_path:
# lldb may not be provided in preference of a manual $PYTHONPATH
return False
params = [lldb_path, '-P']
try:
proc = subprocess.Popen(params, stdout=subprocess.PIPE)
except OSError as err:
error_string = 'Could not run lldb at %s: %s' % (lldb_path, str(err))
raise TestSuiteException(error_string)
stdout = proc.communicate()[0]
if stdout:
os.environ['PYTHONPATH'] = stdout.strip()
return True
return False
def main():
'''The lldb-renderscript test suite entry point.'''
log = None
try:
# parse the command line
state = State()
assert state
# logging is initialised in State()
log = util_log.get_logger()
# if we can, set PYTHONPATH for lldb bindings
if not _deduce_python_path(state):
log.log_and_print('Unable to deduce PYTHONPATH', logging.WARN)
# pre run step
if not _suite_pre_run(state):
raise TestSuiteException('Test suite pre-run step failed')
# discover all tests and execute them
tests = _discover_tests(state)
log.log_and_print('Found {0} tests'.format(len(tests)))
if state.install_only:
log.log_and_print('Test applications installed. Terminating due to '
'--install-only option')
else:
# run the tests
for bundle_type in state.bundle_types:
log.info("Running bundle type '%s'", bundle_type)
for item in tests:
_run_test(state, item, bundle_type)
# post run step
quit(0 if _suite_post_run(state) == 0 else 1)
except AssertionError:
if log:
log.exception('Internal test suite error')
print('Internal test suite error')
quit(1)
except FailFastException:
log.exception('Early exit after first test failure')
quit(1)
except TestSuiteException as error:
if log:
log.exception('Test suite exception')
print('{0}'.format(str(error)))
quit(2)
finally:
_kill_emulator()
logging.shutdown()
def signal_handler(_, _unused):
'''Signal handler for SIGINT, caused by the user typing Ctrl-C.'''
# pylint: disable=unused-argument
# pylint: disable=protected-access
print('Ctrl+C!')
os._exit(1)
# execution trampoline
if __name__ == '__main__':
signal.signal(signal.SIGINT, signal_handler)
main()