| #!/usr/bin/env python3 |
| # |
| # Copyright 2016 - The Android Open Source Project |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| import itertools |
| |
| from acts.metrics.loggers.usage_metadata_logger import UsageMetadataPublisher |
| from future import standard_library |
| |
| standard_library.install_aliases() |
| |
| import importlib |
| import inspect |
| import fnmatch |
| import json |
| import logging |
| import os |
| import pkgutil |
| import sys |
| |
| from acts import base_test |
| from acts import keys |
| from acts import logger |
| from acts import records |
| from acts import signals |
| from acts import utils |
| from acts import error |
| |
| from mobly.records import ExceptionRecord |
| |
| |
| def _find_test_class(): |
| """Finds the test class in a test script. |
| |
| Walk through module members and find the subclass of BaseTestClass. Only |
| one subclass is allowed in a test script. |
| |
| Returns: |
| The test class in the test module. |
| """ |
| test_classes = [] |
| main_module_members = sys.modules['__main__'] |
| for _, module_member in main_module_members.__dict__.items(): |
| if inspect.isclass(module_member): |
| if issubclass(module_member, base_test.BaseTestClass): |
| test_classes.append(module_member) |
| if len(test_classes) != 1: |
| logging.error('Expected 1 test class per file, found %s.', |
| [t.__name__ for t in test_classes]) |
| sys.exit(1) |
| return test_classes[0] |
| |
| |
| def execute_one_test_class(test_class, test_config, test_identifier): |
| """Executes one specific test class. |
| |
| You could call this function in your own cli test entry point if you choose |
| not to use act.py. |
| |
| Args: |
| test_class: A subclass of acts.base_test.BaseTestClass that has the test |
| logic to be executed. |
| test_config: A dict representing one set of configs for a test run. |
| test_identifier: A list of tuples specifying which test cases to run in |
| the test class. |
| |
| Returns: |
| True if all tests passed without any error, False otherwise. |
| |
| Raises: |
| If signals.TestAbortAll is raised by a test run, pipe it through. |
| """ |
| tr = TestRunner(test_config, test_identifier) |
| try: |
| tr.run(test_class) |
| return tr.results.is_all_pass |
| except signals.TestAbortAll: |
| raise |
| except: |
| logging.exception('Exception when executing %s.', tr.testbed_name) |
| finally: |
| tr.stop() |
| |
| |
| class TestRunner(object): |
| """The class that instantiates test classes, executes test cases, and |
| report results. |
| |
| Attributes: |
| test_run_config: The TestRunConfig object specifying what tests to run. |
| id: A string that is the unique identifier of this test run. |
| log: The logger object used throughout this test run. |
| test_classes: A dictionary where we can look up the test classes by name |
| to instantiate. Supports unix shell style wildcards. |
| run_list: A list of tuples specifying what tests to run. |
| results: The test result object used to record the results of this test |
| run. |
| running: A boolean signifies whether this test run is ongoing or not. |
| """ |
| def __init__(self, test_configs, run_list): |
| self.test_run_config = test_configs |
| self.testbed_name = self.test_run_config.testbed_name |
| start_time = logger.get_log_file_timestamp() |
| self.id = '{}@{}'.format(self.testbed_name, start_time) |
| self.test_run_config.log_path = os.path.abspath( |
| os.path.join(self.test_run_config.log_path, self.testbed_name, |
| start_time)) |
| logger.setup_test_logger(self.log_path, self.testbed_name) |
| self.log = logging.getLogger() |
| self.test_run_config.summary_writer = records.TestSummaryWriter( |
| os.path.join(self.log_path, records.OUTPUT_FILE_SUMMARY)) |
| self.run_list = run_list |
| self.dump_config() |
| self.results = records.TestResult() |
| self.running = False |
| self.usage_publisher = UsageMetadataPublisher() |
| |
| @property |
| def log_path(self): |
| """The path to write logs of this test run to.""" |
| return self.test_run_config.log_path |
| |
| @property |
| def summary_writer(self): |
| """The object responsible for writing summary and results data.""" |
| return self.test_run_config.summary_writer |
| |
| def import_test_modules(self, test_paths): |
| """Imports test classes from test scripts. |
| |
| 1. Locate all .py files under test paths. |
| 2. Import the .py files as modules. |
| 3. Find the module members that are test classes. |
| 4. Categorize the test classes by name. |
| |
| Args: |
| test_paths: A list of directory paths where the test files reside. |
| |
| Returns: |
| A dictionary where keys are test class name strings, values are |
| actual test classes that can be instantiated. |
| """ |
| def is_testfile_name(name, ext): |
| if ext == '.py': |
| if name.endswith('Test') or name.endswith('_test'): |
| return True |
| return False |
| |
| file_list = utils.find_files(test_paths, is_testfile_name) |
| test_classes = {} |
| for path, name, _ in file_list: |
| sys.path.append(path) |
| try: |
| with utils.SuppressLogOutput( |
| log_levels=[logging.INFO, logging.ERROR]): |
| module = importlib.import_module(name) |
| except Exception as e: |
| logging.debug('Failed to import %s: %s', path, str(e)) |
| for test_cls_name, _ in self.run_list: |
| alt_name = name.replace('_', '').lower() |
| alt_cls_name = test_cls_name.lower() |
| # Only block if a test class on the run list causes an |
| # import error. We need to check against both naming |
| # conventions: AaaBbb and aaa_bbb. |
| if name == test_cls_name or alt_name == alt_cls_name: |
| msg = ('Encountered error importing test class %s, ' |
| 'abort.') % test_cls_name |
| # This exception is logged here to help with debugging |
| # under py2, because "raise X from Y" syntax is only |
| # supported under py3. |
| self.log.exception(msg) |
| raise ValueError(msg) |
| continue |
| for member_name in dir(module): |
| if not member_name.startswith('__'): |
| if member_name.endswith('Test'): |
| test_class = getattr(module, member_name) |
| if inspect.isclass(test_class): |
| test_classes[member_name] = test_class |
| return test_classes |
| |
| def run_test_class(self, test_cls_name, test_cases=None): |
| """Instantiates and executes a test class. |
| |
| If test_cases is None, the test cases listed by self.tests will be |
| executed instead. If self.tests is empty as well, no test case in this |
| test class will be executed. |
| |
| Args: |
| test_cls_name: Name of the test class to execute. |
| test_cases: List of test case names to execute within the class. |
| |
| Raises: |
| ValueError is raised if the requested test class could not be found |
| in the test_paths directories. |
| """ |
| matches = fnmatch.filter(self.test_classes.keys(), test_cls_name) |
| if not matches: |
| self.log.info( |
| 'Cannot find test class %s or classes matching pattern, ' |
| 'skipping for now.' % test_cls_name) |
| record = records.TestResultRecord('*all*', test_cls_name) |
| record.test_skip(signals.TestSkip('Test class does not exist.')) |
| self.results.add_record(record) |
| return |
| if matches != [test_cls_name]: |
| self.log.info('Found classes matching pattern %s: %s', |
| test_cls_name, matches) |
| |
| for test_cls_name_match in matches: |
| test_cls = self.test_classes[test_cls_name_match] |
| test_cls_instance = test_cls(self.test_run_config) |
| try: |
| cls_result = test_cls_instance.run(test_cases) |
| self.results += cls_result |
| except signals.TestAbortAll as e: |
| self.results += e.results |
| raise e |
| |
| def run(self, test_class=None): |
| """Executes test cases. |
| |
| This will instantiate controller and test classes, and execute test |
| classes. This can be called multiple times to repeatedly execute the |
| requested test cases. |
| |
| A call to TestRunner.stop should eventually happen to conclude the life |
| cycle of a TestRunner. |
| |
| Args: |
| test_class: The python module of a test class. If provided, run this |
| class; otherwise, import modules in under test_paths |
| based on run_list. |
| """ |
| if not self.running: |
| self.running = True |
| |
| if test_class: |
| self.test_classes = {test_class.__name__: test_class} |
| else: |
| t_paths = self.test_run_config.controller_configs[ |
| keys.Config.key_test_paths.value] |
| self.test_classes = self.import_test_modules(t_paths) |
| self.log.debug('Executing run list %s.', self.run_list) |
| for test_cls_name, test_case_names in self.run_list: |
| if not self.running: |
| break |
| |
| if test_case_names: |
| self.log.debug('Executing test cases %s in test class %s.', |
| test_case_names, test_cls_name) |
| else: |
| self.log.debug('Executing test class %s', test_cls_name) |
| |
| try: |
| self.run_test_class(test_cls_name, test_case_names) |
| except error.ActsError as e: |
| self.results.error.append(ExceptionRecord(e)) |
| self.log.error('Test Runner Error: %s' % e.details) |
| except signals.TestAbortAll as e: |
| self.log.warning( |
| 'Abort all subsequent test classes. Reason: %s', e) |
| raise |
| |
| def stop(self): |
| """Releases resources from test run. Should always be called after |
| TestRunner.run finishes. |
| |
| This function concludes a test run and writes out a test report. |
| """ |
| if self.running: |
| msg = '\nSummary for test run %s: %s\n' % ( |
| self.id, self.results.summary_str()) |
| self._write_results_to_file() |
| self.log.info(msg.strip()) |
| logger.kill_test_logger(self.log) |
| self.usage_publisher.publish() |
| self.running = False |
| |
| def _write_results_to_file(self): |
| """Writes test results to file(s) in a serializable format.""" |
| # Old JSON format |
| path = os.path.join(self.log_path, 'test_run_summary.json') |
| with open(path, 'w') as f: |
| f.write(self.results.json_str()) |
| # New YAML format |
| self.summary_writer.dump(self.results.summary_dict(), |
| records.TestSummaryEntryType.SUMMARY) |
| |
| def dump_config(self): |
| """Writes the test config to a JSON file under self.log_path""" |
| config_path = os.path.join(self.log_path, 'test_configs.json') |
| with open(config_path, 'a') as f: |
| json.dump(dict( |
| itertools.chain( |
| self.test_run_config.user_params.items(), |
| self.test_run_config.controller_configs.items())), |
| f, |
| skipkeys=True, |
| indent=4) |
| |
| def write_test_campaign(self): |
| """Log test campaign file.""" |
| path = os.path.join(self.log_path, 'test_campaign.log') |
| with open(path, 'w') as f: |
| for test_class, test_cases in self.run_list: |
| f.write('%s:\n%s' % (test_class, ',\n'.join(test_cases))) |
| f.write('\n\n') |
