| # |
| # Copyright (C) 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. |
| # |
| |
| from future import standard_library |
| standard_library.install_aliases() |
| |
| import copy |
| import importlib |
| import inspect |
| import logging |
| import os |
| import pkgutil |
| import signal |
| import sys |
| import threading |
| |
| from vts.runners.host import base_test |
| from vts.runners.host import config_parser |
| from vts.runners.host import keys |
| from vts.runners.host import logger |
| from vts.runners.host import records |
| from vts.runners.host import signals |
| from vts.runners.host import utils |
| from vts.utils.python.common import timeout_utils |
| from vts.utils.python.instrumentation import test_framework_instrumentation as tfi |
| |
| |
| def main(): |
| """Execute the test class in a test module. |
| |
| This is to be used in a test script's main so the script can be executed |
| directly. It will discover all the classes that inherit from BaseTestClass |
| and excute them. all the test results will be aggregated into one. |
| |
| A VTS host-driven test case has three args: |
| 1st arg: the path of a test case config file. |
| 2nd arg: the serial ID of a target device (device config). |
| 3rd arg: the path of a test case data dir. |
| |
| Returns: |
| The TestResult object that holds the results of the test run. |
| """ |
| event = tfi.Begin('Test runner main method') |
| 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) |
| # TODO(angli): Need to handle the case where more than one test class is in |
| # a test script. The challenge is to handle multiple configs and how to do |
| # default config in this case. |
| if len(test_classes) != 1: |
| logging.error("Expected 1 test class per file, found %s (%s).", |
| len(test_classes), test_classes) |
| sys.exit(1) |
| test_result = runTestClass(test_classes[0]) |
| event.End() |
| tfi.CompileResults() |
| return test_result |
| |
| |
| def runTestClass(test_class): |
| """Execute one test class. |
| |
| This will create a TestRunner, execute one test run with one test class. |
| |
| Args: |
| test_class: The test class to instantiate and execute. |
| |
| Returns: |
| The TestResult object that holds the results of the test run. |
| """ |
| test_cls_name = test_class.__name__ |
| if len(sys.argv) < 2: |
| logging.warning("Missing a configuration file. Using the default.") |
| test_configs = [config_parser.GetDefaultConfig(test_cls_name)] |
| else: |
| try: |
| config_path = sys.argv[1] |
| baseline_config = config_parser.GetDefaultConfig(test_cls_name) |
| baseline_config[keys.ConfigKeys.KEY_TESTBED] = [ |
| baseline_config[keys.ConfigKeys.KEY_TESTBED] |
| ] |
| test_configs = config_parser.load_test_config_file( |
| config_path, baseline_config=baseline_config) |
| except IndexError: |
| logging.error("No valid config file found.") |
| sys.exit(1) |
| except Exception as e: |
| logging.error("Unexpected exception") |
| logging.exception(e) |
| |
| test_identifiers = [(test_cls_name, None)] |
| |
| for config in test_configs: |
| def watchStdin(): |
| while True: |
| line = sys.stdin.readline() |
| if not line: |
| break |
| utils.stop_current_process(base_test.TIMEOUT_SECS_TEARDOWN_CLASS) |
| |
| watcher_thread = threading.Thread(target=watchStdin, name="watchStdin") |
| watcher_thread.daemon = True |
| watcher_thread.start() |
| |
| tr = TestRunner(config, test_identifiers) |
| tr.parseTestConfig(config) |
| try: |
| tr.runTestClass(test_class, None) |
| except (signals.TestAbortAll, KeyboardInterrupt) as e: |
| logging.error("Abort all test") |
| except Exception as e: |
| logging.error("Unexpected exception") |
| logging.exception(e) |
| finally: |
| tr.stop() |
| return tr.results |
| |
| |
| class TestRunner(object): |
| """The class that instantiates test classes, executes test cases, and |
| report results. |
| |
| Attributes: |
| test_run_info: A dictionary containing the information needed by |
| test classes for this test run, including params, |
| controllers, and other objects. All of these will |
| be passed to test classes. |
| test_configs: A dictionary that is the original test configuration |
| passed in by user. |
| id: A string that is the unique identifier of this test run. |
| log_path: A string representing the path of the dir under which |
| all logs from this test run should be written. |
| controller_registry: A dictionary that holds the controller |
| objects used in a test run. |
| controller_destructors: A dictionary that holds the controller |
| distructors. Keys are controllers' names. |
| 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. |
| test_cls_instances: list of test class instances that were executed |
| or scheduled to be executed. |
| log_severity: string, log severity level for the test logger. |
| Currently, this parameter only affects how logs are displayed |
| to the console, and is not recommended to be used. |
| """ |
| |
| def __init__(self, test_configs, run_list): |
| self.test_run_info = {} |
| self.test_run_info[keys.ConfigKeys.IKEY_DATA_FILE_PATH] = getattr( |
| test_configs, keys.ConfigKeys.IKEY_DATA_FILE_PATH, "./") |
| self.test_configs = test_configs |
| self.testbed_configs = self.test_configs[keys.ConfigKeys.KEY_TESTBED] |
| self.testbed_name = self.testbed_configs[ |
| keys.ConfigKeys.KEY_TESTBED_NAME] |
| start_time = logger.getLogFileTimestamp() |
| self.id = "{}@{}".format(self.testbed_name, start_time) |
| # log_path should be set before parsing configs. |
| l_path = os.path.join(self.test_configs[keys.ConfigKeys.KEY_LOG_PATH], |
| self.testbed_name, start_time) |
| self.log_path = os.path.abspath(l_path) |
| self.log_severity = self.test_configs.get( |
| keys.ConfigKeys.KEY_LOG_SEVERITY, "INFO").upper() |
| logger.setupTestLogger( |
| self.log_path, |
| self.testbed_name, |
| filename="test_run_details.txt", |
| log_severity=self.log_severity) |
| self.controller_registry = {} |
| self.controller_destructors = {} |
| self.run_list = run_list |
| self.results = records.TestResult() |
| self.running = False |
| self.test_cls_instances = [] |
| |
| def __enter__(self): |
| return self |
| |
| def __exit__(self, *args): |
| self.stop() |
| |
| def importTestModules(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: |
| module = importlib.import_module(name) |
| except: |
| 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. |
| logging.exception(msg) |
| raise USERError(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 verifyControllerModule(self, module): |
| """Verifies a module object follows the required interface for |
| controllers. |
| |
| Args: |
| module: An object that is a controller module. This is usually |
| imported with import statements or loaded by importlib. |
| |
| Raises: |
| ControllerError is raised if the module does not match the vts.runners.host |
| controller interface, or one of the required members is null. |
| """ |
| required_attributes = ("create", "destroy", |
| "VTS_CONTROLLER_CONFIG_NAME") |
| for attr in required_attributes: |
| if not hasattr(module, attr): |
| raise signals.ControllerError( |
| ("Module %s missing required " |
| "controller module attribute %s.") % (module.__name__, |
| attr)) |
| if not getattr(module, attr): |
| raise signals.ControllerError( |
| ("Controller interface %s in %s " |
| "cannot be null.") % (attr, module.__name__)) |
| |
| def registerController(self, module, start_services=True): |
| """Registers a controller module for a test run. |
| |
| This declares a controller dependency of this test class. If the target |
| module exists and matches the controller interface, the controller |
| module will be instantiated with corresponding configs in the test |
| config file. The module should be imported first. |
| |
| Params: |
| module: A module that follows the controller module interface. |
| start_services: boolean, controls whether services (e.g VTS agent) |
| are started on the target. |
| |
| Returns: |
| A list of controller objects instantiated from controller_module. |
| |
| Raises: |
| ControllerError is raised if no corresponding config can be found, |
| or if the controller module has already been registered. |
| """ |
| event = tfi.Begin('test_runner registerController', |
| tfi.categories.FRAMEWORK_SETUP) |
| logging.debug("cwd: %s", os.getcwd()) |
| logging.info("adb devices: %s", module.list_adb_devices()) |
| self.verifyControllerModule(module) |
| module_ref_name = module.__name__.split('.')[-1] |
| if module_ref_name in self.controller_registry: |
| event.End() |
| raise signals.ControllerError( |
| ("Controller module %s has already " |
| "been registered. It can not be " |
| "registered again.") % module_ref_name) |
| # Create controller objects. |
| create = module.create |
| module_config_name = module.VTS_CONTROLLER_CONFIG_NAME |
| if module_config_name not in self.testbed_configs: |
| msg = "No corresponding config found for %s" % module_config_name |
| event.Remove(msg) |
| raise signals.ControllerError(msg) |
| try: |
| # Make a deep copy of the config to pass to the controller module, |
| # in case the controller module modifies the config internally. |
| original_config = self.testbed_configs[module_config_name] |
| controller_config = copy.deepcopy(original_config) |
| # Add log_severity config to device controller config. |
| if isinstance(controller_config, list): |
| for config in controller_config: |
| if isinstance(config, dict): |
| config["log_severity"] = self.log_severity |
| logging.debug("controller_config: %s", controller_config) |
| if "use_vts_agent" not in self.testbed_configs: |
| objects = create(controller_config, start_services) |
| else: |
| objects = create(controller_config, |
| self.testbed_configs["use_vts_agent"]) |
| except: |
| msg = "Failed to initialize objects for controller %s, abort!" % module_config_name |
| event.Remove(msg) |
| logging.error(msg) |
| raise |
| if not isinstance(objects, list): |
| msg = "Controller module %s did not return a list of objects, abort." % module_ref_name |
| event.Remove(msg) |
| raise signals.ControllerError(msg) |
| |
| self.controller_registry[module_ref_name] = objects |
| logging.debug("Found %d objects for controller %s", |
| len(objects), module_config_name) |
| destroy_func = module.destroy |
| self.controller_destructors[module_ref_name] = destroy_func |
| event.End() |
| return objects |
| |
| @timeout_utils.timeout(base_test.TIMEOUT_SECS_TEARDOWN_CLASS, |
| message='unregisterControllers method timed out.', |
| no_exception=True) |
| def unregisterControllers(self): |
| """Destroy controller objects and clear internal registry. |
| |
| This will be called at the end of each TestRunner.run call. |
| """ |
| event = tfi.Begin('test_runner unregisterControllers', |
| tfi.categories.FRAMEWORK_TEARDOWN) |
| for name, destroy in self.controller_destructors.items(): |
| try: |
| logging.debug("Destroying %s.", name) |
| dut = self.controller_destructors[name] |
| destroy(self.controller_registry[name]) |
| except: |
| logging.exception("Exception occurred destroying %s.", name) |
| self.controller_registry = {} |
| self.controller_destructors = {} |
| event.End() |
| |
| def parseTestConfig(self, test_configs): |
| """Parses the test configuration and unpacks objects and parameters |
| into a dictionary to be passed to test classes. |
| |
| Args: |
| test_configs: A json object representing the test configurations. |
| """ |
| self.test_run_info[ |
| keys.ConfigKeys.IKEY_TESTBED_NAME] = self.testbed_name |
| # Unpack other params. |
| self.test_run_info["registerController"] = self.registerController |
| self.test_run_info[keys.ConfigKeys.IKEY_LOG_PATH] = self.log_path |
| user_param_pairs = [] |
| for item in test_configs.items(): |
| if item[0] not in keys.ConfigKeys.RESERVED_KEYS: |
| user_param_pairs.append(item) |
| self.test_run_info[keys.ConfigKeys.IKEY_USER_PARAM] = copy.deepcopy( |
| dict(user_param_pairs)) |
| |
| def runTestClass(self, test_cls, 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: The test class to be instantiated and executed. |
| test_cases: List of test case names to execute within the class. |
| |
| Returns: |
| A tuple, with the number of cases passed at index 0, and the total |
| number of test cases at index 1. |
| """ |
| self.running = True |
| with test_cls(self.test_run_info) as test_cls_instance: |
| try: |
| if test_cls_instance not in self.test_cls_instances: |
| self.test_cls_instances.append(test_cls_instance) |
| cls_result = test_cls_instance.run(test_cases) |
| finally: |
| self.unregisterControllers() |
| |
| def run(self): |
| """Executes test cases. |
| |
| This will instantiate controller and test classes, and execute test |
| classes. This can be called multiple times to repeatly execute the |
| requested test cases. |
| |
| A call to TestRunner.stop should eventually happen to conclude the life |
| cycle of a TestRunner. |
| |
| Args: |
| test_classes: A dictionary where the key is test class name, and |
| the values are actual test classes. |
| """ |
| if not self.running: |
| self.running = True |
| # Initialize controller objects and pack appropriate objects/params |
| # to be passed to test class. |
| self.parseTestConfig(self.test_configs) |
| t_configs = self.test_configs[keys.ConfigKeys.KEY_TEST_PATHS] |
| test_classes = self.importTestModules(t_configs) |
| logging.debug("Executing run list %s.", self.run_list) |
| try: |
| for test_cls_name, test_case_names in self.run_list: |
| if not self.running: |
| break |
| if test_case_names: |
| logging.debug("Executing test cases %s in test class %s.", |
| test_case_names, test_cls_name) |
| else: |
| logging.debug("Executing test class %s", test_cls_name) |
| try: |
| test_cls = test_classes[test_cls_name] |
| except KeyError: |
| raise USERError( |
| ("Unable to locate class %s in any of the test " |
| "paths specified.") % test_cls_name) |
| try: |
| self.runTestClass(test_cls, test_case_names) |
| except signals.TestAbortAll as e: |
| logging.warning( |
| ("Abort all subsequent test classes. Reason: " |
| "%s"), e) |
| raise |
| except Exception as e: |
| logging.error("Unexpected exception") |
| logging.exception(e) |
| finally: |
| self.unregisterControllers() |
| |
| 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: |
| |
| for test_cls_instance in self.test_cls_instances: |
| self.results += test_cls_instance.results |
| |
| msg = "\nSummary for test run %s: %s\n" % (self.id, |
| self.results.summary()) |
| self._writeResultsJsonString() |
| logging.info(msg.strip()) |
| logger.killTestLogger(logging.getLogger()) |
| self.running = False |
| |
| def _writeResultsJsonString(self): |
| """Writes out a json file with the test result info for easy parsing. |
| """ |
| path = os.path.join(self.log_path, "test_run_summary.json") |
| with open(path, 'w') as f: |
| f.write(self.results.jsonString()) |
