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