| #!/usr/bin/python3 |
| |
| from __future__ import absolute_import |
| from __future__ import division |
| from __future__ import print_function |
| |
| import os, unittest, time, datetime, itertools |
| |
| import common |
| from autotest_lib.client.common_lib.test_utils import mock |
| from autotest_lib.tko import utils |
| from six.moves import zip |
| |
| |
| class get_timestamp_test(unittest.TestCase): |
| def test_zero_time(self): |
| date = utils.get_timestamp({"key": "0"}, "key") |
| timezone = datetime.timedelta(seconds=time.timezone) |
| utc_date = date + timezone |
| # should be equal to epoch, i.e. Jan 1, 1970 |
| self.assertEquals(utc_date.year, 1970) |
| self.assertEquals(utc_date.month, 1) |
| self.assertEquals(utc_date.day, 1) |
| self.assertEquals(utc_date.hour, 0) |
| self.assertEquals(utc_date.minute, 0) |
| self.assertEquals(utc_date.second, 0) |
| self.assertEquals(utc_date.microsecond, 0) |
| |
| |
| def test_returns_none_on_missing_value(self): |
| date = utils.get_timestamp({}, "missing_key") |
| self.assertEquals(date, None) |
| |
| |
| def test_fails_on_non_integer_values(self): |
| self.assertRaises(ValueError, utils.get_timestamp, |
| {"key": "zero"}, "key") |
| |
| |
| def test_date_can_be_string_or_integer(self): |
| int_times = [1, 12, 123, 1234, 12345, 123456] |
| str_times = [str(t) for t in int_times] |
| for int_t, str_t in zip(int_times, str_times): |
| date_int = utils.get_timestamp({"key": int_t}, "key") |
| date_str = utils.get_timestamp({"key": str_t}, "key") |
| self.assertEquals(date_int, date_str) |
| |
| |
| class find_toplevel_job_dir_test(unittest.TestCase): |
| def setUp(self): |
| self.god = mock.mock_god() |
| self.god.stub_function(os.path, "exists") |
| |
| |
| def tearDown(self): |
| self.god.unstub_all() |
| |
| |
| def test_start_is_toplevel(self): |
| jobdir = "/results/job1" |
| os.path.exists.expect_call( |
| jobdir + "/.autoserv_execute").and_return(True) |
| self.assertEqual(utils.find_toplevel_job_dir(jobdir), jobdir) |
| |
| |
| def test_parent_is_toplevel(self): |
| jobdir = "/results/job2" |
| os.path.exists.expect_call( |
| jobdir + "/sub/.autoserv_execute").and_return(False) |
| os.path.exists.expect_call( |
| jobdir + "/.autoserv_execute").and_return(True) |
| self.assertEqual(utils.find_toplevel_job_dir(jobdir + "/sub"), jobdir) |
| |
| |
| def test_grandparent_is_toplevel(self): |
| jobdir = "/results/job3" |
| os.path.exists.expect_call( |
| jobdir + "/sub/sub/.autoserv_execute").and_return(False) |
| os.path.exists.expect_call( |
| jobdir + "/sub/.autoserv_execute").and_return(False) |
| os.path.exists.expect_call( |
| jobdir + "/.autoserv_execute").and_return(True) |
| self.assertEqual(utils.find_toplevel_job_dir(jobdir + "/sub/sub"), |
| jobdir) |
| |
| def test_root_is_toplevel(self): |
| jobdir = "/results/job4" |
| os.path.exists.expect_call( |
| jobdir + "/.autoserv_execute").and_return(False) |
| os.path.exists.expect_call( |
| "/results/.autoserv_execute").and_return(False) |
| os.path.exists.expect_call("/.autoserv_execute").and_return(True) |
| self.assertEqual(utils.find_toplevel_job_dir(jobdir), "/") |
| |
| |
| def test_no_toplevel(self): |
| jobdir = "/results/job5" |
| os.path.exists.expect_call( |
| jobdir + "/.autoserv_execute").and_return(False) |
| os.path.exists.expect_call( |
| "/results/.autoserv_execute").and_return(False) |
| os.path.exists.expect_call("/.autoserv_execute").and_return(False) |
| self.assertEqual(utils.find_toplevel_job_dir(jobdir), None) |
| |
| |
| class drop_redundant_messages(unittest.TestCase): |
| def test_empty_set(self): |
| self.assertEqual(utils.drop_redundant_messages(set()), set()) |
| |
| |
| def test_singleton(self): |
| self.assertEqual(utils.drop_redundant_messages(set(["abc"])), |
| set(["abc"])) |
| |
| |
| def test_distinct_messages(self): |
| self.assertEqual(utils.drop_redundant_messages(set(["abc", "def"])), |
| set(["abc", "def"])) |
| |
| |
| def test_one_unique_message(self): |
| self.assertEqual( |
| utils.drop_redundant_messages(set(["abc", "abcd", "abcde"])), |
| set(["abcde"])) |
| |
| |
| def test_some_unique_some_not(self): |
| self.assertEqual( |
| utils.drop_redundant_messages(set(["abc", "def", "abcdef", |
| "defghi", "cd"])), |
| set(["abcdef", "defghi"])) |
| |
| |
| if __name__ == "__main__": |
| unittest.main() |