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