site_utils/brillo_common.py

# Copyright 2015 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.

import argparse
import logging
import os
import sys

import common
from autotest_lib.client.common_lib import error
from autotest_lib.server import hosts
from autotest_lib.server import utils
from autotest_lib.server.hosts import moblab_host
from autotest_lib.server.hosts import ssh_host


_LOGGING_FORMAT = '%(asctime)s - %(levelname)s - %(message)s'
_TEST_LAUNCH_SCRIPT = 'brillo_test_launcher.py'

# Running against a virtual machine has several intricacies that we need to
# adjust for. Namely SSH requires the use of 'localhost' while HTTP requires
# the use of '127.0.0.1'. Also because we are forwarding the ports from the VM
# to the host system, the ports to use for these services are also different
# from running on a physical machine.
_VIRT_MACHINE_SSH_ADDR = 'localhost:9222'
_VIRT_MACHINE_AFE_ADDR = '127.0.0.1:8888'
_VIRT_MACHINE_DEVSERVER_PORT = '7777'
_PHYS_MACHINE_DEVSERVER_PORT = '8080'
_MOBLAB_MIN_VERSION = 7569
_MOBLAB_IMAGE_DOWNLOAD_URL = ('https://storage.googleapis.com/chromeos-image-'
                              'archive/moblab_brillo_images/'
                              'moblab_brillo_%s.bin' % _MOBLAB_MIN_VERSION)


class BrilloTestError(Exception):
    """A general error while testing Brillo."""


class BrilloMoblabInitializationError(BrilloTestError):
    """An error during Moblab initialization or handling."""


def get_moblab_and_devserver_port(moblab_hostname):
    """Initializes and returns a MobLab Host Object.

    @params moblab_hostname: The Moblab hostname, None if using a local virtual
                             machine.

    @returns A pair consisting of a MoblabHost and a devserver port.

    @raise BrilloMoblabInitializationError: Failed to set up the Moblab.
    """
    if moblab_hostname:
        web_address = moblab_hostname
        devserver_port = _PHYS_MACHINE_DEVSERVER_PORT
        rpc_timeout_min = 2
    else:
        moblab_hostname = _VIRT_MACHINE_SSH_ADDR
        web_address = _VIRT_MACHINE_AFE_ADDR
        devserver_port = _VIRT_MACHINE_DEVSERVER_PORT
        rpc_timeout_min = 5

    try:
        host = hosts.create_host(moblab_hostname,
                                 host_class=moblab_host.MoblabHost,
                                 connectivity_class=ssh_host.SSHHost,
                                 web_address=web_address,
                                 retain_image_storage=True,
                                 rpc_timeout_min=rpc_timeout_min)
    except error.AutoservRunError as e:
        raise BrilloMoblabInitializationError(
                'Unable to connect to the MobLab: %s' % e)

    moblab_version = int(host.get_release_version().split('.')[0])
    if moblab_version < _MOBLAB_MIN_VERSION:
        raise BrilloMoblabInitializationError(
                'The Moblab version (%s) is older than the minimum required '
                '(%s). Download a current version from URL: %s' %
                (moblab_version, _MOBLAB_MIN_VERSION,
                _MOBLAB_IMAGE_DOWNLOAD_URL))

    try:
        host.afe.get_hosts()
    except Exception as e:
        raise BrilloMoblabInitializationError(
                "Unable to communicate with the MobLab's web frontend, "
                "please verify that it is up and running at http://%s/\n"
                "Error: %s" % (host.web_address, e))

    return host, devserver_port


def parse_args(description, setup_parser=None, validate_args=None):
    """Parse command-line arguments.

    @param description: The script description in the help message.
    @param setup_parser: Function that takes a parser object and adds
                         script-specific options to it.
    @param validate_args: Function that takes a parser object and the parsed
                          arguments and validates the arguments. It should use
                          parser.error() to report errors.

    @return Parsed and validated arguments.
    """
    parser = argparse.ArgumentParser(description=description)
    if setup_parser:
        setup_parser(parser)

    # Add common options.
    parser.add_argument('-m', '--moblab_host',
                        help='MobLab hostname or IP to launch tests. If this '
                             'argument is not provided, the test launcher '
                             'will attempt to test a local virtual machine '
                             'instance of MobLab.')
    parser.add_argument('-a', '--adb_host',
                        help='Hostname or IP of the adb_host connected to the '
                             'Brillo DUT. Default is to assume it is connected '
                             'directly to the MobLab.')
    parser.add_argument('-n', '--no_quickmerge', dest='quickmerge',
                        action='store_false',
                        help='Do not update the Autotest code on the Moblab')
    parser.add_argument('-d', '--debug', action='store_true',
                        help='Print log statements.')

    args = parser.parse_args()

    # Configure the root logger.
    logging.getLogger().setLevel(logging.DEBUG if args.debug else logging.INFO)
    for log_handler in logging.getLogger().handlers:
        log_handler.setFormatter(logging.Formatter(fmt=_LOGGING_FORMAT))

    if validate_args:
        validate_args(parser, args)

    return args


def setup_test_action_parser(parser):
    """Add parser options related to test action.

    @param parser: argparse.ArgumentParser of the script.
    """
    launch_opts = parser.add_mutually_exclusive_group()
    launch_opts.add_argument('-A', '--print_args', action='store_true',
                             help='Print test arguments to stdout instead of '
                                  'launching the test.')
    launch_opts.add_argument('-C', '--print_command', action='store_true',
                             help='Print complete test launch command instead '
                                  'of launching the test.')


def _get_arg_strs(test_args):
    """Converts an argument dictionary into a list of 'arg=val' strings."""
    return ['%s=%s' % kv for kv in test_args.iteritems()]


def _get_command(moblab, test_name, test_args, do_quickmerge, do_quote):
    """Returns the test launch command.

    @param moblab: MoblabHost representing the MobLab being used for testing.
    @param test_name: The name of the test to run.
    @param test_args: Dictionary of test arguments.
    @param do_quickmerge: If False, pass the --no-quickmerge flag.
    @param do_quote: If True, add single-quotes around test arguments.

    @return Test launch command as a list of strings.
    """
    # pylint: disable=missing-docstring
    def quote(val):
        return "'%s'" % val if do_quote else val

    cmd = [os.path.join(os.path.dirname(__file__), _TEST_LAUNCH_SCRIPT),
           '-t', quote(test_name)]
    if not do_quickmerge:
        cmd.append('-n')
    if not moblab.hostname.startswith('localhost'):
           cmd += ['-m', quote(moblab.hostname)]
    for arg_str in _get_arg_strs(test_args):
        cmd += ['-A', quote(arg_str)]
    return cmd


def _print_args(test_args):
    """Prints the test arguments to stdout, one per line.

    @param test_args: Dictionary of test arguments.
    """
    print('\n'.join(_get_arg_strs(test_args)))


def _print_command(moblab, test_name, test_args, do_quickmerge):
    """Prints the test launch command to stdout with quoting.

    @param moblab: MoblabHost representing the MobLab being used for testing.
    @param test_name: The name of the test to run.
    @param test_args: Dictionary of test arguments.
    @param do_quickmerge: If False, pass the --no-quickmerge flag.
    """
    print(' '.join(
            _get_command(moblab, test_name, test_args, do_quickmerge, True)))


def _run_command(moblab, test_name, test_args, do_quickmerge):
    """Runs the test launch script.

    @param moblab: MoblabHost representing the MobLab being used for testing.
    @param test_name: The name of the test to run.
    @param test_args: Dictionary of test arguments.
    @param do_quickmerge: If False, pass the --no_quickmerge flag.
    """
    utils.run(_get_command(moblab, test_name, test_args, do_quickmerge, False),
              stdout_tee=sys.stdout, stderr_tee=sys.stderr)


def do_test_action(args, moblab, test_name, test_args):
    """Performs the desired action related to the test.

    @param args: Parsed arguments.
    @param moblab: MoblabHost representing the MobLab being used for testing.
    @param test_name: The name of the test to run.
    @param test_args: Dictionary of test arguments.
    """
    if args.print_args:
        logging.info('Printing test arguments')
        _print_args(test_args)
    elif args.print_command:
        logging.info('Printing test launch command')
        _print_command(moblab, test_name, test_args, args.quickmerge)
    else:
        logging.info('Launching test')
        _run_command(moblab, test_name, test_args, args.quickmerge)