Autotest: Refactor complete_failures slightly.
Move the get_last_pass_times functions to a common area as it will be useful
for future scripts. Also tests are looked up based on status name instead
of status id.
BUG=None
DEPLOY=None
TEST=Unittests and ran script manually to verify
Change-Id: I398237d0f11bef63492e10a1206387bb2058dc40
Reviewed-on: https://gerrit.chromium.org/gerrit/62570
Reviewed-by: Dennis Jeffrey <dennisjeffrey@chromium.org>
Tested-by: Keyar Hood <keyar@chromium.org>
Commit-Queue: Keyar Hood <keyar@chromium.org>
diff --git a/frontend/health/complete_failures.py b/frontend/health/complete_failures.py
index 6688e8f..09cb6c0 100644
--- a/frontend/health/complete_failures.py
+++ b/frontend/health/complete_failures.py
@@ -14,7 +14,7 @@
# Django and the models are only setup after
# the setup_django_readonly_environment module is imported.
from autotest_lib.frontend.tko import models as tko_models
-from django.db import models as django_models
+from autotest_lib.frontend.health import utils
_STORAGE_FILE = 'failure_storage'
@@ -22,7 +22,6 @@
_DAYS_TO_BE_FAILING_TOO_LONG = 60
# Ignore any tests that have not ran in this many days
_DAYS_NOT_RUNNING_CUTOFF = 60
-_TEST_PASS_STATUS_INDEX = 6
_MAIL_RESULTS_FROM = 'chromeos-test-health@google.com'
_MAIL_RESULTS_TO = 'chromeos-lab-infrastructure@google.com'
@@ -69,23 +68,21 @@
return not '/' in name and not name.startswith('try_new_image')
-def get_last_pass_times():
+def prepare_last_passes(last_passes):
"""
- Get all the tests that have passed and the time they last passed.
+ Fix up the last passes so they can be used by the system.
- @return the dict of test_name:last_finish_time pairs for tests that have
- passed.
+ This filters out invalid test names and converts the test names to utf8
+ encoding.
+ @param last_passes: The dictionary of test_name:last_pass pairs.
+
+ @return: Valid entries in encoded as utf8 strings.
"""
- results = tko_models.Test.objects.values('test').filter(
- status=_TEST_PASS_STATUS_INDEX).annotate(
- last_pass=django_models.Max('started_time'))
- results_dict = {result['test']: result['last_pass']
- for result in results}
- valid_test_names = filter(is_valid_test_name, results_dict)
+ valid_test_names = filter(is_valid_test_name, last_passes)
# The shelve module does not accept Unicode objects as keys but does
# accept utf-8 strings.
- return {name.encode('utf8'): results_dict[name]
+ return {name.encode('utf8'): last_passes[name]
for name in valid_test_names}
@@ -116,10 +113,11 @@
"""
recent_test_names = get_recently_ran_test_names()
- last_passes = get_last_pass_times()
+ last_passes = utils.get_last_pass_times()
+ prepared_passes = prepare_last_passes(last_passes)
running_passes = {}
- for test, pass_time in last_passes.items():
+ for test, pass_time in prepared_passes.items():
if test in recent_test_names:
running_passes[test] = pass_time
diff --git a/frontend/health/complete_failures_unittest.py b/frontend/health/complete_failures_unittest.py
index 0d90f9c..17eb209 100755
--- a/frontend/health/complete_failures_unittest.py
+++ b/frontend/health/complete_failures_unittest.py
@@ -14,13 +14,13 @@
from autotest_lib.frontend import setup_django_readonly_environment
from autotest_lib.frontend import setup_test_environment
import complete_failures
+from autotest_lib.frontend.health import utils
from autotest_lib.client.common_lib import mail
from autotest_lib.frontend.tko import models
from django import test
GOOD_STATUS_IDX = 6
-FAIL_STATUS_IDX = 4
# See complte_failurs_functional_tests.py for why we need this.
class MockDatetime(datetime.datetime):
@@ -176,111 +176,21 @@
self.assertFalse(complete_failures.is_valid_test_name(name))
-class GetLastPassTimesTests(mox.MoxTestBase, test.TestCase):
- """Tests the get_last_pass_times function."""
+class PrepareLastPassesTests(test.TestCase):
+ """Tests the prepare_last_passes function."""
def setUp(self):
- super(GetLastPassTimesTests, self).setUp()
- setup_test_environment.set_up()
-
+ super(PrepareLastPassesTests, self).setUp()
def tearDown(self):
- setup_test_environment.tear_down()
- super(GetLastPassTimesTests, self).tearDown()
-
-
- def test_return_most_recent_pass(self):
- """The last time a test passed should be returned."""
- # To add a test entry to the database, Django the test object to
- # be instantiated with various other model instances. We give these
- # instances dummy id values.
- job = models.Job(job_idx=1)
- kernel = models.Kernel(kernel_idx=1)
- machine = models.Machine(machine_idx=1)
- success_status = models.Status(status_idx=GOOD_STATUS_IDX)
-
- early_pass = models.Test(job=job, status=success_status,
- kernel=kernel, machine=machine,
- test='test',
- started_time=datetime.datetime(2012, 1, 1))
- early_pass.save()
- late_pass = models.Test(job=job, status=success_status,
- kernel=kernel, machine=machine,
- test='test',
- started_time=datetime.datetime(2012, 1, 2))
- late_pass.save()
-
- results = complete_failures.get_last_pass_times()
-
- self.assertEquals(results, {'test': datetime.datetime(2012, 1, 2)})
-
-
- def test_only_return_passing_tests(self):
- """Tests that only tests that have passed at some point are returned."""
- job = models.Job(job_idx=1)
- kernel = models.Kernel(kernel_idx=1)
- machine = models.Machine(machine_idx=1)
- success_status = models.Status(status_idx=GOOD_STATUS_IDX)
- fail_status = models.Status(status_idx=FAIL_STATUS_IDX)
-
- passing_test = models.Test(job=job, status=success_status,
- kernel=kernel, machine=machine,
- test='passing_test',
- started_time=datetime.datetime(2012, 1, 1))
- passing_test.save()
- failing_test = models.Test(job=job, status=fail_status,
- kernel=kernel, machine=machine,
- test='failing_test',
- started_time=datetime.datetime(2012, 1, 1))
- failing_test.save()
-
- results = complete_failures.get_last_pass_times()
-
- self.assertEquals(results,
- {'passing_test': datetime.datetime(2012, 1, 1)})
-
-
- def test_return_all_passing_tests(self):
- """This function returns all tests that passed at least once."""
- job = models.Job(job_idx=1)
- kernel = models.Kernel(kernel_idx=1)
- machine = models.Machine(machine_idx=1)
- success_status = models.Status(status_idx=GOOD_STATUS_IDX)
-
- test1 = models.Test(job=job, status=success_status,
- kernel=kernel, machine=machine,
- test='test1',
- started_time=datetime.datetime(2012, 1, 1))
- test1.save()
- test2 = models.Test(job=job, status=success_status,
- kernel=kernel, machine=machine,
- test='test2',
- started_time=datetime.datetime(2012, 1, 2))
- test2.save()
-
- results = complete_failures.get_last_pass_times()
-
- self.assertEquals(results, {'test1': datetime.datetime(2012, 1, 1),
- 'test2': datetime.datetime(2012, 1, 2)})
-
+ super(PrepareLastPassesTests, self).tearDown()
def test_does_not_return_invalid_test_names(self):
"""Tests that tests with invalid test names are not returned."""
- job = models.Job(job_idx=1)
- kernel = models.Kernel(kernel_idx=1)
- machine = models.Machine(machine_idx=1)
- success_status = models.Status(status_idx=GOOD_STATUS_IDX)
- fail_status = models.Status(status_idx=FAIL_STATUS_IDX)
+ results = complete_failures.prepare_last_passes(['invalid_test/name'])
- invalid_test = models.Test(job=job, status=success_status,
- kernel=kernel, machine=machine,
- test='invalid_test/name',
- started_time=datetime.datetime(2012, 1, 1))
- invalid_test.save()
+ self.assertEqual(results, {})
- results = complete_failures.get_last_pass_times()
-
- self.assertTrue(not results)
class GetRecentlyRanTestNamesTests(mox.MoxTestBase, test.TestCase):
@@ -356,38 +266,16 @@
self.assertEqual(len(results), 1)
- def test_does_not_return_invalid_test_names(self):
- """Tests that only tests with invalid test names are not returned."""
- job = models.Job(job_idx=1)
- kernel = models.Kernel(kernel_idx=1)
- machine = models.Machine(machine_idx=1)
- success_status = models.Status(status_idx=GOOD_STATUS_IDX)
-
- invalid_test = models.Test(job=job, status=success_status,
- kernel=kernel, machine=machine,
- test='invalid_test/name',
- started_time=self.datetime(2012, 1, 1))
- invalid_test.save()
-
- datetime.datetime.today().AndReturn(self.datetime(2012, 1, 2))
- complete_failures._DAYS_NOT_RUNNING_CUTOFF = 60
-
- self.mox.ReplayAll()
- results = complete_failures.get_recently_ran_test_names()
-
- self.assertTrue(not results)
-
-
class GetTestsToAnalyzeTests(mox.MoxTestBase):
"""Tests the get_tests_to_analyze function."""
def test_returns_recent_test_names(self):
"""Test should return all the test names in the database."""
- self.mox.StubOutWithMock(complete_failures, 'get_last_pass_times')
+ self.mox.StubOutWithMock(utils, 'get_last_pass_times')
self.mox.StubOutWithMock(complete_failures,
'get_recently_ran_test_names')
- complete_failures.get_last_pass_times().AndReturn({'passing_test':
+ utils.get_last_pass_times().AndReturn({'passing_test':
datetime.datetime(2012, 1 ,1),
'old_passing_test': datetime.datetime(2011, 1, 1)})
complete_failures.get_recently_ran_test_names().AndReturn(
@@ -403,11 +291,11 @@
def test_returns_failing_tests_with_min_datetime(self):
"""Test that never-passed tests are paired with datetime.min."""
- self.mox.StubOutWithMock(complete_failures, 'get_last_pass_times')
+ self.mox.StubOutWithMock(utils, 'get_last_pass_times')
self.mox.StubOutWithMock(complete_failures,
'get_recently_ran_test_names')
- complete_failures.get_last_pass_times().AndReturn({})
+ utils.get_last_pass_times().AndReturn({})
complete_failures.get_recently_ran_test_names().AndReturn({'test'})
self.mox.ReplayAll()
diff --git a/frontend/health/utils.py b/frontend/health/utils.py
new file mode 100644
index 0000000..8b3b58e
--- /dev/null
+++ b/frontend/health/utils.py
@@ -0,0 +1,27 @@
+# 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.
+
+import common
+from autotest_lib.frontend import setup_django_readonly_environment
+
+# Django and the models are only setup after
+# the setup_django_readonly_environment module is imported.
+from autotest_lib.frontend.tko import models as tko_models
+from django.db import models as django_models
+
+_TEST_PASS_STATUS = 'GOOD'
+
+def get_last_pass_times():
+ """
+ Get all the tests that have passed and the time they last passed.
+
+ @return the dict of test_name:last_finish_time pairs for tests that have
+ passed.
+
+ """
+ results = tko_models.Test.objects.values('test').filter(
+ status__word=_TEST_PASS_STATUS).annotate(
+ last_pass=django_models.Max('started_time'))
+ return {result['test']: result['last_pass'] for result in results}
+
diff --git a/frontend/health/utils_unittests.py b/frontend/health/utils_unittests.py
new file mode 100755
index 0000000..b0c1318
--- /dev/null
+++ b/frontend/health/utils_unittests.py
@@ -0,0 +1,121 @@
+#!/usr/bin/python
+#
+# 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.
+
+import datetime, unittest
+
+import mox
+
+import common
+# This must come before the import of utils in order to use the in memory
+# database.
+from autotest_lib.frontend import setup_django_readonly_environment
+from autotest_lib.frontend import setup_test_environment
+from autotest_lib.frontend.health import utils
+from autotest_lib.frontend.tko import models
+from django import test
+
+FAIL_STATUS = models.Status(status_idx=4, word='FAIL')
+GOOD_STATUS = models.Status(status_idx=6, word='GOOD')
+
+
+def add_statuses():
+ """
+ Save the statuses to the in-memory database.
+
+ These normally exist in the database and the code expects them. However, the
+ normal test database setup does not do this for us.
+ """
+ FAIL_STATUS.save()
+ GOOD_STATUS.save()
+
+
+class GetLastPassTimesTests(mox.MoxTestBase, test.TestCase):
+ """Tests the get_last_pass_times function."""
+
+ def setUp(self):
+ super(GetLastPassTimesTests, self).setUp()
+ setup_test_environment.set_up()
+ add_statuses()
+
+
+ def tearDown(self):
+ setup_test_environment.tear_down()
+ super(GetLastPassTimesTests, self).tearDown()
+
+
+ def test_return_most_recent_pass(self):
+ """The last time a test passed should be returned."""
+ # To add a test entry to the database, the test object has to
+ # be instantiated with various other model instances. We give these
+ # instances dummy id values.
+ job = models.Job(job_idx=1)
+ kernel = models.Kernel(kernel_idx=1)
+ machine = models.Machine(machine_idx=1)
+
+ early_pass = models.Test(job=job, status=GOOD_STATUS,
+ kernel=kernel, machine=machine,
+ test='test',
+ started_time=datetime.datetime(2012, 1, 1))
+ early_pass.save()
+ late_pass = models.Test(job=job, status=GOOD_STATUS,
+ kernel=kernel, machine=machine,
+ test='test',
+ started_time=datetime.datetime(2012, 1, 2))
+ late_pass.save()
+
+ results = utils.get_last_pass_times()
+
+ self.assertEquals(results, {'test': datetime.datetime(2012, 1, 2)})
+
+
+ def test_only_return_passing_tests(self):
+ """Tests that only tests that have passed at some point are returned."""
+ job = models.Job(job_idx=1)
+ kernel = models.Kernel(kernel_idx=1)
+ machine = models.Machine(machine_idx=1)
+
+ passing_test = models.Test(job=job, status=GOOD_STATUS,
+ kernel=kernel, machine=machine,
+ test='passing_test',
+ started_time=datetime.datetime(2012, 1, 1))
+ passing_test.save()
+ failing_test = models.Test(job=job, status=FAIL_STATUS,
+ kernel=kernel, machine=machine,
+ test='failing_test',
+ started_time=datetime.datetime(2012, 1, 1))
+ failing_test.save()
+
+ results = utils.get_last_pass_times()
+
+ self.assertEquals(results,
+ {'passing_test': datetime.datetime(2012, 1, 1)})
+
+
+ def test_return_all_passing_tests(self):
+ """This function returns all tests that passed at least once."""
+ job = models.Job(job_idx=1)
+ kernel = models.Kernel(kernel_idx=1)
+ machine = models.Machine(machine_idx=1)
+
+ test1 = models.Test(job=job, status=GOOD_STATUS,
+ kernel=kernel, machine=machine,
+ test='test1',
+ started_time=datetime.datetime(2012, 1, 1))
+ test1.save()
+ test2 = models.Test(job=job, status=GOOD_STATUS,
+ kernel=kernel, machine=machine,
+ test='test2',
+ started_time=datetime.datetime(2012, 1, 2))
+ test2.save()
+
+ results = utils.get_last_pass_times()
+
+ self.assertEquals(results, {'test1': datetime.datetime(2012, 1, 1),
+ 'test2': datetime.datetime(2012, 1, 2)})
+
+
+if __name__ == '__main__':
+ unittest.main()