Home | History | Annotate | Download | only in tko 
      1 #!/usr/bin/python
      2 
      3 import os, unittest, time, datetime, itertools
      4 
      5 import common
      6 from autotest_lib.client.common_lib.test_utils import mock
      7 from autotest_lib.tko import utils
      8 
      9 
     10 class get_timestamp_test(unittest.TestCase):
     11     def test_zero_time(self):
     12         date = utils.get_timestamp({"key": "0"}, "key")
     13         timezone = datetime.timedelta(seconds=time.timezone)
     14         utc_date = date + timezone
     15         # should be equal to epoch, i.e. Jan 1, 1970
     16         self.assertEquals(utc_date.year, 1970)
     17         self.assertEquals(utc_date.month, 1)
     18         self.assertEquals(utc_date.day, 1)
     19         self.assertEquals(utc_date.hour, 0)
     20         self.assertEquals(utc_date.minute, 0)
     21         self.assertEquals(utc_date.second, 0)
     22         self.assertEquals(utc_date.microsecond, 0)
     23 
     24 
     25     def test_returns_none_on_missing_value(self):
     26         date = utils.get_timestamp({}, "missing_key")
     27         self.assertEquals(date, None)
     28 
     29 
     30     def test_fails_on_non_integer_values(self):
     31         self.assertRaises(ValueError, utils.get_timestamp,
     32                           {"key": "zero"}, "key")
     33 
     34 
     35     def test_date_can_be_string_or_integer(self):
     36         int_times = [1, 12, 123, 1234, 12345, 123456]
     37         str_times = [str(t) for t in int_times]
     38         for int_t, str_t in itertools.izip(int_times, str_times):
     39             date_int = utils.get_timestamp({"key": int_t}, "key")
     40             date_str = utils.get_timestamp({"key": str_t}, "key")
     41             self.assertEquals(date_int, date_str)
     42 
     43 
     44 class find_toplevel_job_dir_test(unittest.TestCase):
     45     def setUp(self):
     46         self.god = mock.mock_god()
     47         self.god.stub_function(os.path, "exists")
     48 
     49 
     50     def tearDown(self):
     51         self.god.unstub_all()
     52 
     53 
     54     def test_start_is_toplevel(self):
     55         jobdir = "/results/job1"
     56         os.path.exists.expect_call(
     57             jobdir + "/.autoserv_execute").and_return(True)
     58         self.assertEqual(utils.find_toplevel_job_dir(jobdir), jobdir)
     59 
     60 
     61     def test_parent_is_toplevel(self):
     62         jobdir = "/results/job2"
     63         os.path.exists.expect_call(
     64             jobdir + "/sub/.autoserv_execute").and_return(False)
     65         os.path.exists.expect_call(
     66             jobdir + "/.autoserv_execute").and_return(True)
     67         self.assertEqual(utils.find_toplevel_job_dir(jobdir + "/sub"), jobdir)
     68 
     69 
     70     def test_grandparent_is_toplevel(self):
     71         jobdir = "/results/job3"
     72         os.path.exists.expect_call(
     73             jobdir + "/sub/sub/.autoserv_execute").and_return(False)
     74         os.path.exists.expect_call(
     75             jobdir + "/sub/.autoserv_execute").and_return(False)
     76         os.path.exists.expect_call(
     77             jobdir + "/.autoserv_execute").and_return(True)
     78         self.assertEqual(utils.find_toplevel_job_dir(jobdir + "/sub/sub"),
     79                          jobdir)
     80 
     81     def test_root_is_toplevel(self):
     82         jobdir = "/results/job4"
     83         os.path.exists.expect_call(
     84             jobdir + "/.autoserv_execute").and_return(False)
     85         os.path.exists.expect_call(
     86             "/results/.autoserv_execute").and_return(False)
     87         os.path.exists.expect_call("/.autoserv_execute").and_return(True)
     88         self.assertEqual(utils.find_toplevel_job_dir(jobdir), "/")
     89 
     90 
     91     def test_no_toplevel(self):
     92         jobdir = "/results/job5"
     93         os.path.exists.expect_call(
     94             jobdir + "/.autoserv_execute").and_return(False)
     95         os.path.exists.expect_call(
     96             "/results/.autoserv_execute").and_return(False)
     97         os.path.exists.expect_call("/.autoserv_execute").and_return(False)
     98         self.assertEqual(utils.find_toplevel_job_dir(jobdir), None)
     99 
    100 
    101 class drop_redundant_messages(unittest.TestCase):
    102     def test_empty_set(self):
    103         self.assertEqual(utils.drop_redundant_messages(set()), set())
    104 
    105 
    106     def test_singleton(self):
    107         self.assertEqual(utils.drop_redundant_messages(set(["abc"])),
    108                          set(["abc"]))
    109 
    110 
    111     def test_distinct_messages(self):
    112         self.assertEqual(utils.drop_redundant_messages(set(["abc", "def"])),
    113                          set(["abc", "def"]))
    114 
    115 
    116     def test_one_unique_message(self):
    117         self.assertEqual(
    118                 utils.drop_redundant_messages(set(["abc", "abcd", "abcde"])),
    119                 set(["abcde"]))
    120 
    121 
    122     def test_some_unique_some_not(self):
    123         self.assertEqual(
    124                 utils.drop_redundant_messages(set(["abc", "def", "abcdef",
    125                                                    "defghi", "cd"])),
    126                 set(["abcdef", "defghi"]))
    127 
    128 
    129 if __name__ == "__main__":
    130     unittest.main()
    131