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