site_utils/brillo_test_launcher.py

#!/usr/bin/python
# 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 logging
import os
import sys
import tempfile
import time

import common
try:
    # Ensure the chromite site-package is installed.
    from chromite.lib import *
except ImportError:
    import subprocess
    build_externals_path = os.path.join(
            os.path.dirname(os.path.dirname(os.path.realpath(__file__))),
            'utils', 'build_externals.py')
    subprocess.check_call([build_externals_path, 'chromiterepo'])
    # Restart the script so python now finds the autotest site-packages.
    sys.exit(os.execv(__file__, sys.argv))
from autotest_lib.client.common_lib import control_data
from autotest_lib.server import utils
from autotest_lib.server.cros.dynamic_suite import control_file_getter
from autotest_lib.server.cros.dynamic_suite import tools
from autotest_lib.server.hosts import moblab_host
from autotest_lib.site_utils import brillo_common
from autotest_lib.site_utils import run_suite


_AFE_JOB_PAGE_TEMPLATE = ('http://%(moblab)s/afe/#tab_id=view_job&'
                          'object_id=%(job_id)s')
_AFE_HOST_PAGE_TEMPLATE = ('http://%(moblab)s/afe/#tab_id=view_host&'
                           'object_id=%(host_id)s')
_QUICKMERGE_LIST = ('client/',
                    'global_config.ini',
                    'server/',
                    'site_utils/',
                    'test_suites/',
                    'utils/')


class BrilloTestExecutionError(brillo_common.BrilloTestError):
    """An error while launching and running a test."""


def setup_parser(parser):
    """Add parser options.

    @param parser: argparse.ArgumentParser of the script.
    """
    parser.add_argument('-t', '--test_name',
                        help="Name of the test to run. This is either the "
                             "name in the test's default control file e.g. "
                             "brillo_Gtests or a specific control file's "
                             "filename e.g. control.brillo_GtestsWhitelist.")
    parser.add_argument('-A', '--test_arg', metavar='NAME=VAL',
                        dest='test_args', default=[], action='append',
                        help='An argument to pass to the test.')


def quickmerge(moblab):
    """Transfer over a subset of Autotest directories.

    Quickmerge allows developers to do basic editting of tests and test
    libraries on their workstation without requiring them to emerge and cros
    deploy the autotest-server package.

    @param moblab: MoblabHost representing the MobLab being used to launch the
                   testing.
    """
    autotest_rootdir = os.path.dirname(
            os.path.dirname(os.path.realpath(__file__)))
    # We use rsync -R to copy a bunch of sources in a single run, adding a dot
    # to pinpoint the relative path root.
    rsync_cmd = ['rsync', '-aR', '--exclude', '*.pyc']
    ssh_cmd = 'ssh -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null'
    if int(moblab.port) != 22:
        ssh_cmd += ' -p %s' % moblab.port
    rsync_cmd += ['-e', ssh_cmd]
    rsync_cmd += [os.path.join(autotest_rootdir, '.', path)
                  for path in _QUICKMERGE_LIST]
    rsync_cmd.append('moblab@%s:%s' %
                     (moblab.hostname, moblab_host.AUTOTEST_INSTALL_DIR))
    utils.run(rsync_cmd, timeout=240)


def add_adb_host(moblab, adb_hostname):
    """Add the ADB host to the MobLab's host list.

    @param moblab: MoblabHost representing the MobLab being used to launch the
                   tests.
    @param adb_hostname: Hostname of the ADB Host.

    @returns The adb host to use for launching tests.
    """
    if not adb_hostname:
        adb_hostname = 'localhost'
        moblab.enable_adb_testing()
    if all([host.hostname != adb_hostname for host in moblab.afe.get_hosts()]):
        moblab.add_dut(adb_hostname)
    return adb_hostname


def schedule_test(moblab, host, test, test_args):
    """Schedule a Brillo test.

    @param moblab: MoblabHost representing the MobLab being used for testing.
    @param host: Hostname of the DUT.
    @param test: Test name.
    @param test_args: Iterable of 'NAME=VAL' (strings) encoding argument
                      assignments for the test.

    @returns autotest_lib.server.frontend.Job object representing the scheduled
             job.
    """
    getter = control_file_getter.FileSystemGetter(
            [os.path.dirname(os.path.dirname(os.path.realpath(__file__)))])
    controlfile_conts = getter.get_control_file_contents_by_name(test)

    # TODO(garnold) This should be removed and arguments injected by feeding
    # args=test_args to create_jobs() directly once crbug.com/545572 is fixed.
    if test_args:
        controlfile_conts = tools.inject_vars({'args': test_args},
                                              controlfile_conts)

    job = moblab.afe.create_job(
            controlfile_conts, name=test,
            control_type=control_data.CONTROL_TYPE_NAMES.SERVER,
            hosts=[host], require_ssp=False)
    logging.info('Tests Scheduled. Please wait for results.')
    job_page = _AFE_JOB_PAGE_TEMPLATE % dict(moblab=moblab.web_address,
                                             job_id=job.id)
    logging.info('Progress can be monitored at %s', job_page)
    logging.info('Please note tests that launch other tests (e.g. sequences) '
                 'might complete quickly, but links to child jobs will appear '
                 'shortly at the bottom on the page (Hit Refresh).')
    return job


def get_all_jobs(moblab, parent_job):
    """Generate a list of the parent_job and it's subjobs.

    @param moblab: MoblabHost representing the MobLab being used for testing.
    @param host: Hostname of the DUT.
    @param parent_job: autotest_lib.server.frontend.Job object representing the
                       parent job.

    @returns list of autotest_lib.server.frontend.Job objects.
    """
    jobs_list = moblab.afe.get_jobs(id=parent_job.id)
    jobs_list.extend(moblab.afe.get_jobs(parent_job=parent_job.id))
    return jobs_list


def wait_for_test_completion(moblab, host, parent_job):
    """Wait for the parent job and it's subjobs to complete.

    @param moblab: MoblabHost representing the MobLab being used for testing.
    @param host: Hostname of the DUT.
    @param parent_job: autotest_lib.server.frontend.Job object representing the
                       test job.
    """
    # Wait for the sequence job and it's sub-jobs to finish, while monitoring
    # the DUT state. As long as the DUT does not go into 'Repair Failed' the
    # tests will complete.
    while (moblab.afe.get_jobs(id=parent_job.id, not_yet_run=True,
                               running=True)
           or moblab.afe.get_jobs(parent_job=parent_job.id, not_yet_run=True,
                                  running=True)):
        afe_host = moblab.afe.get_hosts(hostnames=(host,))[0]
        if afe_host.status == 'Repair Failed':
            moblab.afe.abort_jobs(
                [j.id for j in get_all_jobs(moblab, parent_job)])
            host_page = _AFE_HOST_PAGE_TEMPLATE % dict(
                    moblab=moblab.web_address, host_id=afe_host.id)
            raise BrilloTestExecutionError(
                    'ADB dut %s has become Repair Failed. More information '
                    'can be found at %s' % (host, host_page))
        time.sleep(10)


def copy_results(moblab, parent_job):
    """Copy job results locally.

    @param moblab: MoblabHost representing the MobLab being used for testing.
    @param parent_job: autotest_lib.server.frontend.Job object representing the
                       parent job.

    @returns Temporary directory path.
    """
    tempdir = tempfile.mkdtemp(prefix='brillo_test_results')
    for job in get_all_jobs(moblab, parent_job):
        moblab.get_file('/usr/local/autotest/results/%d-moblab' % job.id,
                        tempdir)
    return tempdir


def output_results(moblab, parent_job):
    """Output the Brillo PTS and it's subjobs results.

    @param moblab: MoblabHost representing the MobLab being used for testing.
    @param parent_job: autotest_lib.server.frontend.Job object representing the
                       test job.
    """
    solo_test_run = len(moblab.afe.get_jobs(parent_job=parent_job.id)) == 0
    rc = run_suite.ResultCollector(moblab.web_address, moblab.afe, moblab.tko,
                                   None, None, parent_job.name, parent_job.id,
                                   user='moblab', solo_test_run=solo_test_run)
    rc.run()
    rc.output_results()


def main(args):
    """The main function."""
    args = brillo_common.parse_args('Launch a Brillo test using Moblab.',
                                    setup_parser=setup_parser)
    moblab, _ = brillo_common.get_moblab_and_devserver_port(args.moblab_host)

    if args.quickmerge:
        quickmerge(moblab)

    # Add the adb host object to the MobLab.
    adb_host = add_adb_host(moblab, args.adb_host)

    # Schedule the test job.
    test_job = schedule_test(moblab, adb_host, args.test_name, args.test_args)
    wait_for_test_completion(moblab, adb_host, test_job)

    # Gather and report the test results.
    local_results_folder = copy_results(moblab, test_job)
    output_results(moblab, test_job)
    logging.info('Results have also been copied locally to %s',
                 local_results_folder)


if __name__ == '__main__':
    try:
        main(sys.argv)
        sys.exit(0)
    except brillo_common.BrilloTestError as e:
        logging.error('Error: %s', e)

    sys.exit(1)