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