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. 16import itertools 17 18from acts.metrics.loggers.usage_metadata_logger import UsageMetadataPublisher 19from future import standard_library 20 21standard_library.install_aliases() 22 23import importlib 24import inspect 25import fnmatch 26import json 27import logging 28import os 29import sys 30 31from acts import base_test 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 39from mobly.records import ExceptionRecord 40 41 42def _find_test_class(): 43 """Finds the test class in a test script. 44 45 Walk through module members and find the subclass of BaseTestClass. Only 46 one subclass is allowed in a test script. 47 48 Returns: 49 The test class in the test module. 50 """ 51 test_classes = [] 52 main_module_members = sys.modules['__main__'] 53 for _, module_member in main_module_members.__dict__.items(): 54 if inspect.isclass(module_member): 55 if issubclass(module_member, base_test.BaseTestClass): 56 test_classes.append(module_member) 57 if len(test_classes) != 1: 58 logging.error('Expected 1 test class per file, found %s.', 59 [t.__name__ for t in test_classes]) 60 sys.exit(1) 61 return test_classes[0] 62 63 64def execute_one_test_class(test_class, test_config, test_identifier): 65 """Executes one specific test class. 66 67 You could call this function in your own cli test entry point if you choose 68 not to use act.py. 69 70 Args: 71 test_class: A subclass of acts.base_test.BaseTestClass that has the test 72 logic to be executed. 73 test_config: A dict representing one set of configs for a test run. 74 test_identifier: A list of tuples specifying which test cases to run in 75 the test class. 76 77 Returns: 78 True if all tests passed without any error, False otherwise. 79 80 Raises: 81 If signals.TestAbortAll is raised by a test run, pipe it through. 82 """ 83 tr = TestRunner(test_config, test_identifier) 84 try: 85 tr.run(test_class) 86 return tr.results.is_all_pass 87 except signals.TestAbortAll: 88 raise 89 except: 90 logging.exception('Exception when executing %s.', tr.testbed_name) 91 finally: 92 tr.stop() 93 94 95class TestRunner(object): 96 """The class that instantiates test classes, executes test cases, and 97 report results. 98 99 Attributes: 100 test_run_config: The TestRunConfig object specifying what tests to run. 101 id: A string that is the unique identifier of this test run. 102 log: The logger object used throughout this test run. 103 test_classes: A dictionary where we can look up the test classes by name 104 to instantiate. Supports unix shell style wildcards. 105 run_list: A list of tuples specifying what tests to run. 106 results: The test result object used to record the results of this test 107 run. 108 running: A boolean signifies whether this test run is ongoing or not. 109 """ 110 111 def __init__(self, test_configs, run_list): 112 self.test_run_config = test_configs 113 self.testbed_name = self.test_run_config.testbed_name 114 start_time = logger.get_log_file_timestamp() 115 self.id = '{}@{}'.format(self.testbed_name, start_time) 116 self.test_run_config.log_path = os.path.abspath( 117 os.path.join(self.test_run_config.log_path, self.testbed_name, 118 start_time)) 119 logger.setup_test_logger(self.log_path, self.testbed_name) 120 self.log = logging.getLogger() 121 self.test_run_config.summary_writer = records.TestSummaryWriter( 122 os.path.join(self.log_path, records.OUTPUT_FILE_SUMMARY)) 123 self.run_list = run_list 124 self.dump_config() 125 self.results = records.TestResult() 126 self.running = False 127 self.usage_publisher = UsageMetadataPublisher() 128 129 @property 130 def log_path(self): 131 """The path to write logs of this test run to.""" 132 return self.test_run_config.log_path 133 134 @property 135 def summary_writer(self): 136 """The object responsible for writing summary and results data.""" 137 return self.test_run_config.summary_writer 138 139 def import_test_modules(self, test_paths): 140 """Imports test classes from test scripts. 141 142 1. Locate all .py files under test paths. 143 2. Import the .py files as modules. 144 3. Find the module members that are test classes. 145 4. Categorize the test classes by name. 146 147 Args: 148 test_paths: A list of directory paths where the test files reside. 149 150 Returns: 151 A dictionary where keys are test class name strings, values are 152 actual test classes that can be instantiated. 153 """ 154 155 def is_testfile_name(name, ext): 156 if ext == '.py': 157 if name.endswith('Test') or name.endswith('_test'): 158 return True 159 return False 160 161 file_list = utils.find_files(test_paths, is_testfile_name) 162 test_classes = {} 163 for path, name, _ in file_list: 164 sys.path.append(path) 165 try: 166 with utils.SuppressLogOutput( 167 log_levels=[logging.INFO, logging.ERROR]): 168 module = importlib.import_module(name) 169 except Exception as e: 170 logging.debug('Failed to import %s: %s', path, str(e)) 171 for test_cls_name, _ in self.run_list: 172 alt_name = name.replace('_', '').lower() 173 alt_cls_name = test_cls_name.lower() 174 # Only block if a test class on the run list causes an 175 # import error. We need to check against both naming 176 # conventions: AaaBbb and aaa_bbb. 177 if name == test_cls_name or alt_name == alt_cls_name: 178 msg = ('Encountered error importing test class %s, ' 179 'abort.') % test_cls_name 180 # This exception is logged here to help with debugging 181 # under py2, because "raise X from Y" syntax is only 182 # supported under py3. 183 self.log.exception(msg) 184 raise ValueError(msg) 185 continue 186 for member_name in dir(module): 187 if not member_name.startswith('__'): 188 if member_name.endswith('Test'): 189 test_class = getattr(module, member_name) 190 if inspect.isclass(test_class): 191 test_classes[member_name] = test_class 192 return test_classes 193 194 def run_test_class(self, test_cls_name, test_cases=None): 195 """Instantiates and executes a test class. 196 197 If test_cases is None, the test cases listed by self.tests will be 198 executed instead. If self.tests is empty as well, no test case in this 199 test class will be executed. 200 201 Args: 202 test_cls_name: Name of the test class to execute. 203 test_cases: List of test case names to execute within the class. 204 205 Raises: 206 ValueError is raised if the requested test class could not be found 207 in the test_paths directories. 208 """ 209 matches = fnmatch.filter(self.test_classes.keys(), test_cls_name) 210 if not matches: 211 self.log.info( 212 'Cannot find test class %s or classes matching pattern, ' 213 'skipping for now.' % test_cls_name) 214 record = records.TestResultRecord('*all*', test_cls_name) 215 record.test_skip(signals.TestSkip('Test class does not exist.')) 216 self.results.add_record(record) 217 return 218 if matches != [test_cls_name]: 219 self.log.info('Found classes matching pattern %s: %s', 220 test_cls_name, matches) 221 222 for test_cls_name_match in matches: 223 test_cls = self.test_classes[test_cls_name_match] 224 test_cls_instance = test_cls(self.test_run_config) 225 try: 226 cls_result = test_cls_instance.run(test_cases) 227 self.results += cls_result 228 except signals.TestAbortAll as e: 229 self.results += e.results 230 raise e 231 232 def run(self, test_class=None): 233 """Executes test cases. 234 235 This will instantiate controller and test classes, and execute test 236 classes. This can be called multiple times to repeatedly execute the 237 requested test cases. 238 239 A call to TestRunner.stop should eventually happen to conclude the life 240 cycle of a TestRunner. 241 242 Args: 243 test_class: The python module of a test class. If provided, run this 244 class; otherwise, import modules in under test_paths 245 based on run_list. 246 """ 247 if not self.running: 248 self.running = True 249 250 if test_class: 251 self.test_classes = {test_class.__name__: test_class} 252 else: 253 t_paths = self.test_run_config.controller_configs[ 254 keys.Config.key_test_paths.value] 255 self.test_classes = self.import_test_modules(t_paths) 256 self.log.debug('Executing run list %s.', self.run_list) 257 for test_cls_name, test_case_names in self.run_list: 258 if not self.running: 259 break 260 261 if test_case_names: 262 self.log.debug('Executing test cases %s in test class %s.', 263 test_case_names, test_cls_name) 264 else: 265 self.log.debug('Executing test class %s', test_cls_name) 266 267 try: 268 self.run_test_class(test_cls_name, test_case_names) 269 except error.ActsError as e: 270 self.results.error.append(ExceptionRecord(e)) 271 self.log.error('Test Runner Error: %s' % e.details) 272 except signals.TestAbortAll as e: 273 self.log.warning( 274 'Abort all subsequent test classes. Reason: %s', e) 275 raise 276 277 def stop(self): 278 """Releases resources from test run. Should always be called after 279 TestRunner.run finishes. 280 281 This function concludes a test run and writes out a test report. 282 """ 283 if self.running: 284 msg = '\nSummary for test run %s: %s\n' % ( 285 self.id, self.results.summary_str()) 286 self._write_results_to_file() 287 self.log.info(msg.strip()) 288 logger.kill_test_logger(self.log) 289 self.usage_publisher.publish() 290 self.running = False 291 292 def _write_results_to_file(self): 293 """Writes test results to file(s) in a serializable format.""" 294 # Old JSON format 295 path = os.path.join(self.log_path, 'test_run_summary.json') 296 with open(path, 'w') as f: 297 f.write(self.results.json_str()) 298 # New YAML format 299 self.summary_writer.dump(self.results.summary_dict(), 300 records.TestSummaryEntryType.SUMMARY) 301 302 def dump_config(self): 303 """Writes the test config to a JSON file under self.log_path""" 304 config_path = os.path.join(self.log_path, 'test_configs.json') 305 with open(config_path, 'a') as f: 306 json.dump(dict( 307 itertools.chain( 308 self.test_run_config.user_params.items(), 309 self.test_run_config.controller_configs.items())), 310 f, 311 skipkeys=True, 312 indent=4) 313 314 def write_test_campaign(self): 315 """Log test campaign file.""" 316 path = os.path.join(self.log_path, 'test_campaign.log') 317 with open(path, 'w') as f: 318 for test_class, test_cases in self.run_list: 319 f.write('%s:\n%s' % (test_class, ',\n'.join(test_cases))) 320 f.write('\n\n') 321