3 import os
, unittest
, time
, datetime
, itertools
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
):
46 self
.god
= mock
.mock_god()
47 self
.god
.stub_function(os
.path
, "exists")
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"),
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"])),
111 def test_distinct_messages(self
):
112 self
.assertEqual(utils
.drop_redundant_messages(set(["abc", "def"])),
116 def test_one_unique_message(self
):
118 utils
.drop_redundant_messages(set(["abc", "abcd", "abcde"])),
122 def test_some_unique_some_not(self
):
124 utils
.drop_redundant_messages(set(["abc", "def", "abcdef",
126 set(["abcdef", "defghi"]))
129 if __name__
== "__main__":