site_utils/run_suite.py

#!/usr/bin/python
#
# Copyright (c) 2012 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.

"""Tool for running suites of tests and waiting for completion.

The desired test suite will be scheduled with autotest, and then
this tool will block until the job is complete, printing a summary
at the end.  Error conditions result in exceptions.

This is intended for use only with Chrome OS test suits that leverage the
dynamic suite infrastructure in server/cros/dynamic_suite.py.
"""

import getpass, optparse, time, sys
import common
import logging
from autotest_lib.client.common_lib import global_config
from autotest_lib.server.cros import frontend_wrappers
from autotest_lib.client.common_lib import logging_config, logging_manager

CONFIG = global_config.global_config


class RunSuiteLoggingConfig(logging_config.LoggingConfig):
    def configure_logging(self, verbose=False):
        super(RunSuiteLoggingConfig, self).configure_logging(use_console=True)


def parse_options():
    usage = "usage: %prog [options] control_file"
    parser = optparse.OptionParser(usage=usage)
    parser.add_option("-b", "--board", dest="board")
    parser.add_option("-i", "--build", dest="build")
    #  This should just be a boolean flag, but the autotest "proxy" code
    #  can't handle flags that don't take arguments.
    parser.add_option("-n", "--no_wait", dest="no_wait", default=None)
    parser.add_option("-p", "--pool", dest="pool", default=None)
    parser.add_option("-s", "--suite_name", dest="name")
    parser.add_option("-t", "--timeout_min", dest="timeout_min", default=30)
    parser.add_option("-d", "--delay_sec", dest="delay_sec", default=10)
    options, args = parser.parse_args()
    return parser, options, args


def get_pretty_status(status):
    if status == 'GOOD':
        return '[ PASSED ]'
    return '[ FAILED ]'


def status_is_relevant(status):
    """
    Indicates whether the status of a given test is meaningful or not.

    @param status: frontend.TestStatus object to look at.
    @return True if this is a test result worth looking at further.
    """
    return not (status['test_name'].startswith('SERVER_JOB') or
                status['test_name'].startswith('CLIENT_JOB'))


def generate_log_link(anchor, job_string):
    """
    Generate a link to this job's logs, for consumption by buildbot.

    @param anchor: Link anchor text.
    @param job_id: the job whose logs we'd like to link to.
    @return A link formatted for the buildbot log annotator.
    """
    host = CONFIG.get_config_value('SERVER', 'hostname', type=str)
    pattern = CONFIG.get_config_value('CROS', 'log_url_pattern', type=str)
    return "@@@STEP_LINK@%s@%s@@@" % (anchor, pattern % (host, job_string))


def main():
    parser, options, args = parse_options()
    if args or not options.build or not options.board or not options.name:
        parser.print_help()
        return

    logging_manager.configure_logging(RunSuiteLoggingConfig())

    afe = frontend_wrappers.RetryingAFE(timeout_min=options.timeout_min,
                                        delay_sec=options.delay_sec)

    wait = options.no_wait is None

    job_id = afe.run('create_suite_job',
                     suite_name=options.name,
                     board=options.board,
                     build=options.build,
                     check_hosts=wait,
                     pool=options.pool)
    TKO = frontend_wrappers.RetryingTKO(timeout_min=options.timeout_min,
                                        delay_sec=options.delay_sec)
    # Return code that will be sent back to autotest_rpc_server.py
    code = 0
    while wait and True:
        if not afe.get_jobs(id=job_id, finished=True):
            time.sleep(1)
            continue
        views = TKO.run('get_detailed_test_views', afe_job_id=job_id)
        width = len(max(map(lambda x: x['test_name'], views), key=len)) + 3

        relevant_views = filter(status_is_relevant, views)
        if not relevant_views:
           # The main suite job most likely failed in SERVER_JOB.
           relevant_views = views

        log_links = []
        for entry in relevant_views:
            test_entry = entry['test_name'].ljust(width)
            print "%s%s" % (test_entry, get_pretty_status(entry['status']))
            if entry['status'] != 'GOOD':
                print "%s  %s: %s" % (test_entry,
                                      entry['status'],
                                      entry['reason'])
                job_name = entry['test_name'].split('.')[0]
                if 'job_keyvals' in entry and job_name in entry['job_keyvals']:
                    job_name = entry['job_keyvals'][job_name]
                else:
                  # We are the main suite job since there is no keyval entry
                  # for our job_name.
                  job_name = '%s-%s' % (job_id, getpass.getuser())

                log_links.append(generate_log_link(entry['test_name'],
                                                   job_name))
                if code == 0 and entry['status'] == 'WARN':
                    # IFF this is the only error, and it's a warning...
                    code = 2
                else:
                    code = 1
        for link in log_links:
            print link
        break
    else:
        print "Created suite job: %r" % job_id
        print generate_log_link(options.name,
                                '%s-%s' % (job_id, getpass.getuser()))
        print "--no_wait specified; Exiting."
    return code

if __name__ == "__main__":
    sys.exit(main())