1 #!/usr/bin/env python3.4 2 # 3 # Copyright 2016 - The Android Open Source Project 4 # 5 # Licensed under the Apache License, Version 2.0 (the "License"); 6 # you may not use this file except in compliance with the License. 7 # You may obtain a copy of the License at 8 # 9 # http://www.apache.org/licenses/LICENSE-2.0 10 # 11 # Unless required by applicable law or agreed to in writing, software 12 # distributed under the License is distributed on an "AS IS" BASIS, 13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 # See the License for the specific language governing permissions and 15 # limitations under the License. 16 17 import os 18 19 from acts import asserts 20 from acts import keys 21 from acts import logger 22 from acts import records 23 from acts import signals 24 from acts import test_runner 25 from acts import utils 26 27 # Macro strings for test result reporting 28 TEST_CASE_TOKEN = "[Test Case]" 29 RESULT_LINE_TEMPLATE = TEST_CASE_TOKEN + " %s %s" 30 31 class BaseTestError(Exception): 32 """Raised for exceptions that occured in BaseTestClass.""" 33 34 class BaseTestClass(object): 35 """Base class for all test classes to inherit from. 36 37 This class gets all the controller objects from test_runner and executes 38 the test cases requested within itself. 39 40 Most attributes of this class are set at runtime based on the configuration 41 provided. 42 43 Attributes: 44 tests: A list of strings, each representing a test case name. 45 TAG: A string used to refer to a test class. Default is the test class 46 name. 47 log: A logger object used for logging. 48 results: A records.TestResult object for aggregating test results from 49 the execution of test cases. 50 current_test_name: A string that's the name of the test case currently 51 being executed. If no test is executing, this should 52 be None. 53 """ 54 55 TAG = None 56 57 def __init__(self, configs): 58 self.tests = [] 59 if not self.TAG: 60 self.TAG = self.__class__.__name__ 61 # Set all the controller objects and params. 62 for name, value in configs.items(): 63 setattr(self, name, value) 64 self.results = records.TestResult() 65 self.current_test_name = None 66 67 def __enter__(self): 68 return self 69 70 def __exit__(self, *args): 71 self._exec_func(self.clean_up) 72 73 def unpack_userparams(self, req_param_names=[], opt_param_names=[], 74 **kwargs): 75 """Unpacks user defined parameters in test config into individual 76 variables. 77 78 Instead of accessing the user param with self.user_params["xxx"], the 79 variable can be directly accessed with self.xxx. 80 81 A missing required param will raise an exception. If an optional param 82 is missing, an INFO line will be logged. 83 84 Args: 85 req_param_names: A list of names of the required user params. 86 opt_param_names: A list of names of the optional user params. 87 **kwargs: Arguments that provide default values. 88 e.g. unpack_userparams(required_list, opt_list, arg_a="hello") 89 self.arg_a will be "hello" unless it is specified again in 90 required_list or opt_list. 91 92 Raises: 93 BaseTestError is raised if a required user params is missing from 94 test config. 95 """ 96 for k, v in kwargs.items(): 97 setattr(self, k, v) 98 for name in req_param_names: 99 if name not in self.user_params: 100 raise BaseTestError(("Missing required user param '%s' in test" 101 " configuration.") % name) 102 setattr(self, name, self.user_params[name]) 103 for name in opt_param_names: 104 if name not in self.user_params: 105 self.log.info(("Missing optional user param '%s' in " 106 "configuration, continue."), name) 107 else: 108 setattr(self, name, self.user_params[name]) 109 110 def _setup_class(self): 111 """Proxy function to guarantee the base implementation of setup_class 112 is called. 113 """ 114 return self.setup_class() 115 116 def setup_class(self): 117 """Setup function that will be called before executing any test case in 118 the test class. 119 120 To signal setup failure, return False or raise an exception. If 121 exceptions were raised, the stack trace would appear in log, but the 122 exceptions would not propagate to upper levels. 123 124 Implementation is optional. 125 """ 126 127 def teardown_class(self): 128 """Teardown function that will be called after all the selected test 129 cases in the test class have been executed. 130 131 Implementation is optional. 132 """ 133 134 def _setup_test(self, test_name): 135 """Proxy function to guarantee the base implementation of setup_test is 136 called. 137 """ 138 self.current_test_name = test_name 139 try: 140 # Write test start token to adb log if android device is attached. 141 for ad in self.android_devices: 142 ad.droid.logV("%s BEGIN %s" % (TEST_CASE_TOKEN, test_name)) 143 except: 144 pass 145 return self.setup_test() 146 147 def setup_test(self): 148 """Setup function that will be called every time before executing each 149 test case in the test class. 150 151 To signal setup failure, return False or raise an exception. If 152 exceptions were raised, the stack trace would appear in log, but the 153 exceptions would not propagate to upper levels. 154 155 Implementation is optional. 156 """ 157 158 def _teardown_test(self, test_name): 159 """Proxy function to guarantee the base implementation of teardown_test 160 is called. 161 """ 162 try: 163 # Write test end token to adb log if android device is attached. 164 for ad in self.android_devices: 165 ad.droid.logV("%s END %s" % (TEST_CASE_TOKEN, test_name)) 166 except: 167 pass 168 try: 169 self.teardown_test() 170 finally: 171 self.current_test_name = None 172 173 def teardown_test(self): 174 """Teardown function that will be called every time a test case has 175 been executed. 176 177 Implementation is optional. 178 """ 179 180 def _on_fail(self, record): 181 """Proxy function to guarantee the base implementation of on_fail is 182 called. 183 184 Args: 185 record: The records.TestResultRecord object for the failed test 186 case. 187 """ 188 test_name = record.test_name 189 self.log.error(record.details) 190 begin_time = logger.epoch_to_log_line_timestamp(record.begin_time) 191 self.log.info(RESULT_LINE_TEMPLATE, test_name, record.result) 192 self.on_fail(test_name, begin_time) 193 194 def on_fail(self, test_name, begin_time): 195 """A function that is executed upon a test case failure. 196 197 User implementation is optional. 198 199 Args: 200 test_name: Name of the test that triggered this function. 201 begin_time: Logline format timestamp taken when the test started. 202 """ 203 204 def _on_pass(self, record): 205 """Proxy function to guarantee the base implementation of on_pass is 206 called. 207 208 Args: 209 record: The records.TestResultRecord object for the passed test 210 case. 211 """ 212 test_name = record.test_name 213 begin_time = logger.epoch_to_log_line_timestamp(record.begin_time) 214 msg = record.details 215 if msg: 216 self.log.info(msg) 217 self.log.info(RESULT_LINE_TEMPLATE, test_name, record.result) 218 self.on_pass(test_name, begin_time) 219 220 def on_pass(self, test_name, begin_time): 221 """A function that is executed upon a test case passing. 222 223 Implementation is optional. 224 225 Args: 226 test_name: Name of the test that triggered this function. 227 begin_time: Logline format timestamp taken when the test started. 228 """ 229 230 def _on_skip(self, record): 231 """Proxy function to guarantee the base implementation of on_skip is 232 called. 233 234 Args: 235 record: The records.TestResultRecord object for the skipped test 236 case. 237 """ 238 test_name = record.test_name 239 begin_time = logger.epoch_to_log_line_timestamp(record.begin_time) 240 self.log.info(RESULT_LINE_TEMPLATE, test_name, record.result) 241 self.log.info("Reason to skip: %s", record.details) 242 self.on_skip(test_name, begin_time) 243 244 def on_skip(self, test_name, begin_time): 245 """A function that is executed upon a test case being skipped. 246 247 Implementation is optional. 248 249 Args: 250 test_name: Name of the test that triggered this function. 251 begin_time: Logline format timestamp taken when the test started. 252 """ 253 254 def _on_exception(self, record): 255 """Proxy function to guarantee the base implementation of on_exception 256 is called. 257 258 Args: 259 record: The records.TestResultRecord object for the failed test 260 case. 261 """ 262 test_name = record.test_name 263 self.log.exception(record.details) 264 begin_time = logger.epoch_to_log_line_timestamp(record.begin_time) 265 self.log.info(RESULT_LINE_TEMPLATE, test_name, record.result) 266 self.on_exception(test_name, begin_time) 267 268 def on_exception(self, test_name, begin_time): 269 """A function that is executed upon an unhandled exception from a test 270 case. 271 272 Implementation is optional. 273 274 Args: 275 test_name: Name of the test that triggered this function. 276 begin_time: Logline format timestamp taken when the test started. 277 """ 278 279 def _exec_procedure_func(self, func, tr_record): 280 """Executes a procedure function like on_pass, on_fail etc. 281 282 This function will alternate the 'Result' of the test's record if 283 exceptions happened when executing the procedure function. 284 285 This will let signals.TestAbortAll through so abort_all works in all 286 procedure functions. 287 288 Args: 289 func: The procedure function to be executed. 290 tr_record: The TestResultRecord object associated with the test 291 case executed. 292 """ 293 try: 294 func(tr_record) 295 except signals.TestAbortAll: 296 raise 297 except Exception as e: 298 self.log.exception("Exception happened when executing %s for %s.", 299 func.__name__, self.current_test_name) 300 tr_record.add_error(func.__name__, e) 301 302 def exec_one_testcase(self, test_name, test_func, args, **kwargs): 303 """Executes one test case and update test results. 304 305 Executes one test case, create a records.TestResultRecord object with 306 the execution information, and add the record to the test class's test 307 results. 308 309 Args: 310 test_name: Name of the test. 311 test_func: The test function. 312 args: A tuple of params. 313 kwargs: Extra kwargs. 314 """ 315 is_generate_trigger = False 316 tr_record = records.TestResultRecord(test_name, self.TAG) 317 tr_record.test_begin() 318 self.log.info("%s %s", TEST_CASE_TOKEN, test_name) 319 verdict = None 320 try: 321 ret = self._setup_test(test_name) 322 asserts.assert_true(ret is not False, 323 "Setup for %s failed." % test_name) 324 try: 325 if args or kwargs: 326 verdict = test_func(*args, **kwargs) 327 else: 328 verdict = test_func() 329 finally: 330 self._teardown_test(test_name) 331 except (signals.TestFailure, AssertionError) as e: 332 tr_record.test_fail(e) 333 self._exec_procedure_func(self._on_fail, tr_record) 334 except signals.TestSkip as e: 335 # Test skipped. 336 tr_record.test_skip(e) 337 self._exec_procedure_func(self._on_skip, tr_record) 338 except (signals.TestAbortClass, signals.TestAbortAll) as e: 339 # Abort signals, pass along. 340 tr_record.test_fail(e) 341 raise e 342 except signals.TestPass as e: 343 # Explicit test pass. 344 tr_record.test_pass(e) 345 self._exec_procedure_func(self._on_pass, tr_record) 346 except signals.TestSilent as e: 347 # This is a trigger test for generated tests, suppress reporting. 348 is_generate_trigger = True 349 self.results.requested.remove(test_name) 350 except Exception as e: 351 # Exception happened during test. 352 tr_record.test_unknown(e) 353 self._exec_procedure_func(self._on_exception, tr_record) 354 self._exec_procedure_func(self._on_fail, tr_record) 355 else: 356 # Keep supporting return False for now. 357 # TODO(angli): Deprecate return False support. 358 if verdict or (verdict is None): 359 # Test passed. 360 tr_record.test_pass() 361 self._exec_procedure_func(self._on_pass, tr_record) 362 return 363 # Test failed because it didn't return True. 364 # This should be removed eventually. 365 tr_record.test_fail() 366 self._exec_procedure_func(self._on_fail, tr_record) 367 finally: 368 if not is_generate_trigger: 369 self.results.add_record(tr_record) 370 371 def run_generated_testcases(self, test_func, settings, 372 args=None, kwargs=None, 373 tag="", name_func=None): 374 """Runs generated test cases. 375 376 Generated test cases are not written down as functions, but as a list 377 of parameter sets. This way we reduce code repetition and improve 378 test case scalability. 379 380 Args: 381 test_func: The common logic shared by all these generated test 382 cases. This function should take at least one argument, 383 which is a parameter set. 384 settings: A list of strings representing parameter sets. These are 385 usually json strings that get loaded in the test_func. 386 args: Iterable of additional position args to be passed to 387 test_func. 388 kwargs: Dict of additional keyword args to be passed to test_func 389 tag: Name of this group of generated test cases. Ignored if 390 name_func is provided and operates properly. 391 name_func: A function that takes a test setting and generates a 392 proper test name. The test name should be shorter than 393 utils.MAX_FILENAME_LEN. Names over the limit will be 394 truncated. 395 396 Returns: 397 A list of settings that did not pass. 398 """ 399 args = args or () 400 kwargs = kwargs or {} 401 failed_settings = [] 402 for s in settings: 403 test_name = "{} {}".format(tag, s) 404 if name_func: 405 try: 406 test_name = name_func(s, *args, **kwargs) 407 except: 408 self.log.exception(("Failed to get test name from " 409 "test_func. Fall back to default %s"), 410 test_name) 411 self.results.requested.append(test_name) 412 if len(test_name) > utils.MAX_FILENAME_LEN: 413 test_name = test_name[:utils.MAX_FILENAME_LEN] 414 previous_success_cnt = len(self.results.passed) 415 self.exec_one_testcase(test_name, test_func, (s,) + args, **kwargs) 416 if len(self.results.passed) - previous_success_cnt != 1: 417 failed_settings.append(s) 418 return failed_settings 419 420 def _exec_func(self, func, *args): 421 """Executes a function with exception safeguard. 422 423 This will let signals.TestAbortAll through so abort_all works in all 424 procedure functions. 425 426 Args: 427 func: Function to be executed. 428 args: Arguments to be passed to the function. 429 430 Returns: 431 Whatever the function returns, or False if unhandled exception 432 occured. 433 """ 434 try: 435 return func(*args) 436 except signals.TestAbortAll: 437 raise 438 except: 439 self.log.exception("Exception happened when executing %s in %s.", 440 func.__name__, self.TAG) 441 return False 442 443 def _get_all_test_names(self): 444 """Finds all the function names that match the test case naming 445 convention in this class. 446 447 Returns: 448 A list of strings, each is a test case name. 449 """ 450 test_names = [] 451 for name in dir(self): 452 if name.startswith("test_"): 453 test_names.append(name) 454 return test_names 455 456 def _get_test_funcs(self, test_names): 457 """Obtain the actual functions of test cases based on test names. 458 459 Args: 460 test_names: A list of strings, each string is a test case name. 461 462 Returns: 463 A list of tuples of (string, function). String is the test case 464 name, function is the actual test case function. 465 466 Raises: 467 test_runner.USERError is raised if the test name does not follow 468 naming convention "test_*". This can only be caused by user input 469 here. 470 """ 471 test_funcs = [] 472 for test_name in test_names: 473 if not test_name.startswith("test_"): 474 msg = ("Test case name %s does not follow naming convention " 475 "test_*, abort.") % test_name 476 raise test_runner.USERError(msg) 477 try: 478 test_funcs.append((test_name, getattr(self, test_name))) 479 except AttributeError: 480 self.log.warning("%s does not have test case %s.", self.TAG, 481 test_name) 482 except BaseTestError as e: 483 self.log.warning(str(e)) 484 return test_funcs 485 486 def run(self, test_names=None): 487 """Runs test cases within a test class by the order they appear in the 488 execution list. 489 490 One of these test cases lists will be executed, shown here in priority 491 order: 492 1. The test_names list, which is passed from cmd line. Invalid names 493 are guarded by cmd line arg parsing. 494 2. The self.tests list defined in test class. Invalid names are 495 ignored. 496 3. All function that matches test case naming convention in the test 497 class. 498 499 Args: 500 test_names: A list of string that are test case names requested in 501 cmd line. 502 503 Returns: 504 The test results object of this class. 505 """ 506 self.log.info("==========> %s <==========", self.TAG) 507 # Devise the actual test cases to run in the test class. 508 if not test_names: 509 if self.tests: 510 # Specified by run list in class. 511 test_names = list(self.tests) 512 else: 513 # No test case specified by user, execute all in the test class 514 test_names = self._get_all_test_names() 515 self.results.requested = test_names 516 tests = self._get_test_funcs(test_names) 517 # Setup for the class. 518 try: 519 if self._setup_class() is False: 520 raise signals.TestFailure("Failed to setup %s." % self.TAG) 521 except Exception as e: 522 self.log.exception("Failed to setup %s.", self.TAG) 523 self.results.fail_class(self.TAG, e) 524 self._exec_func(self.teardown_class) 525 return self.results 526 # Run tests in order. 527 try: 528 for test_name, test_func in tests: 529 self.exec_one_testcase(test_name, test_func, self.cli_args) 530 return self.results 531 except signals.TestAbortClass: 532 return self.results 533 except signals.TestAbortAll as e: 534 # Piggy-back test results on this exception object so we don't lose 535 # results from this test class. 536 setattr(e, "results", self.results) 537 raise e 538 finally: 539 self._exec_func(self.teardown_class) 540 self.log.info("Summary for test class %s: %s", self.TAG, 541 self.results.summary_str()) 542 543 def clean_up(self): 544 """A function that is executed upon completion of all tests cases 545 selected in the test class. 546 547 This function should clean up objects initialized in the constructor by 548 user. 549 """ 550
