xref: /frameworks/rs/tests/lldb/tests/run_test.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.

'''This script will run one specific test.'''
from __future__ import print_function, absolute_import

import os
import sys
import atexit
import inspect
import logging
import argparse
import warnings

import harness
from harness import util_constants
from harness import util_log
from harness import util_warnings
from harness.util_functions import load_py_module
from harness.util_lldb import UtilLLDB
from harness.exception import DisconnectedException
from harness.exception import TestSuiteException, TestIgnoredException
from harness.util_timer import Timer


class TestState(object):
    '''Simple mutable mapping (like namedtuple)'''
    def __init__(self, **kwargs):
        for key, val in kwargs.items():
            setattr(self, key, val)


def _test_pre_run(state):
    '''This function is called before a test is executed (setup).

    Args:
        state: Test suite state collection, instance of TestState.

    Returns:
        True if the pre_run step completed without error. Currently the pre-run
        will launch the target test binary on the device and attach an
        lldb-server to it in platform mode.

    Raises:
        AssertionError: If an assertion fails.
        TestSuiteException: Previous processes of this apk required for this
                            test could not be killed.
    '''
    assert state.test
    assert state.bundle

    log = util_log.get_logger()
    log.info('running: {0}'.format(state.name))

    # Remove any cached NDK scripts between tests
    state.bundle.delete_ndk_cache()

    # query our test case for the remote target app it needs
    # First try the legacy behaviour
    try:
        target_name = state.test.get_bundle_target()
        warnings.warn("get_bundle_target() is deprecated and will be removed soon"
                      " - use the `bundle_target` dictionary attribute instead")
    except AttributeError:
        try:
            target_name = state.test.bundle_target[state.bundle_type]
        except KeyError:
            raise TestIgnoredException()

    if target_name is None:
        # test case doesn't require a remote process to debug
        return True
    else:
        # find the pid of our remote test process
        state.pid = state.bundle.launch(target_name)
        if not state.pid:
            log.error('unable to get pid of target')
            return False
        state.android.kill_servers()
        # spawn lldb platform on the target device
        state.android.launch_lldb_platform(state.device_port)
        return True


def _test_post_run(state):
    '''This function is called after a test is executed (cleanup).

    Args:
        state: Test suite state collection, instance of TestState.

    Raises:
        AssertionError: If an assertion fails.
    '''
    assert state.test
    assert state.bundle

    try:
        target_name = state.test.get_bundle_target()
        warnings.warn("get_bundle_target() is deprecated and will be removed soon"
                      " - use the `bundle_target` dictionary attribute instead")
    except AttributeError:
        try:
            target_name = state.test.bundle_target[state.bundle_type]
        except KeyError:
            raise TestIgnoredException()


    if target_name:
        if state.bundle.is_apk(target_name):
            state.android.stop_app(state.bundle.get_package(target_name))
        else:
            state.android.kill_process(target_name)


def _test_run(state):
    '''Execute a single test suite.

    Args:
        state: test suite state collection, instance of TestState.

    Returns:
        True: if the test case ran successfully and passed.
        False: if the test case failed or suffered an error.

    Raises:
        AssertionError: If an assertion fails.
    '''
    assert state.lldb
    assert state.lldb_module
    assert state.test

    test_failures = state.test.run(state.lldb, state.pid, state.lldb_module)

    if test_failures:
        log = util_log.get_logger()
        for test, err in test_failures:
            log.error('test %s:%s failed: %r' % (state.name, test, err))

        return False

    return True


def _initialise_timer(android, interval):
    '''Start a 'timeout' timer, to catch stalled execution.

    This function will start a timer that will act as a timeout killing this
    test session if a test becomes un-responsive.

    Args:
        android: current instance of harness.UtilAndroid
        interval: the interval for the timeout, in seconds

    Returns:
        The instance of the Timer class that was created.
    '''

    def on_timeout():
        '''This is a callback function that will fire if a test takes longer
        then a threshold time to complete.'''
        # Clean up the android properties
        android.reset_all_props()
        # pylint: disable=protected-access
        sys.stdout.flush()
        # hard exit to force kill all threads that may block our exit
        os._exit(util_constants.RC_TEST_TIMEOUT)

    timer = Timer(interval, on_timeout)
    timer.start()
    atexit.register(Timer.stop, timer)
    return timer


def _quit_test(num, timer):
    '''This function will exit making sure the timeout thread is killed.

    Args:
        num: An integer specifying the exit status, 0 meaning "successful
             termination".
        timer: The current Timer instance.
    '''
    if timer:
        timer.stop()
    sys.stdout.flush()
    sys.exit(num)


def _execute_test(state):
    '''Execute a test suite.

    Args:
        state: The current TestState object.
    '''
    log = util_log.get_logger()

    state.test.setup(state.android)
    try:
        if not _test_pre_run(state):
            raise TestSuiteException('test_pre_run() failed')
        if not _test_run(state):
            raise TestSuiteException('test_run() failed')
        _test_post_run(state)
        log.info('Test passed')

    finally:
        state.test.post_run()
        state.test.teardown(state.android)


def _get_test_case_class(module):
    '''Inspect a test case module and return the test case class.

    Args:
        module: A loaded test case module.
    '''
    # We consider only subclasses of TestCase that have `test_` methods`
    log = util_log.get_logger()
    log.debug("loading test suites from %r", module)
    for name, klass in inspect.getmembers(module, inspect.isclass):
        for attr in dir(klass):
            if attr.startswith('test_'):
                log.info("Found test class %r", name)
                return klass
        else:
            log.debug("class %r has no test_ methods", name)
    return None


def get_test_dir(test_name):
    ''' Get the directory that contains a test with a given name.

    Returns:
        A string that is the directory containing the test.

    Raises:
        TestSuiteException: If a test with this name does not exist.
    '''
    tests_dir = os.path.dirname(os.path.realpath(__file__))
    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) and
            test_name in os.listdir(current_test_dir)):
            return current_test_dir

    raise TestSuiteException(
        'unable to find test: {0}'.format(test_name))


def main():
    '''Test runner entry point.'''

    # re-open stdout with no buffering
    sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', 0)

    android = None
    timer = None
    log = None

    # parse the command line (positional arguments only)
    truthy = lambda x: x.lower() in ('true', '1')
    parser = argparse.ArgumentParser("Run a single RenderScript TestSuite against lldb")
    for name, formatter in (
       ('test_name', str),
       ('log_file_path', str),
       ('adb_path', str),
       ('lldb_server_path_device', str),
       ('aosp_product_path', str),
       ('device_port', int),
       ('device', str),
       ('print_to_stdout', truthy),
       ('verbose', truthy),
       ('wimpy', truthy),
       ('timeout', int),
       ('bundle_type', str),
    ):
        parser.add_argument(name, type=formatter)

    args = parser.parse_args()

    try:
        # create utility classes
        harness.util_log.initialise(
            '%s(%s)' % (args.test_name, args.bundle_type),
            print_to_stdout=args.print_to_stdout,
            level=logging.INFO if not args.verbose else logging.DEBUG,
            file_path=args.log_file_path,
            file_mode='a'
        )
        log = util_log.get_logger()
        log.debug('Logger initialised')

        android = harness.UtilAndroid(args.adb_path,
                                      args.lldb_server_path_device,
                                      args.device)

        # start the timeout counter
        timer = _initialise_timer(android, args.timeout)

        # startup lldb and register teardown handler
        atexit.register(UtilLLDB.stop)
        UtilLLDB.start()

        current_test_dir = get_test_dir(args.test_name)

        # load a test case module
        test_module = load_py_module(os.path.join(current_test_dir,
                                                  args.test_name))


        # inspect the test module and locate our test case class
        test_class = _get_test_case_class(test_module)

        # if our test inherits from TestBaseRemote, check we have a valid device
        if (hasattr(test_module, "TestBaseRemote") and
            issubclass(test_class, test_module.TestBaseRemote)):
            android.validate_device()

        # create an instance of our test case
        test_inst = test_class(
            args.device_port,
            args.device,
            timer,
            args.bundle_type,
            wimpy=args.wimpy
        )

        # instantiate a test target bundle
        bundle = harness.UtilBundle(android, args.aosp_product_path)

        # execute the test case
        try:
            for _ in range(2):
                try:
                    # create an lldb instance
                    lldb = UtilLLDB.create_debugger()

                    # create state object to encapsulate instances

                    state = TestState(
                         android=android,
                         bundle=bundle,
                         lldb=lldb,
                         lldb_module=UtilLLDB.get_module(),
                         test=test_inst,
                         pid=None,
                         name=args.test_name,
                         device_port=args.device_port,
                         bundle_type=args.bundle_type
                    )

                    util_warnings.redirect_warnings()

                    _execute_test(state)

                    # tear down the lldb instance
                    UtilLLDB.destroy_debugger(lldb)
                    break
                except DisconnectedException as error:
                    log.warning(error)
                    log.warning('Trying again.')
            else:
                log.fatal('Not trying again, maximum retries exceeded.')
                raise TestSuiteException('Lost connection to lldb-server')

        finally:
            util_warnings.restore_warnings()

        _quit_test(util_constants.RC_TEST_OK, timer)

    except AssertionError:
        if log:
            log.critical('Internal test suite error', exc_info=1)
        print('Internal test suite error', file=sys.stderr)
        _quit_test(util_constants.RC_TEST_FATAL, timer)

    except TestIgnoredException:
        if log:
            log.warn("test ignored")
        _quit_test(util_constants.RC_TEST_IGNORED, timer)

    except TestSuiteException as error:
        if log:
            log.exception(str(error))
        else:
            print(error, file=sys.stderr)
        _quit_test(util_constants.RC_TEST_FAIL, timer)

    # use a global exception handler to be sure that we will
    # exit safely and correctly
    except Exception:
        if log:
            log.exception('INTERNAL ERROR')
        else:
            import traceback
            print('Exception {0}'.format(traceback.format_exc()),
                  file=sys.stderr)
        _quit_test(util_constants.RC_TEST_FATAL, timer)

    finally:
        if android:
            android.reset_all_props()
        if timer:
            timer.stop()


# execution trampoline
if __name__ == '__main__':
    print(' '.join(sys.argv))
    main()