1 #!/usr/bin/python 2 # Copyright 2015 The Chromium OS Authors. All rights reserved. 3 # Use of this source code is governed by a BSD-style license that can be 4 # found in the LICENSE file. 5 # pylint: disable-msg=C0111 6 7 import os, unittest 8 import mox 9 import common 10 import subprocess 11 import types 12 from autotest_lib.server import utils 13 from autotest_lib.server.cros.dynamic_suite import constants 14 from autotest_lib.site_utils import test_runner_utils 15 16 17 class StartsWithList(mox.Comparator): 18 def __init__(self, start_of_list): 19 """Mox comparator which returns True if the argument 20 to the mocked function is a list that begins with the elements 21 in start_of_list. 22 """ 23 self._lhs = start_of_list 24 25 def equals(self, rhs): 26 if len(rhs)<len(self._lhs): 27 return False 28 for (x, y) in zip(self._lhs, rhs): 29 if x != y: 30 return False 31 return True 32 33 34 class ContainsSublist(mox.Comparator): 35 def __init__(self, sublist): 36 """Mox comparator which returns True if the argument 37 to the mocked function is a list that contains sublist 38 as a sub-list. 39 """ 40 self._sublist = sublist 41 42 def equals(self, rhs): 43 n = len(self._sublist) 44 if len(rhs)<n: 45 return False 46 return any((self._sublist == rhs[i:i+n]) 47 for i in xrange(len(rhs) - n + 1)) 48 49 50 class TestRunnerUnittests(unittest.TestCase): 51 52 def test_fetch_local_suite(self): 53 # Deferred until fetch_local_suite knows about non-local builds. 54 pass 55 56 def test_get_predicate_for_test_arg(self): 57 # Assert the type signature of get_predicate_for_test(...) 58 # Because control.test_utils_wrapper calls this function, 59 # it is imperative for backwards compatilbility that 60 # the return type of the tested function does not change. 61 tests = ['dummy_test', 'e:name_expression', 'f:expression', 62 'suite:suitename'] 63 for test in tests: 64 pred, desc = test_runner_utils.get_predicate_for_test_arg(test) 65 self.assertTrue(isinstance(pred, types.FunctionType)) 66 self.assertTrue(isinstance(desc, str)) 67 68 def test_run_job(self): 69 class Object(): 70 pass 71 72 autotest_path = 'htap_tsetotua' 73 autoserv_command = os.path.join(autotest_path, 'server', 'autoserv') 74 remote = 'etomer' 75 results_dir = '/tmp/fakeresults' 76 fast_mode = False 77 job1_results_dir = '/tmp/fakeresults/results-1-gilbert' 78 job2_results_dir = '/tmp/fakeresults/results-2-sullivan' 79 args = 'matey' 80 expected_args_sublist = ['--args', args] 81 experimental_keyval = {constants.JOB_EXPERIMENTAL_KEY: False} 82 self.mox = mox.Mox() 83 84 # Create some dummy job objects. 85 job1 = Object() 86 job2 = Object() 87 setattr(job1, 'control_type', 'cLiEnT') 88 setattr(job1, 'control_file', 'c1') 89 setattr(job1, 'id', 1) 90 setattr(job1, 'name', 'gilbert') 91 setattr(job1, 'keyvals', experimental_keyval) 92 93 setattr(job2, 'control_type', 'Server') 94 setattr(job2, 'control_file', 'c2') 95 setattr(job2, 'id', 2) 96 setattr(job2, 'name', 'sullivan') 97 setattr(job2, 'keyvals', experimental_keyval) 98 99 id_digits = 1 100 101 # Stub out subprocess.Popen and wait calls. 102 # Make them expect correct arguments. 103 def fake_readline(): 104 return b'' 105 mock_process_1 = self.mox.CreateMock(subprocess.Popen) 106 mock_process_2 = self.mox.CreateMock(subprocess.Popen) 107 fake_stdout = self.mox.CreateMock(file) 108 fake_returncode = 0 109 mock_process_1.stdout = fake_stdout 110 mock_process_1.returncode = fake_returncode 111 mock_process_2.stdout = fake_stdout 112 mock_process_2.returncode = fake_returncode 113 114 self.mox.StubOutWithMock(os, 'makedirs') 115 self.mox.StubOutWithMock(utils, 'write_keyval') 116 self.mox.StubOutWithMock(subprocess, 'Popen') 117 118 os.makedirs(job1_results_dir) 119 utils.write_keyval(job1_results_dir, experimental_keyval) 120 arglist_1 = [autoserv_command, '-p', '-r', job1_results_dir, 121 '-m', remote, '--no_console_prefix', '-l', 'gilbert', 122 '-c'] 123 subprocess.Popen(mox.And(StartsWithList(arglist_1), 124 ContainsSublist(expected_args_sublist)), 125 stdout=subprocess.PIPE, 126 stderr=subprocess.STDOUT 127 ).AndReturn(mock_process_1) 128 mock_process_1.stdout.readline().AndReturn(b'') 129 mock_process_1.wait().AndReturn(0) 130 131 os.makedirs(job2_results_dir) 132 utils.write_keyval(job2_results_dir, experimental_keyval) 133 arglist_2 = [autoserv_command, '-p', '-r', job2_results_dir, 134 '-m', remote, '--no_console_prefix', '-l', 'sullivan', 135 '-s'] 136 subprocess.Popen(mox.And(StartsWithList(arglist_2), 137 ContainsSublist(expected_args_sublist)), 138 stdout=subprocess.PIPE, 139 stderr=subprocess.STDOUT 140 ).AndReturn(mock_process_2) 141 mock_process_2.stdout.readline().AndReturn(b'') 142 mock_process_2.wait().AndReturn(0) 143 144 # Test run_job. 145 self.mox.ReplayAll() 146 code, job_res = test_runner_utils.run_job( 147 job1, remote, autotest_path,results_dir, fast_mode, id_digits, 148 0, None, args) 149 self.assertEqual(job_res, job1_results_dir) 150 self.assertEqual(code, 0) 151 code, job_res = test_runner_utils.run_job( 152 job2, remote, autotest_path, results_dir, fast_mode, id_digits, 153 0, None, args) 154 155 self.assertEqual(job_res, job2_results_dir) 156 self.assertEqual(code, 0) 157 self.mox.UnsetStubs() 158 self.mox.VerifyAll() 159 self.mox.ResetAll() 160 161 def test_perform_local_run(self): 162 afe = test_runner_utils.setup_local_afe() 163 autotest_path = 'ottotest_path' 164 suite_name = 'sweet_name' 165 test_arg = 'suite:' + suite_name 166 remote = 'remoat' 167 build = 'bild' 168 board = 'bored' 169 fast_mode = False 170 suite_control_files = ['c1', 'c2', 'c3', 'c4'] 171 results_dir = '/tmp/test_that_results_fake' 172 id_digits = 1 173 ssh_verbosity = 2 174 ssh_options = '-F /dev/null -i /dev/null' 175 args = 'matey' 176 ignore_deps = False 177 178 # Fake suite objects that will be returned by fetch_local_suite 179 class fake_suite(object): 180 def __init__(self, suite_control_files, hosts): 181 self._suite_control_files = suite_control_files 182 self._hosts = hosts 183 184 def schedule(self, *args, **kwargs): 185 for control_file in self._suite_control_files: 186 afe.create_job(control_file, hosts=self._hosts) 187 188 # Mock out scheduling of suite and running of jobs. 189 self.mox = mox.Mox() 190 191 self.mox.StubOutWithMock(test_runner_utils, 'fetch_local_suite') 192 test_runner_utils.fetch_local_suite(autotest_path, mox.IgnoreArg(), 193 afe, test_arg=test_arg, remote=remote, build=build, 194 board=board, results_directory=results_dir, 195 no_experimental=False, 196 ignore_deps=ignore_deps 197 ).AndReturn(fake_suite(suite_control_files, [remote])) 198 self.mox.StubOutWithMock(test_runner_utils, 'run_job') 199 self.mox.StubOutWithMock(test_runner_utils, 'run_provisioning_job') 200 self.mox.StubOutWithMock(test_runner_utils, '_auto_detect_labels') 201 202 test_runner_utils._auto_detect_labels(afe, remote) 203 # Test perform_local_run. Enforce that run_provisioning_job, 204 # run_job and _auto_detect_labels are called correctly. 205 test_runner_utils.run_provisioning_job( 206 'cros-version:' + build, remote, autotest_path, 207 results_dir, fast_mode, 208 ssh_verbosity, ssh_options, 209 False, False) 210 211 for control_file in suite_control_files: 212 test_runner_utils.run_job( 213 mox.ContainsAttributeValue('control_file', control_file), 214 remote, autotest_path, results_dir, fast_mode,id_digits, 215 ssh_verbosity, ssh_options,args, False, 216 False, {}).AndReturn((0, '/fake/dir')) 217 self.mox.ReplayAll() 218 test_runner_utils.perform_local_run( 219 afe, autotest_path, ['suite:'+suite_name], remote, fast_mode, 220 build=build, board=board, ignore_deps=False, 221 ssh_verbosity=ssh_verbosity, ssh_options=ssh_options, 222 args=args, results_directory=results_dir) 223 self.mox.UnsetStubs() 224 self.mox.VerifyAll() 225 226 227 if __name__ == '__main__': 228 unittest.main() 229
