xref: /frameworks/rs/tests/lldb/run_tests.py

#!/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()