1#!/usr/bin/env python3 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 17from future import standard_library 18 19standard_library.install_aliases() 20 21import copy 22import importlib 23import inspect 24import fnmatch 25import logging 26import os 27import pkgutil 28import sys 29 30from acts import base_test 31from acts import config_parser 32from acts import keys 33from acts import logger 34from acts import records 35from acts import signals 36from acts import utils 37from acts import error 38 39 40def _find_test_class(): 41 """Finds the test class in a test script. 42 43 Walk through module members and find the subclass of BaseTestClass. Only 44 one subclass is allowed in a test script. 45 46 Returns: 47 The test class in the test module. 48 """ 49 test_classes = [] 50 main_module_members = sys.modules["__main__"] 51 for _, module_member in main_module_members.__dict__.items(): 52 if inspect.isclass(module_member): 53 if issubclass(module_member, base_test.BaseTestClass): 54 test_classes.append(module_member) 55 if len(test_classes) != 1: 56 logging.error("Expected 1 test class per file, found %s.", 57 [t.__name__ for t in test_classes]) 58 sys.exit(1) 59 return test_classes[0] 60 61 62def execute_one_test_class(test_class, test_config, test_identifier): 63 """Executes one specific test class. 64 65 You could call this function in your own cli test entry point if you choose 66 not to use act.py. 67 68 Args: 69 test_class: A subclass of acts.base_test.BaseTestClass that has the test 70 logic to be executed. 71 test_config: A dict representing one set of configs for a test run. 72 test_identifier: A list of tuples specifying which test cases to run in 73 the test class. 74 75 Returns: 76 True if all tests passed without any error, False otherwise. 77 78 Raises: 79 If signals.TestAbortAll is raised by a test run, pipe it through. 80 """ 81 tr = TestRunner(test_config, test_identifier) 82 try: 83 tr.run(test_class) 84 return tr.results.is_all_pass 85 except signals.TestAbortAll: 86 raise 87 except: 88 logging.exception("Exception when executing %s.", tr.testbed_name) 89 finally: 90 tr.stop() 91 92 93class TestRunner(object): 94 """The class that instantiates test classes, executes test cases, and 95 report results. 96 97 Attributes: 98 self.test_run_info: A dictionary containing the information needed by 99 test classes for this test run, including params, 100 controllers, and other objects. All of these will 101 be passed to test classes. 102 self.test_configs: A dictionary that is the original test configuration 103 passed in by user. 104 self.id: A string that is the unique identifier of this test run. 105 self.log_path: A string representing the path of the dir under which 106 all logs from this test run should be written. 107 self.log: The logger object used throughout this test run. 108 self.summary_writer: The TestSummaryWriter object used to stream test 109 results to a file. 110 self.test_classes: A dictionary where we can look up the test classes 111 by name to instantiate. Supports unix shell style 112 wildcards. 113 self.run_list: A list of tuples specifying what tests to run. 114 self.results: The test result object used to record the results of 115 this test run. 116 self.running: A boolean signifies whether this test run is ongoing or 117 not. 118 """ 119 120 def __init__(self, test_configs, run_list): 121 self.test_run_info = {} 122 self.test_configs = test_configs 123 self.testbed_configs = self.test_configs[keys.Config.key_testbed.value] 124 self.testbed_name = self.testbed_configs[ 125 keys.Config.key_testbed_name.value] 126 start_time = logger.get_log_file_timestamp() 127 self.id = "{}@{}".format(self.testbed_name, start_time) 128 # log_path should be set before parsing configs. 129 l_path = os.path.join( 130 self.test_configs[keys.Config.key_log_path.value], 131 self.testbed_name, start_time) 132 self.log_path = os.path.abspath(l_path) 133 logger.setup_test_logger(self.log_path, self.testbed_name) 134 self.log = logging.getLogger() 135 self.summary_writer = records.TestSummaryWriter( 136 os.path.join(self.log_path, records.OUTPUT_FILE_SUMMARY)) 137 if self.test_configs.get(keys.Config.key_random.value): 138 test_case_iterations = self.test_configs.get( 139 keys.Config.key_test_case_iterations.value, 10) 140 self.log.info( 141 "Campaign randomizer is enabled with test_case_iterations %s", 142 test_case_iterations) 143 self.run_list = config_parser.test_randomizer( 144 run_list, test_case_iterations=test_case_iterations) 145 self.write_test_campaign() 146 else: 147 self.run_list = run_list 148 self.results = records.TestResult() 149 self.running = False 150 151 def import_test_modules(self, test_paths): 152 """Imports test classes from test scripts. 153 154 1. Locate all .py files under test paths. 155 2. Import the .py files as modules. 156 3. Find the module members that are test classes. 157 4. Categorize the test classes by name. 158 159 Args: 160 test_paths: A list of directory paths where the test files reside. 161 162 Returns: 163 A dictionary where keys are test class name strings, values are 164 actual test classes that can be instantiated. 165 """ 166 167 def is_testfile_name(name, ext): 168 if ext == ".py": 169 if name.endswith("Test") or name.endswith("_test"): 170 return True 171 return False 172 173 file_list = utils.find_files(test_paths, is_testfile_name) 174 test_classes = {} 175 for path, name, _ in file_list: 176 sys.path.append(path) 177 try: 178 module = importlib.import_module(name) 179 except: 180 for test_cls_name, _ in self.run_list: 181 alt_name = name.replace('_', '').lower() 182 alt_cls_name = test_cls_name.lower() 183 # Only block if a test class on the run list causes an 184 # import error. We need to check against both naming 185 # conventions: AaaBbb and aaa_bbb. 186 if name == test_cls_name or alt_name == alt_cls_name: 187 msg = ("Encountered error importing test class %s, " 188 "abort.") % test_cls_name 189 # This exception is logged here to help with debugging 190 # under py2, because "raise X from Y" syntax is only 191 # supported under py3. 192 self.log.exception(msg) 193 raise ValueError(msg) 194 continue 195 for member_name in dir(module): 196 if not member_name.startswith("__"): 197 if member_name.endswith("Test"): 198 test_class = getattr(module, member_name) 199 if inspect.isclass(test_class): 200 test_classes[member_name] = test_class 201 return test_classes 202 203 def parse_config(self, test_configs): 204 """Parses the test configuration and unpacks objects and parameters 205 into a dictionary to be passed to test classes. 206 207 Args: 208 test_configs: A json object representing the test configurations. 209 """ 210 self.test_run_info[ 211 keys.Config.ikey_testbed_name.value] = self.testbed_name 212 self.test_run_info['testbed_configs'] = copy.deepcopy( 213 self.testbed_configs) 214 # Unpack other params. 215 self.test_run_info[keys.Config.ikey_logpath.value] = self.log_path 216 self.test_run_info[keys.Config.ikey_logger.value] = self.log 217 self.test_run_info[ 218 keys.Config.ikey_summary_writer.value] = self.summary_writer 219 cli_args = test_configs.get(keys.Config.ikey_cli_args.value) 220 self.test_run_info[keys.Config.ikey_cli_args.value] = cli_args 221 user_param_pairs = [] 222 for item in test_configs.items(): 223 if item[0] not in keys.Config.reserved_keys.value: 224 user_param_pairs.append(item) 225 self.test_run_info[keys.Config.ikey_user_param.value] = copy.deepcopy( 226 dict(user_param_pairs)) 227 228 def set_test_util_logs(self, module=None): 229 """Sets the log object to each test util module. 230 231 This recursively include all modules under acts.test_utils and sets the 232 main test logger to each module. 233 234 Args: 235 module: A module under acts.test_utils. 236 """ 237 # Initial condition of recursion. 238 if not module: 239 module = importlib.import_module("acts.test_utils") 240 # Somehow pkgutil.walk_packages is not working for me. 241 # Using iter_modules for now. 242 pkg_iter = pkgutil.iter_modules(module.__path__, module.__name__ + '.') 243 for _, module_name, ispkg in pkg_iter: 244 m = importlib.import_module(module_name) 245 if ispkg: 246 self.set_test_util_logs(module=m) 247 else: 248 self.log.debug("Setting logger to test util module %s", 249 module_name) 250 setattr(m, "log", self.log) 251 252 def run_test_class(self, test_cls_name, test_cases=None): 253 """Instantiates and executes a test class. 254 255 If test_cases is None, the test cases listed by self.tests will be 256 executed instead. If self.tests is empty as well, no test case in this 257 test class will be executed. 258 259 Args: 260 test_cls_name: Name of the test class to execute. 261 test_cases: List of test case names to execute within the class. 262 263 Raises: 264 ValueError is raised if the requested test class could not be found 265 in the test_paths directories. 266 """ 267 matches = fnmatch.filter(self.test_classes.keys(), test_cls_name) 268 if not matches: 269 self.log.info( 270 "Cannot find test class %s or classes matching pattern, " 271 "skipping for now." % test_cls_name) 272 record = records.TestResultRecord("*all*", test_cls_name) 273 record.test_skip(signals.TestSkip("Test class does not exist.")) 274 self.results.add_record(record) 275 return 276 if matches != [test_cls_name]: 277 self.log.info("Found classes matching pattern %s: %s", 278 test_cls_name, matches) 279 280 for test_cls_name_match in matches: 281 test_cls = self.test_classes[test_cls_name_match] 282 if self.test_configs.get(keys.Config.key_random.value) or ( 283 "Preflight" in test_cls_name_match) or ( 284 "Postflight" in test_cls_name_match): 285 test_case_iterations = 1 286 else: 287 test_case_iterations = self.test_configs.get( 288 keys.Config.key_test_case_iterations.value, 1) 289 290 with test_cls(self.test_run_info) as test_cls_instance: 291 try: 292 cls_result = test_cls_instance.run(test_cases, 293 test_case_iterations) 294 self.results += cls_result 295 self._write_results_to_file() 296 except signals.TestAbortAll as e: 297 self.results += e.results 298 raise e 299 300 def run(self, test_class=None): 301 """Executes test cases. 302 303 This will instantiate controller and test classes, and execute test 304 classes. This can be called multiple times to repeatedly execute the 305 requested test cases. 306 307 A call to TestRunner.stop should eventually happen to conclude the life 308 cycle of a TestRunner. 309 310 Args: 311 test_class: The python module of a test class. If provided, run this 312 class; otherwise, import modules in under test_paths 313 based on run_list. 314 """ 315 if not self.running: 316 self.running = True 317 # Initialize controller objects and pack appropriate objects/params 318 # to be passed to test class. 319 self.parse_config(self.test_configs) 320 if test_class: 321 self.test_classes = {test_class.__name__: test_class} 322 else: 323 t_paths = self.test_configs[keys.Config.key_test_paths.value] 324 self.test_classes = self.import_test_modules(t_paths) 325 self.log.debug("Executing run list %s.", self.run_list) 326 for test_cls_name, test_case_names in self.run_list: 327 if not self.running: 328 break 329 330 if test_case_names: 331 self.log.debug("Executing test cases %s in test class %s.", 332 test_case_names, test_cls_name) 333 else: 334 self.log.debug("Executing test class %s", test_cls_name) 335 336 try: 337 self.run_test_class(test_cls_name, test_case_names) 338 except error.ActsError as e: 339 self.results.errors.append(e) 340 self.log.error("Test Runner Error: %s" % e.message) 341 except signals.TestAbortAll as e: 342 self.log.warning( 343 "Abort all subsequent test classes. Reason: %s", e) 344 raise 345 346 def stop(self): 347 """Releases resources from test run. Should always be called after 348 TestRunner.run finishes. 349 350 This function concludes a test run and writes out a test report. 351 """ 352 if self.running: 353 msg = "\nSummary for test run %s: %s\n" % ( 354 self.id, self.results.summary_str()) 355 self._write_results_to_file() 356 self.log.info(msg.strip()) 357 logger.kill_test_logger(self.log) 358 self.running = False 359 360 def _write_results_to_file(self): 361 """Writes test results to file(s) in a serializable format.""" 362 # Old JSON format 363 path = os.path.join(self.log_path, "test_run_summary.json") 364 with open(path, 'w') as f: 365 f.write(self.results.json_str()) 366 # New YAML format 367 self.summary_writer.dump( 368 self.results.summary_dict(), records.TestSummaryEntryType.SUMMARY) 369 370 def write_test_campaign(self): 371 """Log test campaign file.""" 372 path = os.path.join(self.log_path, "test_campaign.log") 373 with open(path, 'w') as f: 374 for test_class, test_cases in self.run_list: 375 f.write("%s:\n%s" % (test_class, ",\n".join(test_cases))) 376 f.write("\n\n") 377