1 #!/usr/bin/python 2 # 3 # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. 4 # Use of this source code is governed by a BSD-style license that can be 5 # found in the LICENSE file. 6 7 8 import argparse, datetime, sys 9 10 import common 11 from autotest_lib.client.common_lib import mail 12 from autotest_lib.frontend import setup_django_readonly_environment 13 14 # Django and the models are only setup after 15 # the setup_django_readonly_environment module is imported. 16 from autotest_lib.frontend.tko import models as tko_models 17 from autotest_lib.frontend.health import utils 18 19 20 # Mark a test as failing too long if it has not passed in this many days 21 _DAYS_TO_BE_FAILING_TOO_LONG = 60 22 # Ignore any tests that have not ran in this many days 23 _DAYS_NOT_RUNNING_CUTOFF = 60 24 _MAIL_RESULTS_FROM = 'chromeos-test-health (at] google.com' 25 _MAIL_RESULTS_TO = 'chromeos-lab-infrastructure (at] google.com' 26 27 28 def is_valid_test_name(name): 29 """ 30 Returns if a test name is valid or not. 31 32 There is a bunch of entries in the tko_test table that are not actually 33 test names. They are there as a side effect of how Autotest uses this 34 table. 35 36 Two examples of bad tests names are as follows: 37 link-release/R29-4228.0.0/faft_ec/firmware_ECPowerG3_SERVER_JOB 38 try_new_image-chormeos1-rack2-host2 39 40 @param name: The candidate test names to check. 41 @return True if name is a valid test name and false otherwise. 42 43 """ 44 return not '/' in name and not name.startswith('try_new_image') 45 46 47 def prepare_last_passes(last_passes): 48 """ 49 Fix up the last passes so they can be used by the system. 50 51 This filters out invalid test names and converts the test names to utf8 52 encoding. 53 54 @param last_passes: The dictionary of test_name:last_pass pairs. 55 56 @return: Valid entries in encoded as utf8 strings. 57 """ 58 valid_test_names = filter(is_valid_test_name, last_passes) 59 # The shelve module does not accept Unicode objects as keys but does 60 # accept utf-8 strings. 61 return {name.encode('utf8'): last_passes[name] 62 for name in valid_test_names} 63 64 65 def get_recently_ran_test_names(): 66 """ 67 Get all the test names from the database that have been recently ran. 68 69 @return a set of the recently ran tests. 70 71 """ 72 cutoff_delta = datetime.timedelta(_DAYS_NOT_RUNNING_CUTOFF) 73 cutoff_date = datetime.datetime.today() - cutoff_delta 74 results = tko_models.Test.objects.filter( 75 started_time__gte=cutoff_date).values('test').distinct() 76 test_names = [test['test'] for test in results] 77 valid_test_names = filter(is_valid_test_name, test_names) 78 return {test.encode('utf8') for test in valid_test_names} 79 80 81 def get_tests_to_analyze(recent_test_names, last_pass_times): 82 """ 83 Get all the recently ran tests as well as the last time they have passed. 84 85 The minimum datetime is given as last pass time for tests that have never 86 passed. 87 88 @param recent_test_names: The set of the names of tests that have been 89 recently ran. 90 @param last_pass_times: The dictionary of test_name:last_pass_time pairs. 91 92 @return the dict of test_name:last_finish_time pairs. 93 94 """ 95 prepared_passes = prepare_last_passes(last_pass_times) 96 97 running_passes = {} 98 for test, pass_time in prepared_passes.items(): 99 if test in recent_test_names: 100 running_passes[test] = pass_time 101 102 failures_names = recent_test_names.difference(running_passes) 103 always_failed = {test: datetime.datetime.min for test in failures_names} 104 return dict(always_failed.items() + running_passes.items()) 105 106 107 def email_about_test_failure(failed_tests, all_tests): 108 """ 109 Send an email about all the tests that have failed if there are any. 110 111 @param failed_tests: The list of failed tests. This will be sorted in this 112 function. 113 @param all_tests: All the names of tests that have been recently ran. 114 115 """ 116 if failed_tests: 117 failed_tests.sort() 118 mail.send(_MAIL_RESULTS_FROM, 119 [_MAIL_RESULTS_TO], 120 [], 121 'Long Failing Tests', 122 '%d/%d tests have been failing for at least %d days.\n' 123 'They are the following:\n\n%s' 124 % (len(failed_tests), len(all_tests), 125 _DAYS_TO_BE_FAILING_TOO_LONG, 126 '\n'.join(failed_tests))) 127 128 129 def filter_out_good_tests(tests): 130 """ 131 Remove all tests that have passed recently enough to be good. 132 133 @param tests: The tests to filter on. 134 135 @return: A list of tests that have not passed for a long time. 136 137 """ 138 cutoff = (datetime.datetime.today() - 139 datetime.timedelta(_DAYS_TO_BE_FAILING_TOO_LONG)) 140 return [name for name, last_pass in tests.items() if last_pass < cutoff] 141 142 143 def parse_options(args): 144 """Parse the command line options.""" 145 146 description = ('Collects information about which tests have been ' 147 'failing for a long time and creates an email summarizing ' 148 'the results.') 149 parser = argparse.ArgumentParser(description=description) 150 parser.parse_args(args) 151 152 153 def main(args=None): 154 """ 155 The script code. 156 157 Allows other python code to import and run this code. This will be more 158 important if a nice way to test this code can be determined. 159 160 @param args: The command line arguments being passed in. 161 162 """ 163 args = [] if args is None else args 164 parse_options(args) 165 all_test_names = get_recently_ran_test_names() 166 last_passes = utils.get_last_pass_times() 167 tests = get_tests_to_analyze(all_test_names, last_passes) 168 failures = filter_out_good_tests(tests) 169 email_about_test_failure(failures, all_test_names) 170 171 172 173 if __name__ == '__main__': 174 sys.exit(main(sys.argv[1:])) 175
