# # 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())