1 #!/usr/bin/python 2 3 import os 4 import tempfile 5 import unittest 6 7 import common 8 from autotest_lib.server import server_job 9 from autotest_lib.client.common_lib import base_job_unittest 10 from autotest_lib.client.common_lib.test_utils import mock 11 12 13 class test_find_base_directories( 14 unittest.TestCase, 15 base_job_unittest.test_find_base_directories.generic_tests): 16 17 def setUp(self): 18 self.job = server_job.server_job.__new__(server_job.server_job) 19 20 21 def test_relative_path_layout(self): 22 existing_file = server_job.__file__ 23 server_job.__file__ = '/rootdir/atest/server/server_job.py' 24 try: 25 autodir, clientdir, serverdir = ( 26 self.job._find_base_directories()) 27 self.assertEqual(autodir, '/rootdir/atest') 28 self.assertEqual(clientdir, '/rootdir/atest/client') 29 self.assertEqual(serverdir, '/rootdir/atest/server') 30 finally: 31 server_job.__file__ = existing_file 32 33 34 class test_init(base_job_unittest.test_init.generic_tests, unittest.TestCase): 35 OPTIONAL_ATTRIBUTES = ( 36 base_job_unittest.test_init.generic_tests.OPTIONAL_ATTRIBUTES 37 - set(['serverdir', 'num_tests_run', 'num_tests_failed', 38 'warning_manager', 'warning_loggers', 'in_lab'])) 39 40 def setUp(self): 41 self.god = mock.mock_god() 42 self.job = server_job.server_job.__new__( 43 server_job.server_job) 44 self.job._job_directory = base_job_unittest.stub_job_directory 45 _, self.control_file = tempfile.mkstemp() 46 47 def tearDown(self): 48 """Cleanup the test control file.""" 49 os.remove(self.control_file) 50 51 def call_init(self): 52 # TODO(jadmanski): refactor more of the __init__ code to not need to 53 # stub out countless random APIs 54 self.god.stub_with(server_job.os, 'mkdir', lambda p: None) 55 class manager: 56 pass 57 self.god.stub_with(server_job.logging_manager, 'get_logging_manager', 58 lambda *a,**k: manager()) 59 class sysi: 60 log_per_reboot_data = lambda self: None 61 self.god.stub_with(server_job.sysinfo, 'sysinfo', lambda r: sysi()) 62 63 self.job.__init__( 64 self.control_file, 65 (), 66 None, 67 'job_label', 68 'auser', 69 ['mach1', 'mach2'], 70 [ 71 {'hostname': 'mach1'}, # Incomplete machine_dict. 72 {'hostname': 'mach2'}, # Incomplete machine_dict. 73 ], 74 ) 75 self.god.unstub_all() 76 77 78 class WarningManagerTest(unittest.TestCase): 79 def test_never_disabled(self): 80 manager = server_job.warning_manager() 81 self.assertEqual(manager.is_valid(10, "MSGTYPE"), True) 82 83 84 def test_only_enabled(self): 85 manager = server_job.warning_manager() 86 manager.enable_warnings("MSGTYPE", lambda: 10) 87 self.assertEqual(manager.is_valid(20, "MSGTYPE"), True) 88 89 90 def test_disabled_once(self): 91 manager = server_job.warning_manager() 92 manager.disable_warnings("MSGTYPE", lambda: 10) 93 self.assertEqual(manager.is_valid(5, "MSGTYPE"), True) 94 self.assertEqual(manager.is_valid(15, "MSGTYPE"), False) 95 96 97 def test_disable_and_enabled(self): 98 manager = server_job.warning_manager() 99 manager.disable_warnings("MSGTYPE", lambda: 10) 100 manager.enable_warnings("MSGTYPE", lambda: 20) 101 self.assertEqual(manager.is_valid(15, "MSGTYPE"), False) 102 self.assertEqual(manager.is_valid(25, "MSGTYPE"), True) 103 104 105 def test_disabled_changes_is_valid(self): 106 manager = server_job.warning_manager() 107 self.assertEqual(manager.is_valid(15, "MSGTYPE"), True) 108 manager.disable_warnings("MSGTYPE", lambda: 10) 109 self.assertEqual(manager.is_valid(15, "MSGTYPE"), False) 110 111 112 def test_multiple_disabled_calls(self): 113 manager = server_job.warning_manager() 114 manager.disable_warnings("MSGTYPE", lambda: 10) 115 manager.disable_warnings("MSGTYPE", lambda: 20) 116 manager.enable_warnings("MSGTYPE", lambda: 30) 117 self.assertEqual(manager.is_valid(15, "MSGTYPE"), False) 118 self.assertEqual(manager.is_valid(25, "MSGTYPE"), False) 119 self.assertEqual(manager.is_valid(35, "MSGTYPE"), True) 120 121 122 def test_multiple_types(self): 123 manager = server_job.warning_manager() 124 manager.disable_warnings("MSGTYPE1", lambda: 10) 125 manager.disable_warnings("MSGTYPE2", lambda: 20) 126 manager.enable_warnings("MSGTYPE2", lambda: 30) 127 self.assertEqual(manager.is_valid(15, "MSGTYPE1"), False) 128 self.assertEqual(manager.is_valid(15, "MSGTYPE2"), True) 129 self.assertEqual(manager.is_valid(25, "MSGTYPE1"), False) 130 self.assertEqual(manager.is_valid(25, "MSGTYPE2"), False) 131 self.assertEqual(manager.is_valid(35, "MSGTYPE1"), False) 132 self.assertEqual(manager.is_valid(35, "MSGTYPE2"), True) 133 134 135 if __name__ == "__main__": 136 unittest.main() 137
