1 #!/usr/bin/python 2 #pylint: disable-msg=C0111 3 import logging 4 import os 5 import shutil 6 import StringIO 7 import sys 8 import unittest 9 10 import common 11 from autotest_lib.client.bin import job, setup_job 12 from autotest_lib.client.bin import utils 13 from autotest_lib.client.common_lib import base_job 14 from autotest_lib.client.common_lib import logging_manager, logging_config 15 from autotest_lib.client.common_lib import base_job_unittest 16 from autotest_lib.client.common_lib.test_utils import mock 17 18 19 class setup_job_test_case(unittest.TestCase): 20 """Generic job TestCase class that defines a standard job setUp and 21 tearDown, with some standard stubs.""" 22 23 job_class = setup_job.setup_job 24 25 def setUp(self): 26 self.god = mock.mock_god() 27 self.god.stub_with(setup_job.setup_job, '_get_environ_autodir', 28 classmethod(lambda cls: '/adir')) 29 self.job = self.job_class.__new__(self.job_class) 30 self.job._job_directory = base_job_unittest.stub_job_directory 31 self.job.args = [] 32 33 34 def tearDown(self): 35 self.god.unstub_all() 36 37 38 class test_find_base_directories( 39 base_job_unittest.test_find_base_directories.generic_tests, 40 setup_job_test_case): 41 42 def test_autodir_equals_clientdir(self): 43 autodir, clientdir, _ = self.job._find_base_directories() 44 self.assertEqual(autodir, '/adir') 45 self.assertEqual(clientdir, '/adir') 46 47 48 def test_serverdir_is_none(self): 49 _, _, serverdir = self.job._find_base_directories() 50 self.assertEqual(serverdir, None) 51 52 53 class abstract_test_init(base_job_unittest.test_init.generic_tests): 54 """Generic client job mixin used when defining variations on the 55 job.__init__ generic tests.""" 56 PUBLIC_ATTRIBUTES = ( 57 base_job_unittest.test_init.generic_tests.PUBLIC_ATTRIBUTES 58 - set(['bootloader', 'control', 'drop_caches', 59 'drop_caches_between_iterations', 'harness', 'hosts', 'logging', 60 'machines', 'num_tests_failed', 'num_tests_run', 'profilers', 61 'sysinfo', 'user', 'warning_loggers', 'warning_manager'])) 62 63 64 class test_init_minimal_options(abstract_test_init, setup_job_test_case): 65 def call_init(self): 66 # TODO(jadmanski): refactor more of the __init__ code to not need to 67 # stub out countless random APIs 68 self.god.stub_function_to_return(setup_job.os, 'mkdir', None) 69 self.god.stub_function_to_return(setup_job.os.path, 'exists', True) 70 self.god.stub_function_to_return(self.job, '_load_state', None) 71 self.god.stub_function_to_return(setup_job.logging_manager, 72 'configure_logging', None) 73 class manager: 74 def start_logging(self): 75 return None 76 self.god.stub_function_to_return(setup_job.logging_manager, 77 'get_logging_manager', manager()) 78 79 class options: 80 tag = '' 81 verbose = False 82 cont = False 83 harness = 'stub' 84 hostname = None 85 user = None 86 log = False 87 output_dir = False 88 89 self.job.__init__(options) 90 91 92 class dummy(object): 93 """A simple placeholder for attributes""" 94 pass 95 96 97 class first_line_comparator(mock.argument_comparator): 98 def __init__(self, first_line): 99 self.first_line = first_line 100 101 102 def is_satisfied_by(self, parameter): 103 return self.first_line == parameter.splitlines()[0] 104 105 106 class test_setup_job(unittest.TestCase): 107 def setUp(self): 108 # make god 109 self.god = mock.mock_god() 110 111 # need to set some environ variables 112 self.autodir = "autodir" 113 os.environ['AUTODIR'] = self.autodir 114 115 # set up some variables 116 self.jobtag = "jobtag" 117 118 # get rid of stdout and logging 119 sys.stdout = StringIO.StringIO() 120 logging_manager.configure_logging(logging_config.TestingConfig()) 121 logging.disable(logging.CRITICAL) 122 def dummy_configure_logging(*args, **kwargs): 123 pass 124 self.god.stub_with(logging_manager, 'configure_logging', 125 dummy_configure_logging) 126 real_get_logging_manager = logging_manager.get_logging_manager 127 def get_logging_manager_no_fds(manage_stdout_and_stderr=False, 128 redirect_fds=False): 129 return real_get_logging_manager(manage_stdout_and_stderr, False) 130 self.god.stub_with(logging_manager, 'get_logging_manager', 131 get_logging_manager_no_fds) 132 133 # stub out some stuff 134 self.god.stub_function(os.path, 'exists') 135 self.god.stub_function(os.path, 'isdir') 136 self.god.stub_function(os, 'makedirs') 137 self.god.stub_function(os, 'mkdir') 138 self.god.stub_function(os, 'remove') 139 self.god.stub_function(shutil, 'rmtree') 140 self.god.stub_function(shutil, 'copyfile') 141 self.god.stub_function(setup_job, 'open') 142 self.god.stub_function(utils, 'system') 143 144 self.god.stub_class_method(job.base_client_job, 145 '_cleanup_debugdir_files') 146 self.god.stub_class_method(job.base_client_job, '_cleanup_results_dir') 147 148 self.god.stub_with(base_job.job_directory, '_ensure_valid', 149 lambda *_: None) 150 151 152 def tearDown(self): 153 sys.stdout = sys.__stdout__ 154 self.god.unstub_all() 155 156 157 def _setup_pre_record_init(self): 158 resultdir = os.path.join(self.autodir, 'results', self.jobtag) 159 tmpdir = os.path.join(self.autodir, 'tmp') 160 job.base_client_job._cleanup_debugdir_files.expect_call() 161 job.base_client_job._cleanup_results_dir.expect_call() 162 163 return resultdir 164 165 def construct_job(self): 166 # will construct class instance using __new__ 167 self.job = setup_job.setup_job.__new__(setup_job.setup_job) 168 169 resultdir = self._setup_pre_record_init() 170 171 # finish constructor 172 options = dummy() 173 options.tag = self.jobtag 174 options.log = False 175 options.verbose = False 176 options.hostname = 'localhost' 177 options.user = 'my_user' 178 options.output_dir = False 179 self.job.__init__(options) 180 181 # check 182 self.god.check_playback() 183 184 185 def get_partition_mock(self, devname): 186 """ 187 Create a mock of a partition object and return it. 188 """ 189 class mock(object): 190 device = devname 191 get_mountpoint = self.god.create_mock_function('get_mountpoint') 192 return mock 193 194 195 def test_constructor_first_run(self): 196 self.construct_job() 197 198 199 def test_constructor_continuation(self): 200 self.construct_job() 201 202 203 def test_setup_dirs_raise(self): 204 self.construct_job() 205 206 # setup 207 results_dir = 'foo' 208 tmp_dir = 'bar' 209 210 # record 211 os.path.exists.expect_call(tmp_dir).and_return(True) 212 os.path.isdir.expect_call(tmp_dir).and_return(False) 213 214 # test 215 self.assertRaises(ValueError, self.job.setup_dirs, results_dir, tmp_dir) 216 self.god.check_playback() 217 218 219 def test_setup_dirs(self): 220 self.construct_job() 221 222 # setup 223 results_dir1 = os.path.join(self.job.resultdir, 'build') 224 results_dir2 = os.path.join(self.job.resultdir, 'build.2') 225 results_dir3 = os.path.join(self.job.resultdir, 'build.3') 226 tmp_dir = 'bar' 227 228 # record 229 os.path.exists.expect_call(tmp_dir).and_return(False) 230 os.mkdir.expect_call(tmp_dir) 231 os.path.isdir.expect_call(tmp_dir).and_return(True) 232 os.path.exists.expect_call(results_dir1).and_return(True) 233 os.path.exists.expect_call(results_dir2).and_return(True) 234 os.path.exists.expect_call(results_dir3).and_return(False) 235 os.path.exists.expect_call(results_dir3).and_return(False) 236 os.mkdir.expect_call(results_dir3) 237 238 # test 239 self.assertEqual(self.job.setup_dirs(None, tmp_dir), 240 (results_dir3, tmp_dir)) 241 self.god.check_playback() 242 243 244 if __name__ == "__main__": 245 unittest.main() 246
