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 tap_report = None 88 output_dir = False 89 90 self.job.__init__(options) 91 92 93 class dummy(object): 94 """A simple placeholder for attributes""" 95 pass 96 97 98 class first_line_comparator(mock.argument_comparator): 99 def __init__(self, first_line): 100 self.first_line = first_line 101 102 103 def is_satisfied_by(self, parameter): 104 return self.first_line == parameter.splitlines()[0] 105 106 107 class test_setup_job(unittest.TestCase): 108 def setUp(self): 109 # make god 110 self.god = mock.mock_god() 111 112 # need to set some environ variables 113 self.autodir = "autodir" 114 os.environ['AUTODIR'] = self.autodir 115 116 # set up some variables 117 self.jobtag = "jobtag" 118 119 # get rid of stdout and logging 120 sys.stdout = StringIO.StringIO() 121 logging_manager.configure_logging(logging_config.TestingConfig()) 122 logging.disable(logging.CRITICAL) 123 def dummy_configure_logging(*args, **kwargs): 124 pass 125 self.god.stub_with(logging_manager, 'configure_logging', 126 dummy_configure_logging) 127 real_get_logging_manager = logging_manager.get_logging_manager 128 def get_logging_manager_no_fds(manage_stdout_and_stderr=False, 129 redirect_fds=False): 130 return real_get_logging_manager(manage_stdout_and_stderr, False) 131 self.god.stub_with(logging_manager, 'get_logging_manager', 132 get_logging_manager_no_fds) 133 134 # stub out some stuff 135 self.god.stub_function(os.path, 'exists') 136 self.god.stub_function(os.path, 'isdir') 137 self.god.stub_function(os, 'makedirs') 138 self.god.stub_function(os, 'mkdir') 139 self.god.stub_function(os, 'remove') 140 self.god.stub_function(shutil, 'rmtree') 141 self.god.stub_function(shutil, 'copyfile') 142 self.god.stub_function(setup_job, 'open') 143 self.god.stub_function(utils, 'system') 144 145 self.god.stub_class_method(job.base_client_job, 146 '_cleanup_debugdir_files') 147 self.god.stub_class_method(job.base_client_job, '_cleanup_results_dir') 148 149 self.god.stub_with(base_job.job_directory, '_ensure_valid', 150 lambda *_: None) 151 152 153 def tearDown(self): 154 sys.stdout = sys.__stdout__ 155 self.god.unstub_all() 156 157 158 def _setup_pre_record_init(self): 159 resultdir = os.path.join(self.autodir, 'results', self.jobtag) 160 tmpdir = os.path.join(self.autodir, 'tmp') 161 job.base_client_job._cleanup_debugdir_files.expect_call() 162 job.base_client_job._cleanup_results_dir.expect_call() 163 164 return resultdir 165 166 def construct_job(self): 167 # will construct class instance using __new__ 168 self.job = setup_job.setup_job.__new__(setup_job.setup_job) 169 170 resultdir = self._setup_pre_record_init() 171 172 # finish constructor 173 options = dummy() 174 options.tag = self.jobtag 175 options.log = False 176 options.verbose = False 177 options.hostname = 'localhost' 178 options.user = 'my_user' 179 options.tap_report = None 180 options.output_dir = False 181 self.job.__init__(options) 182 183 # check 184 self.god.check_playback() 185 186 187 def get_partition_mock(self, devname): 188 """ 189 Create a mock of a partition object and return it. 190 """ 191 class mock(object): 192 device = devname 193 get_mountpoint = self.god.create_mock_function('get_mountpoint') 194 return mock 195 196 197 def test_constructor_first_run(self): 198 self.construct_job() 199 200 201 def test_constructor_continuation(self): 202 self.construct_job() 203 204 205 def test_setup_dirs_raise(self): 206 self.construct_job() 207 208 # setup 209 results_dir = 'foo' 210 tmp_dir = 'bar' 211 212 # record 213 os.path.exists.expect_call(tmp_dir).and_return(True) 214 os.path.isdir.expect_call(tmp_dir).and_return(False) 215 216 # test 217 self.assertRaises(ValueError, self.job.setup_dirs, results_dir, tmp_dir) 218 self.god.check_playback() 219 220 221 def test_setup_dirs(self): 222 self.construct_job() 223 224 # setup 225 results_dir1 = os.path.join(self.job.resultdir, 'build') 226 results_dir2 = os.path.join(self.job.resultdir, 'build.2') 227 results_dir3 = os.path.join(self.job.resultdir, 'build.3') 228 tmp_dir = 'bar' 229 230 # record 231 os.path.exists.expect_call(tmp_dir).and_return(False) 232 os.mkdir.expect_call(tmp_dir) 233 os.path.isdir.expect_call(tmp_dir).and_return(True) 234 os.path.exists.expect_call(results_dir1).and_return(True) 235 os.path.exists.expect_call(results_dir2).and_return(True) 236 os.path.exists.expect_call(results_dir3).and_return(False) 237 os.path.exists.expect_call(results_dir3).and_return(False) 238 os.mkdir.expect_call(results_dir3) 239 240 # test 241 self.assertEqual(self.job.setup_dirs(None, tmp_dir), 242 (results_dir3, tmp_dir)) 243 self.god.check_playback() 244 245 246 if __name__ == "__main__": 247 unittest.main() 248
