1#!/usr/bin/python
2
3import os, unittest, time, datetime, itertools
4
5import common
6from autotest_lib.client.common_lib.test_utils import mock
7from autotest_lib.tko import utils
8
9
10class 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
44class 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
101class 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
129if __name__ == "__main__":
130    unittest.main()
131