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
8import argparse, datetime, sys
9
10import common
11from autotest_lib.client.common_lib import mail
12from 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.
16from autotest_lib.frontend.tko import models as tko_models
17from 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@google.com'
25_MAIL_RESULTS_TO = 'chromeos-lab-infrastructure@google.com'
26
27
28def 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
47def 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
65def 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
81def 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
107def 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
129def 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
143def 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
153def 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
173if __name__ == '__main__':
174    sys.exit(main(sys.argv[1:]))
175