1#!/usr/bin/python3 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 11import unittest 12from unittest.mock import patch 13 14import common 15 16import shutil 17import tempfile 18import types 19from autotest_lib.client.common_lib import control_data 20from autotest_lib.server.cros.dynamic_suite import control_file_getter 21from autotest_lib.server.cros.dynamic_suite import suite as suite_module 22from autotest_lib.server.hosts import host_info 23from autotest_lib.site_utils import test_runner_utils 24 25 26class TypeMatcher(object): 27 """Matcher for object is of type.""" 28 29 def __init__(self, expected_type): 30 self.expected_type = expected_type 31 32 def __eq__(self, other): 33 return isinstance(other, self.expected_type) 34 35 36class JobMatcher(object): 37 """Matcher for JobObject + Name.""" 38 39 def __init__(self, expected_type, name): 40 self.expected_type = expected_type 41 self.name = name 42 43 def __eq__(self, other): 44 return (isinstance(other, self.expected_type) 45 and self.name in other.name) 46 47 48class hostinfoMatcher(object): 49 """Match hostinfo stuff""" 50 51 def __init__(self, labels, attributes): 52 self.labels = labels.split(' ') 53 self.attributes = attributes 54 55 def __eq__(self, other): 56 return self.labels == other.labels and self.attributes == other.attributes 57 58 59class ContainsMatcher: 60 """Matcher for object contains attr.""" 61 62 def __init__(self, key, value): 63 self.key = key 64 self.value = value 65 66 def __eq__(self, rhs): 67 try: 68 return getattr(rhs, self._key) == self._value 69 except Exception: 70 return False 71 72 73class SampleJob(object): 74 """Sample to be used for mocks.""" 75 76 def __init__(self, id=1): 77 self.id = id 78 79 80class FakeTests(object): 81 """A fake test to be used for mocks.""" 82 83 def __init__(self, text, deps=[], py_version=None): 84 self.text = text 85 self.test_type = 'client' 86 self.dependencies = deps 87 self.name = text 88 self.py_version = py_version 89 90 91class TestRunnerUnittests(unittest.TestCase): 92 """Test test_runner_utils.""" 93 94 autotest_path = 'ottotest_path' 95 suite_name = 'sweet_name' 96 test_arg = 'suite:' + suite_name 97 remote = 'remoat' 98 build = 'bild' 99 board = 'bored' 100 fast_mode = False 101 suite_control_files = ['c1', 'c2', 'c3', 'c4'] 102 results_dir = '/tmp/test_that_results_fake' 103 id_digits = 1 104 ssh_verbosity = 2 105 ssh_options = '-F /dev/null -i /dev/null' 106 args = 'matey' 107 retry = True 108 109 def _results_directory_from_results_list(self, results_list): 110 """Generate a temp directory filled with provided test results. 111 112 @param results_list: List of results, each result is a tuple of strings 113 (test_name, test_status_message). 114 @returns: Absolute path to the results directory. 115 """ 116 global_dir = tempfile.mkdtemp() 117 for index, (test_name, test_status_message) in enumerate(results_list): 118 dir_name = '-'.join(['results', 119 "%02.f" % (index + 1), 120 test_name]) 121 local_dir = os.path.join(global_dir, dir_name) 122 os.mkdir(local_dir) 123 os.mkdir('%s/debug' % local_dir) 124 with open("%s/status.log" % local_dir, mode='w+') as status: 125 status.write(test_status_message) 126 status.flush() 127 return global_dir 128 129 def test_handle_local_result_for_good_test(self): 130 patcher = patch.object(control_file_getter, 'DevServerGetter') 131 getter = patcher.start() 132 self.addCleanup(patcher.stop) 133 getter.get_control_file_list.return_value = [] 134 job = SampleJob() 135 136 test_patcher = patch.object(control_data, 'ControlData') 137 test = test_patcher.start() 138 self.addCleanup(test_patcher.stop) 139 test.job_retries = 5 140 141 suite = test_runner_utils.LocalSuite([], "tag", [], None, getter, 142 job_retry=True) 143 suite._retry_handler = suite_module.RetryHandler({job.id: test}) 144 145 #No calls, should not be retried 146 directory = self._results_directory_from_results_list([ 147 ("dummy_Good", "GOOD: nonexistent test completed successfully")]) 148 new_id = suite.handle_local_result( 149 job.id, directory, 150 lambda log_entry, log_in_subdir=False: None) 151 self.assertIsNone(new_id) 152 shutil.rmtree(directory) 153 154 def test_handle_local_result_for_bad_test(self): 155 patcher = patch.object(control_file_getter, 'DevServerGetter') 156 getter = patcher.start() 157 self.addCleanup(patcher.stop) 158 getter.get_control_file_list.return_value = [] 159 160 job = SampleJob() 161 162 test_patcher = patch.object(control_data, 'ControlData') 163 test = test_patcher.start() 164 self.addCleanup(test_patcher.stop) 165 test.job_retries = 5 166 167 utils_mock = patch.object(test_runner_utils.LocalSuite, 168 '_retry_local_result') 169 test_runner_utils_mock = utils_mock.start() 170 self.addCleanup(utils_mock.stop) 171 test_runner_utils_mock._retry_local_result.return_value = 42 172 173 suite = test_runner_utils.LocalSuite([], "tag", [], None, getter, 174 job_retry=True) 175 suite._retry_handler = suite_module.RetryHandler({job.id: test}) 176 177 directory = self._results_directory_from_results_list([ 178 ("dummy_Bad", "FAIL")]) 179 new_id = suite.handle_local_result( 180 job.id, directory, 181 lambda log_entry, log_in_subdir=False: None) 182 self.assertIsNotNone(new_id) 183 shutil.rmtree(directory) 184 185 186 def test_generate_report_status_code_success_with_retries(self): 187 global_dir = self._results_directory_from_results_list([ 188 ("dummy_Flaky", "FAIL"), 189 ("dummy_Flaky", "GOOD: nonexistent test completed successfully")]) 190 status_code = test_runner_utils.generate_report( 191 global_dir, just_status_code=True) 192 self.assertEquals(status_code, 0) 193 shutil.rmtree(global_dir) 194 195 196 def test_generate_report_status_code_failure_with_retries(self): 197 global_dir = self._results_directory_from_results_list([ 198 ("dummy_Good", "GOOD: nonexistent test completed successfully"), 199 ("dummy_Bad", "FAIL"), 200 ("dummy_Bad", "FAIL")]) 201 status_code = test_runner_utils.generate_report( 202 global_dir, just_status_code=True) 203 self.assertNotEquals(status_code, 0) 204 shutil.rmtree(global_dir) 205 206 207 def test_get_predicate_for_test_arg(self): 208 # Assert the type signature of get_predicate_for_test(...) 209 # Because control.test_utils_wrapper calls this function, 210 # it is imperative for backwards compatilbility that 211 # the return type of the tested function does not change. 212 tests = ['dummy_test', 'e:name_expression', 'f:expression', 213 'suite:suitename'] 214 for test in tests: 215 pred, desc = test_runner_utils.get_predicate_for_test_arg(test) 216 self.assertTrue(isinstance(pred, types.FunctionType)) 217 self.assertTrue(isinstance(desc, str)) 218 219 def test_perform_local_run(self): 220 """Test a local run that should pass.""" 221 patcher = patch.object(test_runner_utils, '_auto_detect_labels') 222 _auto_detect_labels_mock = patcher.start() 223 self.addCleanup(patcher.stop) 224 225 patcher2 = patch.object(test_runner_utils, 'get_all_control_files') 226 get_all_control_files_mock = patcher2.start() 227 self.addCleanup(patcher2.stop) 228 229 _auto_detect_labels_mock.return_value = [ 230 'os:cros', 'has_chameleon:True' 231 ] 232 233 get_all_control_files_mock.return_value = [ 234 FakeTests(test, deps=['has_chameleon:True']) 235 for test in self.suite_control_files 236 ] 237 238 patcher3 = patch.object(test_runner_utils, 'run_job') 239 run_job_mock = patcher3.start() 240 self.addCleanup(patcher3.stop) 241 242 for control_file in self.suite_control_files: 243 run_job_mock.return_value = (0, '/fake/dir') 244 test_runner_utils.perform_local_run(self.autotest_path, 245 ['suite:' + self.suite_name], 246 self.remote, 247 self.fast_mode, 248 build=self.build, 249 board=self.board, 250 ssh_verbosity=self.ssh_verbosity, 251 ssh_options=self.ssh_options, 252 args=self.args, 253 results_directory=self.results_dir, 254 job_retry=self.retry, 255 ignore_deps=False, 256 minus=[]) 257 258 run_job_mock.assert_called_with(job=TypeMatcher( 259 test_runner_utils.SimpleJob), 260 host=self.remote, 261 info=TypeMatcher(host_info.HostInfo), 262 autotest_path=self.autotest_path, 263 results_directory=self.results_dir, 264 fast_mode=self.fast_mode, 265 id_digits=self.id_digits, 266 ssh_verbosity=self.ssh_verbosity, 267 ssh_options=self.ssh_options, 268 args=TypeMatcher(str), 269 pretend=False, 270 autoserv_verbose=False, 271 companion_hosts=None, 272 dut_servers=None, 273 is_cft=False, 274 ch_info={}) 275 276 def test_perform_local_run_missing_deps(self): 277 """Test a local run with missing dependencies. No tests should run.""" 278 patcher = patch.object(test_runner_utils, '_auto_detect_labels') 279 getter = patcher.start() 280 self.addCleanup(patcher.stop) 281 282 getter.return_value = ['os:cros', 'has_chameleon:True'] 283 284 patcher2 = patch.object(test_runner_utils, 'get_all_control_files') 285 test_runner_utils_mock = patcher2.start() 286 self.addCleanup(patcher2.stop) 287 test_runner_utils_mock.return_value = [ 288 FakeTests(test, deps=['has_chameleon:False']) 289 for test in self.suite_control_files 290 ] 291 292 res = test_runner_utils.perform_local_run( 293 self.autotest_path, ['suite:' + self.suite_name], 294 self.remote, 295 self.fast_mode, 296 build=self.build, 297 board=self.board, 298 ssh_verbosity=self.ssh_verbosity, 299 ssh_options=self.ssh_options, 300 args=self.args, 301 results_directory=self.results_dir, 302 job_retry=self.retry, 303 ignore_deps=False, 304 minus=[]) 305 306 # Verify when the deps are not met, the tests are not run. 307 self.assertEquals(res, []) 308 309 def test_minus_flag(self): 310 """Verify the minus flag skips tests.""" 311 patcher = patch.object(test_runner_utils, '_auto_detect_labels') 312 getter = patcher.start() 313 self.addCleanup(patcher.stop) 314 315 getter.return_value = ['os:cros', 'has_chameleon:True'] 316 317 patcher2 = patch.object(test_runner_utils, 'get_all_control_files') 318 test_runner_utils_mock = patcher2.start() 319 self.addCleanup(patcher2.stop) 320 321 patcher3 = patch.object(test_runner_utils, 'run_job') 322 run_job_mock = patcher3.start() 323 self.addCleanup(patcher3.stop) 324 325 minus_tests = [FakeTests(self.suite_control_files[0])] 326 all_tests = [ 327 FakeTests(test, deps=[]) for test in self.suite_control_files 328 ] 329 330 test_runner_utils_mock.side_effect = [minus_tests, all_tests] 331 run_job_mock.side_effect = [(0, 'fakedir') for _ in range(3)] 332 test_labels = "'a' 'test' 'label'" 333 test_attributes = {"servo": "yes"} 334 335 res = test_runner_utils.perform_local_run( 336 self.autotest_path, ['suite:' + self.suite_name], 337 self.remote, 338 self.fast_mode, 339 build=self.build, 340 board=self.board, 341 ssh_verbosity=self.ssh_verbosity, 342 ssh_options=self.ssh_options, 343 args=self.args, 344 results_directory=self.results_dir, 345 host_attributes=test_attributes, 346 job_retry=self.retry, 347 ignore_deps=False, 348 minus=[self.suite_control_files[0]], 349 is_cft=True, 350 host_labels=test_labels, 351 label=None) 352 353 from mock import call 354 355 calls = [] 356 for name in self.suite_control_files[1:]: 357 calls.append( 358 call(job=JobMatcher(test_runner_utils.SimpleJob, 359 name=name), 360 host=self.remote, 361 info=hostinfoMatcher(labels=test_labels, 362 attributes=test_attributes), 363 autotest_path=self.autotest_path, 364 results_directory=self.results_dir, 365 fast_mode=self.fast_mode, 366 id_digits=self.id_digits, 367 ssh_verbosity=self.ssh_verbosity, 368 ssh_options=self.ssh_options, 369 args=TypeMatcher(str), 370 pretend=False, 371 autoserv_verbose=False, 372 companion_hosts=None, 373 dut_servers=None, 374 is_cft=True, 375 ch_info={})) 376 377 run_job_mock.assert_has_calls(calls, any_order=True) 378 assert run_job_mock.call_count == len(calls) 379 380 def test_set_pyversion(self): 381 """Test the tests can properly set the python version.""" 382 383 # When a test is missing a version, use the current setting. 384 starting_version = os.getenv('PY_VERSION') 385 386 try: 387 fake_test1 = FakeTests('foo') 388 fake_test2 = FakeTests('foo', py_version=2) 389 fake_test3 = FakeTests('foo', py_version=3) 390 391 test_runner_utils._set_pyversion( 392 [fake_test1, fake_test2, fake_test3]) 393 self.assertEqual(os.getenv('PY_VERSION'), starting_version) 394 395 # When there is a mix, use the current setting. 396 starting_version = os.getenv('PY_VERSION') 397 fake_test1 = FakeTests('foo', py_version=2) 398 fake_test2 = FakeTests('foo', py_version=2) 399 fake_test3 = FakeTests('foo', py_version=3) 400 401 test_runner_utils._set_pyversion( 402 [fake_test1, fake_test2, fake_test3]) 403 self.assertEqual(os.getenv('PY_VERSION'), starting_version) 404 405 # When all agree, but still 1 missing, use the current setting. 406 fake_test1 = FakeTests('foo') 407 fake_test2 = FakeTests('foo', py_version=3) 408 fake_test3 = FakeTests('foo', py_version=3) 409 410 test_runner_utils._set_pyversion( 411 [fake_test1, fake_test2, fake_test3]) 412 self.assertEqual(os.getenv('PY_VERSION'), starting_version) 413 414 # When all are set to 3, use 3. 415 fake_test1 = FakeTests('foo', py_version=3) 416 fake_test2 = FakeTests('foo', py_version=3) 417 fake_test3 = FakeTests('foo', py_version=3) 418 419 test_runner_utils._set_pyversion( 420 [fake_test1, fake_test2, fake_test3]) 421 self.assertEqual(os.getenv('PY_VERSION'), '3') 422 423 # When all are set to 2, use 2. 424 fake_test1 = FakeTests('foo', py_version=2) 425 fake_test2 = FakeTests('foo', py_version=2) 426 fake_test3 = FakeTests('foo', py_version=2) 427 428 test_runner_utils._set_pyversion( 429 [fake_test1, fake_test2, fake_test3]) 430 self.assertEqual(os.getenv('PY_VERSION'), '2') 431 finally: 432 # In the event something breaks, reset the pre-test version. 433 os.environ['PY_VERSION'] = starting_version 434 435 def test_host_info_write(self): 436 437 dirpath = tempfile.mkdtemp() 438 439 info = host_info.HostInfo(['some', 'labels'], {'attrib1': '1'}) 440 import pathlib 441 expected_path = os.path.join( 442 pathlib.Path(__file__).parent.absolute(), 443 'host_info_store_testfile') 444 try: 445 446 test_runner_utils._write_host_info(dirpath, 'host_info_store', 447 'localhost:1234', info) 448 test_path = os.path.join(dirpath, 'host_info_store', 449 'localhost:1234.store') 450 with open(test_path, 'r') as rf: 451 test_data = rf.read() 452 with open(expected_path, 'r') as rf: 453 expected_data = rf.read() 454 self.assertEqual(test_data, expected_data) 455 456 finally: 457 shutil.rmtree(dirpath) 458 459 460if __name__ == '__main__': 461 unittest.main() 462