xref: /aosp_15_r20/external/autotest/site_utils/test_that.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)

#!/usr/bin/python3
# Copyright (c) 2013 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import argparse
import json
import os
import signal
import subprocess
import sys

import logging
# Turn the logging level to INFO before importing other autotest
# code, to avoid having failed import logging messages confuse the
# test_that user.
logging.basicConfig(level=logging.INFO)

import common
from autotest_lib.client.common_lib import error, logging_manager
from autotest_lib.server import server_logging_config
from autotest_lib.server.cros.dynamic_suite import constants
from autotest_lib.server.hosts import factory
from autotest_lib.site_utils import test_runner_utils


_QUICKMERGE_SCRIPTNAME = '/mnt/host/source/chromite/bin/autotest_quickmerge'


def _get_info_from_host(remote, board=None, model=None, ssh_options=''):
    """Get the info of the remote host if needed.

    @param remote: string representing the IP of the remote host.
    @param board: board arg from CLI.
    @param model: model arg from CLI.

    @return: board, model string representing the board, model
        of the remote host.
    """

    if board and model:
        return board, model

    host = factory.create_host(remote, ssh_options=ssh_options)

    if not board:
        logging.info(
                'Board unspecified, attempting to determine board from host.')
        try:
            board = host.get_board().replace(constants.BOARD_PREFIX, '')
        except error.AutoservRunError:
            raise test_runner_utils.TestThatRunError(
                    'Cannot determine board, please specify a --board option.')
        logging.info('Detected host board: %s', board)

    if not model:
        logging.info(
                'Model unspecified, attempting to determine model from host.')
        try:
            model = host.get_platform()
        except error.AutoservRunError:
            raise test_runner_utils.TestThatRunError(
                    'Cannot determine model, please specify a --model option.')
        logging.info('Detected host model: %s', model)

    return board, model


def validate_arguments(arguments):
    """
    Validates parsed arguments.

    @param arguments: arguments object, as parsed by ParseArguments
    @raises: ValueError if arguments were invalid.
    """
    if arguments.remote == ':lab:':
        if arguments.args:
            raise ValueError('--args flag not supported when running against '
                             ':lab:')
        if arguments.pretend:
            raise ValueError('--pretend flag not supported when running '
                             'against :lab:')
        if arguments.ssh_verbosity:
            raise ValueError('--ssh_verbosity flag not supported when running '
                             'against :lab:')
        if not arguments.board or arguments.build == test_runner_utils.NO_BUILD:
            raise ValueError('--board and --build are both required when '
                             'running against :lab:')
    else:
        if arguments.web:
            raise ValueError('--web flag not supported when running locally')

    try:
        json.loads(arguments.host_attributes)
    except TypeError:
        raise ValueError("--host_attributes must be quoted dict, got: %s" %
                         arguments.host_attributes)


def parse_arguments(argv):
    """
    Parse command line arguments

    @param argv: argument list to parse
    @returns:    parsed arguments
    @raises SystemExit if arguments are malformed, or required arguments
            are not present.
    """
    return _parse_arguments_internal(argv)[0]


def _parse_arguments_internal(argv):
    """
    Parse command line arguments

    @param argv: argument list to parse
    @returns:    tuple of parsed arguments and argv suitable for remote runs
    @raises SystemExit if arguments are malformed, or required arguments
            are not present.
    """
    local_parser, remote_argv = parse_local_arguments(argv)

    parser = argparse.ArgumentParser(description='Run remote tests.',
                                     parents=[local_parser])

    parser.add_argument('remote', metavar='REMOTE',
                        help='hostname[:port] for remote device. Specify '
                             ':lab: to run in test lab. When tests are run in '
                             'the lab, test_that will use the client autotest '
                             'package for the build specified with --build, '
                             'and the lab server code rather than local '
                             'changes.')
    test_runner_utils.add_common_args(parser)
    parser.add_argument('-b', '--board', metavar='BOARD',
                        action='store',
                        help='Board for which the test will run. '
                             'Default: %(default)s')
    parser.add_argument('-m',
                        '--model',
                        metavar='MODEL',
                        help='Specific model the test will run against. '
                        'Matches the model:FAKE_MODEL label for the host.')
    parser.add_argument('-i', '--build', metavar='BUILD',
                        default=test_runner_utils.NO_BUILD,
                        help='Build to test. Device will be reimaged if '
                             'necessary. Omit flag to skip reimage and test '
                             'against already installed DUT image. Examples: '
                             'link-paladin/R34-5222.0.0-rc2, '
                             'lumpy-release/R34-5205.0.0')
    parser.add_argument('-p', '--pool', metavar='POOL', default='suites',
                        help='Pool to use when running tests in the lab. '
                             'Default is "suites"')
    parser.add_argument('--autotest_dir', metavar='AUTOTEST_DIR',
                        help='Use AUTOTEST_DIR instead of normal board sysroot '
                             'copy of autotest, and skip the quickmerge step.')
    parser.add_argument('--no-quickmerge', action='store_true', default=False,
                        dest='no_quickmerge',
                        help='Skip the quickmerge step and use the sysroot '
                             'as it currently is. May result in un-merged '
                             'source tree changes not being reflected in the '
                             'run. If using --autotest_dir, this flag is '
                             'automatically applied.')
    parser.add_argument('--allow-chrome-crashes',
                        action='store_true',
                        default=False,
                        dest='allow_chrome_crashes',
                        help='Ignore chrome crashes when producing test '
                        'report. This flag gets passed along to the '
                        'report generation tool.')
    parser.add_argument('--ssh_private_key', action='store',
                        default=test_runner_utils.TEST_KEY_PATH,
                        help='Path to the private ssh key.')
    parser.add_argument(
            '--companion_hosts',
            action='store',
            default=None,
            help='Companion duts for the test, quoted space seperated strings')
    parser.add_argument('--dut_servers',
                        action='store',
                        default=None,
                        help='DUT servers for the test.')
    parser.add_argument('--minus',
                        dest='minus',
                        nargs='*',
                        help='List of tests to not use.',
                        default=[''])
    parser.add_argument('--py_version',
                        dest='py_version',
                        help='Python version to use, passed '
                        'to Autotest modules, defaults to 2.',
                        default=None)
    parser.add_argument('--CFT',
                        action='store_true',
                        default=False,
                        dest='CFT',
                        help="If running in, or mocking, the CFT env.")
    parser.add_argument('--host_attributes',
                        action='store',
                        default='{}',
                        help='host_attributes')
    parser.add_argument('--host_labels',
                        action='store',
                        default="",
                        help='host_labels, quoted space seperated strings')
    parser.add_argument('--label',
                        action='store',
                        default="",
                        help='label for test name')
    return parser.parse_args(argv), remote_argv


def parse_local_arguments(argv):
    """
    Strips out arguments that are not to be passed through to runs.

    Add any arguments that should not be passed to remote test_that runs here.

    @param argv: argument list to parse.
    @returns: tuple of local argument parser and remaining argv.
    """
    parser = argparse.ArgumentParser(add_help=False)
    parser.add_argument('-w', '--web', dest='web', default=None,
                        help='Address of a webserver to receive test requests.')
    parser.add_argument('-x', '--max_runtime_mins', type=int,
                        dest='max_runtime_mins', default=20,
                        help='Default time allowed for the tests to complete.')
    parser.add_argument('--no-retries', '--no-retry',
                        dest='retry', action='store_false', default=True,
                        help='For local runs only, ignore any retries '
                             'specified in the control files.')
    _, remaining_argv = parser.parse_known_args(argv)
    return parser, remaining_argv


def perform_bootstrap_into_autotest_root(arguments, autotest_path, argv):
    """
    Perfoms a bootstrap to run test_that from the |autotest_path|.

    This function is to be called from test_that's main() script, when
    test_that is executed from the source tree location. It runs
    autotest_quickmerge to update the sysroot unless arguments.no_quickmerge
    is set. It then executes and waits on the version of test_that.py
    in |autotest_path|.

    @param arguments: A parsed arguments object, as returned from
                      test_that.parse_arguments(...).
    @param autotest_path: Full absolute path to the autotest root directory.
    @param argv: The arguments list, as passed to main(...)

    @returns: The return code of the test_that script that was executed in
              |autotest_path|.
    """
    logging_manager.configure_logging(
            server_logging_config.ServerLoggingConfig(),
            use_console=True,
            verbose=arguments.debug)
    if arguments.no_quickmerge:
        logging.info('Skipping quickmerge step.')
    else:
        logging.info('Running autotest_quickmerge step.')
        command = [_QUICKMERGE_SCRIPTNAME, '--board='+arguments.board]
        s = subprocess.Popen(command,
                             stdout=subprocess.PIPE,
                             stderr=subprocess.STDOUT)
        for message in iter(s.stdout.readline, b''):
            logging.info('quickmerge| %s', message.strip())
        return_code = s.wait()
        if return_code:
            raise test_runner_utils.TestThatRunError(
                    'autotest_quickmerge failed with error code %s.' %
                    return_code)

    logging.info('Re-running test_that script in %s copy of autotest.',
                 autotest_path)
    script_command = os.path.join(autotest_path, 'site_utils',
                                  'test_that.py')
    if not os.path.exists(script_command):
        raise test_runner_utils.TestThatRunError(
            'Unable to bootstrap to autotest root, %s not found.' %
            script_command)
    proc = None
    def resend_sig(signum, stack_frame):
        #pylint: disable-msg=C0111
        if proc:
            proc.send_signal(signum)
    signal.signal(signal.SIGINT, resend_sig)
    signal.signal(signal.SIGTERM, resend_sig)

    proc = subprocess.Popen([script_command] + argv)

    return proc.wait()


def _main_for_local_run(argv, arguments):
    """
    Effective entry point for local test_that runs.

    @param argv: Script command line arguments.
    @param arguments: Parsed command line arguments.
    """
    results_directory = test_runner_utils.create_results_directory(
            arguments.results_dir, arguments.board)
    test_runner_utils.add_ssh_identity(results_directory,
                                       arguments.ssh_private_key)
    arguments.results_dir = results_directory

    # If the board and/or model is not specified through --board and/or
    # --model, and is not set in the default_board file, determine the board by
    # ssh-ing into the host. Also prepend it to argv so we can re-use it when we
    # run test_that from the sysroot.
    arguments.board, arguments.model = _get_info_from_host(
            arguments.remote,
            arguments.board,
            arguments.model,
            ssh_options=arguments.ssh_options)
    argv = ['--board=%s' % (arguments.board, )] + argv
    argv = ['--model=%s' % (arguments.model, )] + argv

    if arguments.autotest_dir:
        autotest_path = arguments.autotest_dir
        arguments.no_quickmerge = True
    else:
        sysroot_path = os.path.join('/build', arguments.board, '')

        if not os.path.exists(sysroot_path):
            print(('%s does not exist. Have you run '
                   'setup_board?' % sysroot_path), file=sys.stderr)
            return 1

        path_ending = 'usr/local/build/autotest'
        autotest_path = os.path.join(sysroot_path, path_ending)

    site_utils_path = os.path.join(autotest_path, 'site_utils')

    if not os.path.exists(autotest_path):
        print(('%s does not exist. Have you run '
               'build_packages? Or if you are using '
               '--autotest_dir, make sure it points to '
               'a valid autotest directory.' % autotest_path), file=sys.stderr)
        return 1

    realpath = os.path.realpath(__file__)
    site_utils_path = os.path.realpath(site_utils_path)

    # If we are not running the sysroot version of script, perform
    # a quickmerge if necessary and then re-execute
    # the sysroot version of script with the same arguments.
    if os.path.dirname(realpath) != site_utils_path:
        return perform_bootstrap_into_autotest_root(
                arguments, autotest_path, argv)
    else:
        return test_runner_utils.perform_run_from_autotest_root(
                autotest_path,
                argv,
                arguments.tests,
                arguments.remote,
                build=arguments.build,
                board=arguments.board,
                model=arguments.model,
                args=arguments.args,
                ignore_deps=not arguments.enforce_deps,
                results_directory=results_directory,
                ssh_verbosity=arguments.ssh_verbosity,
                ssh_options=arguments.ssh_options,
                iterations=arguments.iterations,
                fast_mode=arguments.fast_mode,
                debug=arguments.debug,
                allow_chrome_crashes=arguments.allow_chrome_crashes,
                pretend=arguments.pretend,
                job_retry=arguments.retry,
                companion_hosts=arguments.companion_hosts,
                minus=arguments.minus,
                dut_servers=arguments.dut_servers,
                is_cft=arguments.CFT,
                host_attributes=json.loads(arguments.host_attributes),
                host_labels=arguments.host_labels,
                label=arguments.label)


def _main_for_lab_run(argv, arguments):
    """
    Effective entry point for lab test_that runs.

    @param argv: Script command line arguments.
    @param arguments: Parsed command line arguments.
    """
    autotest_path = os.path.realpath(os.path.join(
            os.path.dirname(os.path.realpath(__file__)),
            '..',
    ))
    command = [os.path.join(autotest_path, 'site_utils',
                            'run_suite.py'),
               '--board=%s' % (arguments.board,),
               '--build=%s' % (arguments.build,),
               '--model=%s' % (arguments.model,),
               '--suite_name=%s' % 'test_that_wrapper',
               '--pool=%s' % (arguments.pool,),
               '--max_runtime_mins=%s' % str(arguments.max_runtime_mins),
               '--suite_args=%s'
               % repr({'tests': _suite_arg_tests(argv)})]
    if arguments.web:
        command.extend(['--web=%s' % (arguments.web,)])
    logging.info('About to start lab suite with command %s.', command)
    return subprocess.call(command)


def _suite_arg_tests(argv):
    """
    Construct a list of tests to pass into suite_args.

    This is passed in suite_args to run_suite for running a test in the
    lab.

    @param argv: Remote Script command line arguments.
    """
    arguments = parse_arguments(argv)
    return arguments.tests


def main(argv):
    """
    Entry point for test_that script.

    @param argv: arguments list
    """
    arguments, remote_argv = _parse_arguments_internal(argv)
    try:
        validate_arguments(arguments)
    except ValueError as err:
        print(('Invalid arguments. %s' % str(err)), file=sys.stderr)
        return 1

    if arguments.remote == ':lab:':
        return _main_for_lab_run(remote_argv, arguments)
    else:
        return _main_for_local_run(argv, arguments)


if __name__ == '__main__':
    sys.exit(main(sys.argv[1:]))