common_lib.base_packages: Add parallel bzip2 support to package manager
[autotest-zwu.git] / tko / utils_unittest.py
blob7d184079d268b28d14621422ee3adb3b9042015e
1 #!/usr/bin/python
3 import os, unittest, time, datetime, itertools
5 import common
6 from autotest_lib.client.common_lib.test_utils import mock
7 from autotest_lib.tko import utils
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)
25 def test_returns_none_on_missing_value(self):
26 date = utils.get_timestamp({}, "missing_key")
27 self.assertEquals(date, None)
30 def test_fails_on_non_integer_values(self):
31 self.assertRaises(ValueError, utils.get_timestamp,
32 {"key": "zero"}, "key")
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)
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")
50 def tearDown(self):
51 self.god.unstub_all()
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)
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)
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)
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), "/")
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)
101 class drop_redundant_messages(unittest.TestCase):
102 def test_empty_set(self):
103 self.assertEqual(utils.drop_redundant_messages(set()), set())
106 def test_singleton(self):
107 self.assertEqual(utils.drop_redundant_messages(set(["abc"])),
108 set(["abc"]))
111 def test_distinct_messages(self):
112 self.assertEqual(utils.drop_redundant_messages(set(["abc", "def"])),
113 set(["abc", "def"]))
116 def test_one_unique_message(self):
117 self.assertEqual(
118 utils.drop_redundant_messages(set(["abc", "abcd", "abcde"])),
119 set(["abcde"]))
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"]))
129 if __name__ == "__main__":
130 unittest.main()