1#!/usr/bin/python2 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 7from __future__ import absolute_import 8from __future__ import division 9from __future__ import print_function 10import os, unittest 11import mox 12import common 13import shutil 14import tempfile 15import types 16from autotest_lib.client.common_lib import control_data 17from autotest_lib.server.cros.dynamic_suite import control_file_getter 18from autotest_lib.server.cros.dynamic_suite import suite as suite_module 19from autotest_lib.server.hosts import host_info 20from autotest_lib.site_utils import test_runner_utils 21from six.moves import range 22from six.moves import zip 23 24 25class StartsWithList(mox.Comparator): 26 def __init__(self, start_of_list): 27 """Mox comparator which returns True if the argument 28 to the mocked function is a list that begins with the elements 29 in start_of_list. 30 """ 31 self._lhs = start_of_list 32 33 def equals(self, rhs): 34 if len(rhs)<len(self._lhs): 35 return False 36 for (x, y) in zip(self._lhs, rhs): 37 if x != y: 38 return False 39 return True 40 41 42class ContainsSublist(mox.Comparator): 43 def __init__(self, sublist): 44 """Mox comparator which returns True if the argument 45 to the mocked function is a list that contains sublist 46 as a sub-list. 47 """ 48 self._sublist = sublist 49 50 def equals(self, rhs): 51 n = len(self._sublist) 52 if len(rhs)<n: 53 return False 54 return any((self._sublist == rhs[i:i+n]) 55 for i in range(len(rhs) - n + 1)) 56 57class DummyJob(object): 58 def __init__(self, id=1): 59 self.id = id 60 61class TestRunnerUnittests(mox.MoxTestBase): 62 63 def setUp(self): 64 mox.MoxTestBase.setUp(self) 65 66 67 def test_fetch_local_suite(self): 68 # Deferred until fetch_local_suite knows about non-local builds. 69 pass 70 71 72 def _results_directory_from_results_list(self, results_list): 73 """Generate a temp directory filled with provided test results. 74 75 @param results_list: List of results, each result is a tuple of strings 76 (test_name, test_status_message). 77 @returns: Absolute path to the results directory. 78 """ 79 global_dir = tempfile.mkdtemp() 80 for index, (test_name, test_status_message) in enumerate(results_list): 81 dir_name = '-'.join(['results', 82 "%02.f" % (index + 1), 83 test_name]) 84 local_dir = os.path.join(global_dir, dir_name) 85 os.mkdir(local_dir) 86 os.mkdir('%s/debug' % local_dir) 87 with open("%s/status.log" % local_dir, mode='w+') as status: 88 status.write(test_status_message) 89 status.flush() 90 return global_dir 91 92 93 def test_handle_local_result_for_good_test(self): 94 getter = self.mox.CreateMock(control_file_getter.DevServerGetter) 95 getter.get_control_file_list(suite_name=mox.IgnoreArg()).AndReturn([]) 96 job = DummyJob() 97 test = self.mox.CreateMock(control_data.ControlData) 98 test.job_retries = 5 99 self.mox.StubOutWithMock(test_runner_utils.LocalSuite, 100 '_retry_local_result') 101 self.mox.ReplayAll() 102 suite = test_runner_utils.LocalSuite([], "tag", [], None, getter, 103 job_retry=True) 104 suite._retry_handler = suite_module.RetryHandler({job.id: test}) 105 106 #No calls, should not be retried 107 directory = self._results_directory_from_results_list([ 108 ("dummy_Good", "GOOD: nonexistent test completed successfully")]) 109 new_id = suite.handle_local_result( 110 job.id, directory, 111 lambda log_entry, log_in_subdir=False: None) 112 self.assertIsNone(new_id) 113 shutil.rmtree(directory) 114 115 116 def test_handle_local_result_for_bad_test(self): 117 getter = self.mox.CreateMock(control_file_getter.DevServerGetter) 118 getter.get_control_file_list(suite_name=mox.IgnoreArg()).AndReturn([]) 119 job = DummyJob() 120 test = self.mox.CreateMock(control_data.ControlData) 121 test.job_retries = 5 122 self.mox.StubOutWithMock(test_runner_utils.LocalSuite, 123 '_retry_local_result') 124 test_runner_utils.LocalSuite._retry_local_result( 125 job.id, mox.IgnoreArg()).AndReturn(42) 126 self.mox.ReplayAll() 127 suite = test_runner_utils.LocalSuite([], "tag", [], None, getter, 128 job_retry=True) 129 suite._retry_handler = suite_module.RetryHandler({job.id: test}) 130 131 directory = self._results_directory_from_results_list([ 132 ("dummy_Bad", "FAIL")]) 133 new_id = suite.handle_local_result( 134 job.id, directory, 135 lambda log_entry, log_in_subdir=False: None) 136 self.assertIsNotNone(new_id) 137 shutil.rmtree(directory) 138 139 140 def test_generate_report_status_code_success_with_retries(self): 141 global_dir = self._results_directory_from_results_list([ 142 ("dummy_Flaky", "FAIL"), 143 ("dummy_Flaky", "GOOD: nonexistent test completed successfully")]) 144 status_code = test_runner_utils.generate_report( 145 global_dir, just_status_code=True) 146 self.assertEquals(status_code, 0) 147 shutil.rmtree(global_dir) 148 149 150 def test_generate_report_status_code_failure_with_retries(self): 151 global_dir = self._results_directory_from_results_list([ 152 ("dummy_Good", "GOOD: nonexistent test completed successfully"), 153 ("dummy_Bad", "FAIL"), 154 ("dummy_Bad", "FAIL")]) 155 status_code = test_runner_utils.generate_report( 156 global_dir, just_status_code=True) 157 self.assertNotEquals(status_code, 0) 158 shutil.rmtree(global_dir) 159 160 161 def test_get_predicate_for_test_arg(self): 162 # Assert the type signature of get_predicate_for_test(...) 163 # Because control.test_utils_wrapper calls this function, 164 # it is imperative for backwards compatilbility that 165 # the return type of the tested function does not change. 166 tests = ['dummy_test', 'e:name_expression', 'f:expression', 167 'suite:suitename'] 168 for test in tests: 169 pred, desc = test_runner_utils.get_predicate_for_test_arg(test) 170 self.assertTrue(isinstance(pred, types.FunctionType)) 171 self.assertTrue(isinstance(desc, str)) 172 173 def test_perform_local_run(self): 174 afe = test_runner_utils.setup_local_afe() 175 autotest_path = 'ottotest_path' 176 suite_name = 'sweet_name' 177 test_arg = 'suite:' + suite_name 178 remote = 'remoat' 179 build = 'bild' 180 board = 'bored' 181 fast_mode = False 182 suite_control_files = ['c1', 'c2', 'c3', 'c4'] 183 results_dir = '/tmp/test_that_results_fake' 184 id_digits = 1 185 ssh_verbosity = 2 186 ssh_options = '-F /dev/null -i /dev/null' 187 args = 'matey' 188 ignore_deps = False 189 retry = True 190 191 # Fake suite objects that will be returned by fetch_local_suite 192 class fake_suite(object): 193 def __init__(self, suite_control_files, hosts): 194 self._suite_control_files = suite_control_files 195 self._hosts = hosts 196 self._jobs = [] 197 self._jobs_to_tests = {} 198 self.retry_hack = True 199 200 def schedule(self, *args, **kwargs): 201 for control_file in self._suite_control_files: 202 job_id = afe.create_job(control_file, hosts=self._hosts) 203 self._jobs.append(job_id) 204 self._jobs_to_tests[job_id] = control_file 205 206 def handle_local_result(self, job_id, results_dir, logger, 207 **kwargs): 208 if results_dir == "success_directory": 209 return None 210 retries = True 211 if 'retries' in kwargs: 212 retries = kwargs['retries'] 213 if retries and self.retry_hack: 214 self.retry_hack = False 215 else: 216 return None 217 control_file = self._jobs_to_tests.get(job_id) 218 job_id = afe.create_job(control_file, hosts=self._hosts) 219 self._jobs.append(job_id) 220 self._jobs_to_tests[job_id] = control_file 221 return job_id 222 223 @property 224 def jobs(self): 225 return self._jobs 226 227 def test_name_from_job(self, id): 228 return "" 229 230 # Mock out scheduling of suite and running of jobs. 231 self.mox.StubOutWithMock(test_runner_utils, 'fetch_local_suite') 232 test_runner_utils.fetch_local_suite(autotest_path, mox.IgnoreArg(), 233 afe, test_arg=test_arg, remote=remote, build=build, 234 board=board, results_directory=results_dir, 235 no_experimental=False, 236 ignore_deps=ignore_deps, 237 job_retry=retry 238 ).AndReturn(fake_suite(suite_control_files, [remote])) 239 self.mox.StubOutWithMock(test_runner_utils, 'run_job') 240 self.mox.StubOutWithMock(test_runner_utils, 'run_provisioning_job') 241 self.mox.StubOutWithMock(test_runner_utils, '_auto_detect_labels') 242 243 test_runner_utils._auto_detect_labels(afe, remote).AndReturn([]) 244 # Test perform_local_run. Enforce that run_provisioning_job, 245 # run_job and _auto_detect_labels are called correctly. 246 test_runner_utils.run_provisioning_job( 247 'cros-version:' + build, 248 remote, 249 mox.IsA(host_info.HostInfo), 250 autotest_path, 251 results_dir, 252 fast_mode, 253 ssh_verbosity, 254 ssh_options, 255 False, 256 False, 257 ) 258 259 for control_file in suite_control_files: 260 test_runner_utils.run_job( 261 mox.ContainsAttributeValue('control_file', control_file), 262 remote, 263 mox.IsA(host_info.HostInfo), 264 autotest_path, 265 results_dir, 266 fast_mode, 267 id_digits, 268 ssh_verbosity, 269 ssh_options, 270 mox.StrContains(args), 271 False, 272 False, 273 ).AndReturn((0, '/fake/dir')) 274 self.mox.ReplayAll() 275 test_runner_utils.perform_local_run( 276 afe, autotest_path, ['suite:'+suite_name], remote, fast_mode, 277 build=build, board=board, ignore_deps=False, 278 ssh_verbosity=ssh_verbosity, ssh_options=ssh_options, 279 args=args, results_directory=results_dir, job_retry=retry) 280 281 282if __name__ == '__main__': 283 unittest.main() 284