1 # this doctest contains tests for miscellaneous features of the RPC interface 2 # that would clutter the main rpc_test 3 4 # setup 5 >>> from autotest_lib.frontend.afe import rpc_interface 6 7 >>> rpc_interface.add_profiler(name='oprofile') 8 1 9 >>> rpc_interface.add_profiler(name='iostat') 10 2 11 12 # profiler support in control file generation 13 >>> cf_info = rpc_interface.generate_control_file( 14 ... tests=['sleeptest'], 15 ... profilers=['oprofile', 'iostat']) 16 >>> print cf_info['control_file'] 17 def step_init(): 18 job.next_step('step0') 19 job.next_step('step1') 20 job.next_step('step2') 21 job.next_step('step3') 22 job.next_step('step4') 23 <BLANKLINE> 24 def step0(): 25 job.profilers.add('oprofile') 26 <BLANKLINE> 27 def step1(): 28 job.profilers.add('iostat') 29 <BLANKLINE> 30 def step2(): 31 job.run_test('testname') 32 <BLANKLINE> 33 def step3(): 34 job.profilers.delete('oprofile') 35 <BLANKLINE> 36 def step4(): 37 job.profilers.delete('iostat') 38 39 # profile_only=False support in control file generation 40 >>> cf_info = rpc_interface.generate_control_file( 41 ... tests=['sleeptest'], 42 ... profilers=['oprofile'], 43 ... profile_only=False) 44 >>> print cf_info['control_file'] 45 def step_init(): 46 job.next_step('step0') 47 job.next_step('step1') 48 job.next_step('step2') 49 job.next_step('step3') 50 <BLANKLINE> 51 def step0(): 52 job.default_profile_only = False 53 <BLANKLINE> 54 def step1(): 55 job.profilers.add('oprofile') 56 <BLANKLINE> 57 def step2(): 58 job.run_test('testname') 59 <BLANKLINE> 60 def step3(): 61 job.profilers.delete('oprofile') 62 63 # profile_only=True support in control file generation 64 >>> cf_info = rpc_interface.generate_control_file( 65 ... tests=['sleeptest'], 66 ... profilers=['iostat'], 67 ... profile_only=True) 68 >>> print cf_info['control_file'] 69 def step_init(): 70 job.next_step('step0') 71 job.next_step('step1') 72 job.next_step('step2') 73 job.next_step('step3') 74 <BLANKLINE> 75 def step0(): 76 job.default_profile_only = True 77 <BLANKLINE> 78 def step1(): 79 job.profilers.add('iostat') 80 <BLANKLINE> 81 def step2(): 82 job.run_test('testname') 83 <BLANKLINE> 84 def step3(): 85 job.profilers.delete('iostat') 86 87 # test preserving of the kernel config file setting 88 >>> cf_info = rpc_interface.generate_control_file( 89 ... tests=['sleeptest'], 90 ... kernel=[{'version': '2.6.18', 'config_file': 'foo/bar'}]) 91 >>> print cf_info['control_file'] #doctest: +NORMALIZE_WHITESPACE 92 kernel_list = [{'version': '2.6.18', 'config_file': 'foo/bar'}] 93 def step_init(): 94 for kernel_info in kernel_list: 95 job.next_step(boot_kernel, kernel_info) 96 job.next_step(step_test, kernel_info['version']) 97 if len(kernel_list) > 1: 98 job.use_sequence_number = True # include run numbers in directory names 99 def boot_kernel(kernel_info): 100 # remove kernels (and associated data) not referenced by the bootloader 101 for host in job.hosts: 102 host.cleanup_kernels() 103 testkernel = job.kernel(kernel_info['version']) 104 if kernel_info['config_file']: 105 testkernel.config(kernel_info['config_file']) 106 testkernel.build() 107 testkernel.install() 108 cmdline = ' '.join((kernel_info.get('cmdline', ''), '')) 109 testkernel.boot(args=cmdline) 110 def step_test(kernel_version): 111 global kernel 112 kernel = kernel_version # Set the global in case anyone is using it. 113 if len(kernel_list) > 1: 114 # this is local to a machine, safe to assume there's only one host 115 host, = job.hosts 116 job.automatic_test_tag = host.get_kernel_ver() 117 job.next_step('step0') 118 def step0(): 119 job.run_test('testname') 120 121 # upload_kernel_config requires server side test 122 >>> rpc_interface.generate_control_file(tests=['sleeptest'], 123 ... kernel=[{'version': '2.6.18', 'config_file': 'foo/bar'}], 124 ... upload_kernel_config=True) 125 Traceback (most recent call last): 126 ValidationError: {'upload_kernel_config': 'Cannot use upload_kernel_config with client side tests'} 127 128 # server-side control file generation 129 >>> rpc_interface.modify_test('sleeptest', test_type='Server') 130 >>> cf_info = rpc_interface.generate_control_file(tests=['sleeptest'], 131 ... kernel=[{'version': '2.6.18'}, {'version': '2.6.22'}]) 132 >>> print cf_info['control_file'] #doctest: +NORMALIZE_WHITESPACE 133 kernel_list = [{'version': '2.6.18', 'config_file': None}, {'version': '2.6.22', 'config_file': None}] 134 kernel_install_control = """ 135 kernel_list = %(client_kernel_list)s 136 def step_init(): 137 for kernel_info in kernel_list: 138 job.next_step(boot_kernel, kernel_info) 139 job.next_step(step_test, kernel_info['version']) 140 if len(kernel_list) > 1: 141 job.use_sequence_number = True # include run numbers in directory names 142 def boot_kernel(kernel_info): 143 # remove kernels (and associated data) not referenced by the bootloader 144 for host in job.hosts: 145 host.cleanup_kernels() 146 testkernel = job.kernel(kernel_info['version']) 147 if kernel_info['config_file']: 148 testkernel.config(kernel_info['config_file']) 149 testkernel.build() 150 testkernel.install() 151 cmdline = ' '.join((kernel_info.get('cmdline', ''), '')) 152 testkernel.boot(args=cmdline) 153 def step_test(kernel_version): 154 global kernel 155 kernel = kernel_version # Set the global in case anyone is using it. 156 if len(kernel_list) > 1: 157 # this is local to a machine, safe to assume there's only one host 158 host, = job.hosts 159 job.automatic_test_tag = host.get_kernel_ver() 160 pass 161 """ 162 from autotest_lib.client.common_lib import error 163 at = autotest.Autotest() 164 def install_kernel(machine, kernel_info): 165 host = hosts.create_host(machine) 166 at.install(host=host) 167 at.run(kernel_install_control % 168 {'client_kernel_list': repr([kernel_info])}, host=host) 169 num_machines_required = len(machines) 170 if len(machines) > 4: 171 # Allow a large multi-host tests to proceed despite a couple of hosts 172 # failing to properly install the desired kernel (exclude those hosts). 173 # TODO(gps): Figure out how to get and use SYNC_COUNT here. It is defined 174 # within some control files and will end up inside of stepN functions below. 175 num_machines_required = len(machines) - 2 176 def step_init(): 177 # a host object we use solely for the purpose of finding out the booted 178 # kernel version, we use machines[0] since we already check that the same 179 # kernel has been booted on all machines 180 if len(kernel_list) > 1: 181 kernel_host = hosts.create_host(machines[0]) 182 for kernel_info in kernel_list: 183 func = lambda machine: install_kernel(machine, kernel_info) 184 good_machines = job.parallel_on_machines(func, machines) 185 if len(good_machines) < num_machines_required: 186 raise error.TestError( 187 "kernel installed on only %d of %d machines." 188 % (len(good_machines), num_machines_required)) 189 # Replace the machines list that step_test() will use with the 190 # ones that successfully installed the kernel. 191 machines[:] = good_machines 192 # have server_job.run_test() automatically add the kernel version as 193 # a suffix to the test name otherwise we cannot run the same test on 194 # different kernel versions 195 if len(kernel_list) > 1: 196 job.automatic_test_tag = kernel_host.get_kernel_ver() 197 step_test() 198 def step_test(): 199 step0() 200 def step0(): 201 job.run_test('testname') 202 step_init() 203 >>> cf_info['is_server'] 204 True 205 206 # server-side control file generation with kernel config upload code 207 >>> cf_info = rpc_interface.generate_control_file(tests=['sleeptest'], 208 ... kernel=[{'version': '2.6.18', 'config_file': 'foo/bar'}], 209 ... upload_kernel_config=True) 210 >>> print cf_info['control_file'] #doctest: +NORMALIZE_WHITESPACE 211 kernel_list = [{'version': '2.6.18', 'config_file': 'foo/bar'}] 212 kernel_install_control = """ 213 kernel_list = %(client_kernel_list)s 214 def step_init(): 215 for kernel_info in kernel_list: 216 job.next_step(boot_kernel, kernel_info) 217 job.next_step(step_test, kernel_info['version']) 218 if len(kernel_list) > 1: 219 job.use_sequence_number = True # include run numbers in directory names 220 def boot_kernel(kernel_info): 221 # remove kernels (and associated data) not referenced by the bootloader 222 for host in job.hosts: 223 host.cleanup_kernels() 224 testkernel = job.kernel(kernel_info['version']) 225 if kernel_info['config_file']: 226 testkernel.config(kernel_info['config_file']) 227 testkernel.build() 228 testkernel.install() 229 cmdline = ' '.join((kernel_info.get('cmdline', ''), '')) 230 testkernel.boot(args=cmdline) 231 def step_test(kernel_version): 232 global kernel 233 kernel = kernel_version # Set the global in case anyone is using it. 234 if len(kernel_list) > 1: 235 # this is local to a machine, safe to assume there's only one host 236 host, = job.hosts 237 job.automatic_test_tag = host.get_kernel_ver() 238 pass 239 """ 240 from autotest_lib.client.common_lib import error 241 at = autotest.Autotest() 242 def upload_kernel_config(host, kernel_info): 243 """ 244 If the kernel_info['config_file'] is a URL it will be downloaded 245 locally and then uploaded to the client and a copy of the original 246 dictionary with the new path to the config file will be returned. 247 If the config file is not a URL the function returns the original 248 dictionary. 249 """ 250 import os 251 from autotest_lib.client.common_lib import autotemp, utils 252 config_orig = kernel_info.get('config_file') 253 # if the file is not an URL then we assume it's a local client path 254 if not config_orig or not utils.is_url(config_orig): 255 return kernel_info 256 # download it locally (on the server) and send it to the client 257 config_tmp = autotemp.tempfile('kernel_config_upload', dir=job.tmpdir) 258 try: 259 utils.urlretrieve(config_orig, config_tmp.name) 260 config_new = os.path.join(host.get_autodir(), 'tmp', 261 os.path.basename(config_orig)) 262 host.send_file(config_tmp.name, config_new) 263 finally: 264 config_tmp.clean() 265 return dict(kernel_info, config_file=config_new) 266 def install_kernel(machine, kernel_info): 267 host = hosts.create_host(machine) 268 at.install(host=host) 269 kernel_info = upload_kernel_config(host, kernel_info) 270 at.run(kernel_install_control % 271 {'client_kernel_list': repr([kernel_info])}, host=host) 272 num_machines_required = len(machines) 273 if len(machines) > 4: 274 # Allow a large multi-host tests to proceed despite a couple of hosts 275 # failing to properly install the desired kernel (exclude those hosts). 276 # TODO(gps): Figure out how to get and use SYNC_COUNT here. It is defined 277 # within some control files and will end up inside of stepN functions below. 278 num_machines_required = len(machines) - 2 279 def step_init(): 280 # a host object we use solely for the purpose of finding out the booted 281 # kernel version, we use machines[0] since we already check that the same 282 # kernel has been booted on all machines 283 if len(kernel_list) > 1: 284 kernel_host = hosts.create_host(machines[0]) 285 for kernel_info in kernel_list: 286 func = lambda machine: install_kernel(machine, kernel_info) 287 good_machines = job.parallel_on_machines(func, machines) 288 if len(good_machines) < num_machines_required: 289 raise error.TestError( 290 "kernel installed on only %d of %d machines." 291 % (len(good_machines), num_machines_required)) 292 # Replace the machines list that step_test() will use with the 293 # ones that successfully installed the kernel. 294 machines[:] = good_machines 295 # have server_job.run_test() automatically add the kernel version as 296 # a suffix to the test name otherwise we cannot run the same test on 297 # different kernel versions 298 if len(kernel_list) > 1: 299 job.automatic_test_tag = kernel_host.get_kernel_ver() 300 step_test() 301 def step_test(): 302 step0() 303 def step0(): 304 job.run_test('testname') 305 step_init() 306 307 # test that multiline quoted strings are not indented 308 >>> import common 309 >>> from autotest_lib.frontend.afe import test, control_file 310 >>> import os 311 >>> control_path = os.path.join(os.path.dirname(test.__file__), 312 ... 'doctests', 'test.control.3') 313 >>> control_path = os.path.abspath(control_path) 314 >>> class FakeTest(object): 315 ... path = control_path 316 ... 317 >>> print control_file.generate_control([FakeTest()], is_server=True) #doctest: +NORMALIZE_WHITESPACE 318 def step_init(): 319 step0() 320 def step0(): 321 client_code = """ 322 some content\"""quoted content\""" 323 '''other quoted content\"""''' 324 \\""" 325 client_code2 = ''' 326 some content\'''quoted content\''' 327 """other quoted content\'''""" 328 \\''' 329 job.run_test('testname') 330 step_init() 331
