blob: 1d7e0312b2b791d0626ac30a40c97ed3a9b00756 [file] [log] [blame]
# Copyright (c) 2013 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 to validate code in prod branch before pushing to lab.
The script runs push_to_prod suite to verify code in prod branch is ready to be
pushed. Link to design document:
To verify if prod branch can be pushed to lab, run following command in
chromeos-autotest.cbf server:
/usr/local/autotest/site_utils/ -e
The script uses latest stumpy canary build as test build by default.
import argparse
import getpass
import os
import re
import subprocess
import sys
import urllib2
import common
from autotest_lib.frontend import setup_django_environment
from autotest_lib.frontend.afe import models
except ImportError:
# Unittest may not have Django database configured and will fail to import.
from autotest_lib.client.common_lib import global_config, mail
from autotest_lib.server import site_utils
from autotest_lib.server.cros.dynamic_suite import frontend_wrappers
from autotest_lib.server.cros.dynamic_suite import reporting
from autotest_lib.server.hosts import cros_host
CONFIG = global_config.global_config
DEVSERVERS = CONFIG.get_config_value('CROS', 'dev_server', type=list,
BUILD_REGEX = '^R[\d]+-[\d]+\.[\d]+\.[\d]+$'
PUSH_TO_PROD_SUITE = 'push_to_prod'
AU_SUITE = 'paygen_au_canary'
SUITE_JOB_START_INFO_REGEX = ('^.*Created suite job:.*'
# Dictionary of test results keyed by test name regular expression.
# This is related to dummy_Fail/control.dependency.
'dummy_Fail.dependency$': 'TEST_NA',
'telemetry_CrosTests.*': 'GOOD',
'platform_InstallTestImage_SERVER_JOB$': 'GOOD',
'dummy_Pass.*': 'GOOD',
'dummy_Fail.Fail$': 'FAIL',
'dummy_Fail.RetryFail$': 'FAIL',
'dummy_Fail.RetrySuccess': 'GOOD',
'dummy_Fail.Error$': 'ERROR',
'dummy_Fail.Warn$': 'WARN',
'dummy_Fail.NAError$': 'TEST_NA',
'dummy_Fail.Crash$': 'GOOD',
'autoupdate_EndToEndTest.paygen_au_canary_test_delta.*': 'GOOD',
'autoupdate_EndToEndTest.paygen_au_canary_test_full.*': 'GOOD',
# Anchor for the auto-filed bug for dummy_Fail tests.
BUG_ANCHOR = 'TestFailure(push_to_prod,dummy_Fail.Fail,always fail)'
URL_HOST = CONFIG.get_config_value('SERVER', 'hostname', type=str)
URL_PATTERN = CONFIG.get_config_value('CROS', 'log_url_pattern', type=str)
# Save all run_suite command output.
run_suite_output = []
class TestPushException(Exception):
"""Exception to be raised when the test to push to prod failed."""
def get_default_build(devserver=None, board='stumpy'):
"""Get the default build to be used for test.
@param devserver: devserver used to look for latest staged build. If value
is None, all devservers in config will be tried.
@param board: Name of board to be tested, default is stumpy.
@return: Build to be tested, e.g., stumpy-release/R36-5881.0.0
LATEST_BUILD_URL_PATTERN = '%s/latestbuild?target=%s-release'
build = None
if not devserver:
for server in DEVSERVERS:
url = LATEST_BUILD_URL_PATTERN % (server, board)
build = urllib2.urlopen(url).read()
if build and re.match(BUILD_REGEX, build):
return '%s-release/%s' % (board, build)
# If no devserver has any build staged for the given board, use the stable
# build in config.
build = CONFIG.get_config_value('CROS', 'stable_cros_version')
return '%s-release/%s' % (board, build)
def parse_arguments():
"""Parse arguments for test_push tool.
@return: Parsed arguments.
parser = argparse.ArgumentParser()
parser.add_argument('-b', '--board', dest='board', default='stumpy',
help='Default is stumpy.')
parser.add_argument('-i', '--build', dest='build', default=None,
help='Default is the latest canary build of given '
'board. Must be a canary build, otherwise AU test '
'will fail.')
parser.add_argument('-p', '--pool', dest='pool', default='bvt')
parser.add_argument('-u', '--num', dest='num', type=int, default=3,
help='Run on at most NUM machines.')
parser.add_argument('-f', '--file_bugs', dest='file_bugs', default='True',
help='File bugs on test failures. Must pass "True" or '
'"False" if used.')
parser.add_argument('-e', '--email', dest='email', default=None,
help='Email address for the notification to be sent to '
'after the script finished running.')
parser.add_argument('-d', '--devserver', dest='devserver',
help='devserver to find what\'s the latest build.')
arguments = parser.parse_args(sys.argv[1:])
# Get latest canary build as default build.
if not = get_default_build(arguments.devserver,
return arguments
def do_run_suite(suite_name, arguments):
"""Call run_suite to run a suite job, and return the suite job id.
The script waits the suite job to finish before returning the suite job id.
Also it will echo the run_suite output to stdout.
@param suite_name: Name of a suite, e.g., dummy.
@param arguments: Arguments for run_suite command.
@return: Suite job ID.
dir = os.path.dirname(os.path.realpath(__file__))
cmd = [os.path.join(dir, RUN_SUITE_COMMAND),
'-s', suite_name,
'-b', arguments.board,
'-p', arguments.pool,
'-u', str(arguments.num),
'-f', arguments.file_bugs]
suite_job_id = None
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
while True:
line = proc.stdout.readline()
# Break when run_suite process completed.
if not line and proc.poll() != None:
print line.rstrip()
if not suite_job_id:
m = re.match(SUITE_JOB_START_INFO_REGEX, line)
if m and
suite_job_id = int(
if not suite_job_id:
raise TestPushException('Failed to retrieve suite job ID.')
print 'Suite job %s is completed.' % suite_job_id
return suite_job_id
def check_dut_image(build, suite_job_id):
"""Confirm all DUTs used for the suite are imaged to expected build.
@param build: Expected build to be imaged.
@param suite_job_id: job ID of the suite job.
@raise TestPushException: If a DUT does not have expected build imaged.
print 'Checking image installed in DUTs...'
job_ids = [ for job in
hqes = [models.HostQueueEntry.objects.filter(job_id=job_id)[0]
for job_id in job_ids]
hostnames = set([ for hqe in hqes])
for hostname in hostnames:
host = cros_host.CrosHost(hostname)
found_build = host.get_build()
if found_build != build:
raise TestPushException('DUT is not imaged properly. Host %s has '
'build %s, while build %s is expected.' %
(hostname, found_build, build))
def test_suite(suite_name, expected_results, arguments):
"""Call run_suite to start a suite job and verify results.
@param suite_name: Name of a suite, e.g., dummy
@param expected_results: A dictionary of test name to test result.
@param arguments: Arguments for run_suite command.
suite_job_id = do_run_suite(suite_name, arguments)
# Confirm all DUTs used for the suite are imaged to expected build.
if suite_name != AU_SUITE:
check_dut_image(, suite_job_id)
# Find all tests and their status
print 'Comparing test results...'
TKO = frontend_wrappers.RetryingTKO(timeout_min=0.1, delay_sec=10)
test_views = site_utils.get_test_views_from_tko(suite_job_id, TKO)
mismatch_errors = []
extra_test_errors = []
found_keys = set()
for test_name,test_status in test_views.items():
print "%s%s" % (test_name.ljust(30), test_status)
test_found = False
for key,val in expected_results.items():
if, test_name):
test_found = True
# TODO(dshi): result for this test is ignored until servo is
# added to a host accessible by cbf server (
if key == 'platform_InstallTestImage_SERVER_JOB$':
# TODO(dshi): result for this test is ignored until the bug is
# fixed in Telemetry (
if key == 'telemetry_CrosTests.*':
if val != test_status:
error = ('%s Expected: [%s], Actual: [%s]' %
(test_name, val, test_status))
if not test_found:
missing_test_errors = set(expected_results.keys()) - found_keys
# For latest build, npo_test_delta does not exist.
if missing_test_errors == set(['autoupdate_EndToEndTest.npo_test_delta.*']):
missing_test_errors = set([])
# For trybot build, nmo_test_delta does not exist.
if missing_test_errors == set(['autoupdate_EndToEndTest.nmo_test_delta.*']):
missing_test_errors = set([])
summary = []
if mismatch_errors:
summary.append(('Results of %d test(s) do not match expected '
'values:') % len(mismatch_errors))
if extra_test_errors:
summary.append('%d test(s) are not expected to be run:' %
if missing_test_errors:
summary.append('%d test(s) are missing from the results:' %
# Test link to log can be loaded.
job_name = '%s-%s' % (suite_job_id, getpass.getuser())
log_link = URL_PATTERN % (URL_HOST, job_name)
except urllib2.URLError:
summary.append('Failed to load page for link to log: %s.' % log_link)
if summary:
raise TestPushException('\n'.join(summary))
def close_bug():
"""Close all existing bugs filed for dummy_Fail.
@return: A list of issue ids to be used in check_bug_filed_and_deduped.
old_issue_ids = []
reporter = reporting.Reporter()
while True:
issue = reporter.find_issue_by_marker(BUG_ANCHOR)
if not issue:
return old_issue_ids
if in old_issue_ids:
raise TestPushException('Failed to close issue %d' %
comment='Issue closed by test_push script.',
def check_bug_filed_and_deduped(old_issue_ids):
"""Confirm bug related to dummy_Fail was filed and deduped.
@param old_issue_ids: A list of issue ids that was closed earlier. id of the
new issue must be not in this list.
@raise TestPushException: If auto bug file failed to create a new issue or
dedupe multiple failures.
reporter = reporting.Reporter()
issue = reporter.find_issue_by_marker(BUG_ANCHOR)
if not issue:
raise TestPushException('Auto bug file failed. Unable to locate bug '
'with marker %s' % BUG_ANCHOR)
if old_issue_ids and in old_issue_ids:
raise TestPushException('Auto bug file failed to create a new issue. '
'id of the old issue found is %d.' %
if not ('%s2' % reporter.AUTOFILED_COUNT) in issue.labels:
raise TestPushException(('Auto bug file failed to dedupe for issue %d '
'with labels of %s.') %
(, issue.labels))
# Close the bug, and do the search again, which should return None.
comment='Issue closed by test_push script.',
second_issue = reporter.find_issue_by_marker(BUG_ANCHOR)
if second_issue:
ids = '%d, %d' % (,
raise TestPushException(('Auto bug file failed. Multiple issues (%s) '
'filed with marker %s') % (ids, BUG_ANCHOR))
print 'Issue %d was filed and deduped successfully.' %
def main():
"""Entry point for test_push script."""
arguments = parse_arguments()
# Close existing bugs. New bug should be filed in dummy_Fail test.
old_issue_ids = close_bug()
# TODO(dshi): Remove following line after is fixed.
test_suite(AU_SUITE, EXPECTED_TEST_RESULTS_AU, arguments)
except Exception as e:
print 'Test for pushing to prod failed:\n'
print str(e)
# Send out email about the test failure.
'Test for pushing to prod failed. Do NOT push!',
('Errors occurred during the test:\n\n%s\n\n' % str(e) +
'run_suite output:\n\n%s' % '\n'.join(run_suite_output)))
message = ('\nAll tests are completed successfully, prod branch is ready to'
' be pushed.')
print message
# Send out email about test completed successfully.
'Test for pushing to prod completed successfully',
if __name__ == '__main__':