| #!/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 fnmatch |
| import importlib |
| import logging |
| import os |
| import traceback |
| from concurrent.futures import ThreadPoolExecutor |
| |
| from acts import asserts |
| from acts import keys |
| from acts import logger |
| from acts import records |
| from acts import signals |
| from acts import error |
| from acts import tracelogger |
| from acts import utils |
| from acts.event import event_bus |
| from acts.event import subscription_bundle |
| from acts.event.decorators import subscribe_static |
| from acts.event.event import TestCaseBeginEvent |
| from acts.event.event import TestCaseEndEvent |
| from acts.event.event import TestClassBeginEvent |
| from acts.event.event import TestClassEndEvent |
| from acts.event.subscription_bundle import SubscriptionBundle |
| |
| from mobly import controller_manager |
| from mobly.base_test import BaseTestClass as MoblyBaseTest |
| from mobly.records import ExceptionRecord |
| |
| # Macro strings for test result reporting |
| TEST_CASE_TOKEN = "[Test Case]" |
| RESULT_LINE_TEMPLATE = TEST_CASE_TOKEN + " %s %s" |
| |
| |
| @subscribe_static(TestCaseBeginEvent) |
| def _logcat_log_test_begin(event): |
| """Ensures that logcat is running. Write a logcat line indicating test case |
| begin.""" |
| test_instance = event.test_class |
| try: |
| for ad in getattr(test_instance, 'android_devices', []): |
| if not ad.is_adb_logcat_on: |
| ad.start_adb_logcat() |
| # Write test start token to adb log if android device is attached. |
| if not ad.skip_sl4a: |
| ad.droid.logV("%s BEGIN %s" % (TEST_CASE_TOKEN, |
| event.test_case_name)) |
| |
| except error.ActsError as e: |
| test_instance.results.error.append( |
| ExceptionRecord( |
| e, 'Logcat for test begin: %s' % event.test_case_name)) |
| test_instance.log.error('BaseTest setup_test error: %s' % e.message) |
| |
| except Exception as e: |
| test_instance.log.warning( |
| 'Unable to send BEGIN log command to all devices.') |
| test_instance.log.warning('Error: %s' % e) |
| |
| |
| @subscribe_static(TestCaseEndEvent) |
| def _logcat_log_test_end(event): |
| """Write a logcat line indicating test case end.""" |
| test_instance = event.test_class |
| try: |
| # Write test end token to adb log if android device is attached. |
| for ad in getattr(test_instance, 'android_devices', []): |
| if not ad.skip_sl4a: |
| ad.droid.logV("%s END %s" % (TEST_CASE_TOKEN, |
| event.test_case_name)) |
| |
| except error.ActsError as e: |
| test_instance.results.error.append( |
| ExceptionRecord( |
| e, 'Logcat for test end: %s' % event.test_case_name)) |
| test_instance.log.error('BaseTest teardown_test error: %s' % e.message) |
| |
| except Exception as e: |
| test_instance.log.warning( |
| 'Unable to send END log command to all devices.') |
| test_instance.log.warning('Error: %s' % e) |
| |
| |
| @subscribe_static(TestCaseBeginEvent) |
| def _syslog_log_test_begin(event): |
| """This adds a BEGIN log message with the test name to the syslog of any |
| Fuchsia device""" |
| test_instance = event.test_class |
| try: |
| for fd in getattr(test_instance, 'fuchsia_devices', []): |
| if not fd.skip_sl4f: |
| fd.logging_lib.logI("%s BEGIN %s" % (TEST_CASE_TOKEN, |
| event.test_case_name)) |
| |
| except Exception as e: |
| test_instance.log.warning( |
| 'Unable to send BEGIN log command to all devices.') |
| test_instance.log.warning('Error: %s' % e) |
| |
| |
| @subscribe_static(TestCaseEndEvent) |
| def _syslog_log_test_end(event): |
| """This adds a END log message with the test name to the syslog of any |
| Fuchsia device""" |
| test_instance = event.test_class |
| try: |
| for fd in getattr(test_instance, 'fuchsia_devices', []): |
| if not fd.skip_sl4f: |
| fd.logging_lib.logI("%s END %s" % (TEST_CASE_TOKEN, |
| event.test_case_name)) |
| |
| except Exception as e: |
| test_instance.log.warning( |
| 'Unable to send END log command to all devices.') |
| test_instance.log.warning('Error: %s' % e) |
| |
| |
| event_bus.register_subscription(_logcat_log_test_begin.subscription) |
| event_bus.register_subscription(_logcat_log_test_end.subscription) |
| event_bus.register_subscription(_syslog_log_test_begin.subscription) |
| event_bus.register_subscription(_syslog_log_test_end.subscription) |
| |
| |
| class Error(Exception): |
| """Raised for exceptions that occured in BaseTestClass.""" |
| |
| |
| class BaseTestClass(MoblyBaseTest): |
| """Base class for all test classes to inherit from. Inherits some |
| functionality from Mobly's base test class. |
| |
| This class gets all the controller objects from test_runner and executes |
| the test cases requested within itself. |
| |
| Most attributes of this class are set at runtime based on the configuration |
| provided. |
| |
| Attributes: |
| tests: A list of strings, each representing a test case name. |
| TAG: A string used to refer to a test class. Default is the test class |
| name. |
| log: A logger object used for logging. |
| results: A records.TestResult object for aggregating test results from |
| the execution of test cases. |
| consecutive_failures: Tracks the number of consecutive test case |
| failures within this class. |
| consecutive_failure_limit: Number of consecutive test failures to allow |
| before blocking remaining tests in the same |
| test class. |
| size_limit_reached: True if the size of the log directory has reached |
| its limit. |
| current_test_name: A string that's the name of the test case currently |
| being executed. If no test is executing, this should |
| be None. |
| """ |
| |
| TAG = None |
| |
| def __init__(self, configs): |
| self.class_subscriptions = SubscriptionBundle() |
| self.class_subscriptions.register() |
| self.all_subscriptions = [self.class_subscriptions] |
| |
| self.tests = [] |
| if not self.TAG: |
| self.TAG = self.__class__.__name__ |
| # Set all the controller objects and params. |
| self.user_params = {} |
| self.testbed_configs = {} |
| for name, value in configs.items(): |
| setattr(self, name, value) |
| self.results = records.TestResult() |
| self.current_test_name = None |
| self.log = tracelogger.TraceLogger(self.log) |
| self.consecutive_failures = 0 |
| self.consecutive_failure_limit = self.user_params.get( |
| 'consecutive_failure_limit', -1) |
| self.size_limit_reached = False |
| |
| # Initialize a controller manager (Mobly) |
| self._controller_manager = controller_manager.ControllerManager( |
| class_name=self.__class__.__name__, |
| controller_configs=self.testbed_configs) |
| |
| # Import and register the built-in controller modules specified |
| # in testbed config. |
| for module in self._import_builtin_controllers(): |
| self.register_controller(module, builtin=True) |
| if hasattr(self, 'android_devices'): |
| for ad in self.android_devices: |
| if ad.droid: |
| utils.set_location_service(ad, False) |
| utils.sync_device_time(ad) |
| self.testbed_name = '' |
| |
| def _import_builtin_controllers(self): |
| """Import built-in controller modules. |
| |
| Go through the testbed configs, find any built-in controller configs |
| and import the corresponding controller module from acts.controllers |
| package. |
| |
| Returns: |
| A list of controller modules. |
| """ |
| builtin_controllers = [] |
| for ctrl_name in keys.Config.builtin_controller_names.value: |
| if ctrl_name in self.testbed_configs: |
| module_name = keys.get_module_name(ctrl_name) |
| module = importlib.import_module( |
| "acts.controllers.%s" % module_name) |
| builtin_controllers.append(module) |
| return builtin_controllers |
| |
| @staticmethod |
| def get_module_reference_name(a_module): |
| """Returns the module's reference name. |
| |
| This is largely for backwards compatibility with log parsing. If the |
| module defines ACTS_CONTROLLER_REFERENCE_NAME, it will return that |
| value, or the module's submodule name. |
| |
| Args: |
| a_module: Any module. Ideally, a controller module. |
| Returns: |
| A string corresponding to the module's name. |
| """ |
| if hasattr(a_module, 'ACTS_CONTROLLER_REFERENCE_NAME'): |
| return a_module.ACTS_CONTROLLER_REFERENCE_NAME |
| else: |
| return a_module.__name__.split('.')[-1] |
| |
| def register_controller(self, |
| controller_module, |
| required=True, |
| builtin=False): |
| """Registers an ACTS controller module for a test class. Invokes Mobly's |
| implementation of register_controller. |
| |
| An ACTS controller module is a Python lib that can be used to control |
| a device, service, or equipment. To be ACTS compatible, a controller |
| module needs to have the following members: |
| |
| def create(configs): |
| [Required] Creates controller objects from configurations. |
| Args: |
| configs: A list of serialized data like string/dict. Each |
| element of the list is a configuration for a |
| controller object. |
| Returns: |
| A list of objects. |
| |
| def destroy(objects): |
| [Required] Destroys controller objects created by the create |
| function. Each controller object shall be properly cleaned up |
| and all the resources held should be released, e.g. memory |
| allocation, sockets, file handlers etc. |
| Args: |
| A list of controller objects created by the create function. |
| |
| def get_info(objects): |
| [Optional] Gets info from the controller objects used in a test |
| run. The info will be included in test_result_summary.json under |
| the key "ControllerInfo". Such information could include unique |
| ID, version, or anything that could be useful for describing the |
| test bed and debugging. |
| Args: |
| objects: A list of controller objects created by the create |
| function. |
| Returns: |
| A list of json serializable objects, each represents the |
| info of a controller object. The order of the info object |
| should follow that of the input objects. |
| def get_post_job_info(controller_list): |
| [Optional] Returns information about the controller after the |
| test has run. This info is sent to test_run_summary.json's |
| "Extras" key. |
| Args: |
| The list of controller objects created by the module |
| Returns: |
| A (name, data) tuple. |
| Registering a controller module declares a test class's dependency the |
| controller. If the module config exists and the module matches the |
| controller interface, controller objects will be instantiated with |
| corresponding configs. The module should be imported first. |
| |
| Args: |
| controller_module: A module that follows the controller module |
| interface. |
| required: A bool. If True, failing to register the specified |
| controller module raises exceptions. If False, returns None upon |
| failures. |
| builtin: Specifies that the module is a builtin controller module in |
| ACTS. If true, adds itself to test attributes. |
| Returns: |
| A list of controller objects instantiated from controller_module, or |
| None. |
| |
| Raises: |
| When required is True, ControllerError is raised if no corresponding |
| config can be found. |
| Regardless of the value of "required", ControllerError is raised if |
| the controller module has already been registered or any other error |
| occurred in the registration process. |
| """ |
| module_ref_name = self.get_module_reference_name(controller_module) |
| |
| # Substitute Mobly controller's module config name with the ACTS one |
| module_config_name = controller_module.ACTS_CONTROLLER_CONFIG_NAME |
| controller_module.MOBLY_CONTROLLER_CONFIG_NAME = module_config_name |
| |
| # Get controller objects from Mobly's register_controller |
| controllers = self._controller_manager.register_controller( |
| controller_module, required=required) |
| if not controllers: |
| return None |
| |
| # Collect controller information and write to test result. |
| # Implementation of "get_info" is optional for a controller module. |
| if hasattr(controller_module, "get_info"): |
| controller_info = controller_module.get_info(controllers) |
| self.log.info("Controller %s: %s", module_config_name, |
| controller_info) |
| self.results.add_controller_info(module_config_name, |
| controller_info) |
| else: |
| self.log.warning("No controller info obtained for %s", |
| module_config_name) |
| self._record_controller_info() |
| |
| if builtin: |
| setattr(self, module_ref_name, controllers) |
| return controllers |
| |
| def unregister_controllers(self): |
| """Destroy controller objects and clear internal registry. Invokes |
| Mobly's controller manager's unregister_controllers. |
| |
| This will be called upon test class teardown. |
| """ |
| controller_modules = self._controller_manager._controller_modules |
| controller_objects = self._controller_manager._controller_objects |
| # Record post job info for the controller |
| for name, controller_module in controller_modules.items(): |
| if hasattr(controller_module, 'get_post_job_info'): |
| self.log.debug('Getting post job info for %s', name) |
| try: |
| name, value = controller_module.get_post_job_info( |
| controller_objects[name]) |
| self.results.set_extra_data(name, value) |
| self.summary_writer.dump( |
| {name: value}, records.TestSummaryEntryType.USER_DATA) |
| except: |
| self.log.error("Fail to get post job info for %s", name) |
| self._controller_manager.unregister_controllers() |
| |
| def _record_controller_info(self): |
| """Collect controller information and write to summary file.""" |
| try: |
| manager = self._controller_manager |
| for record in manager.get_controller_info_records(): |
| self.summary_writer.dump( |
| record.to_dict(), |
| records.TestSummaryEntryType.CONTROLLER_INFO) |
| except Exception: |
| self.log.exception('Unable to write controller info records to' |
| ' summary file') |
| |
| def _setup_class(self): |
| """Proxy function to guarantee the base implementation of setup_class |
| is called. |
| """ |
| event_bus.post(TestClassBeginEvent(self)) |
| return self.setup_class() |
| |
| def _teardown_class(self): |
| """Proxy function to guarantee the base implementation of teardown_class |
| is called. |
| """ |
| self.teardown_class() |
| self.unregister_controllers() |
| event_bus.post(TestClassEndEvent(self, self.results)) |
| |
| def _setup_test(self, test_name): |
| """Proxy function to guarantee the base implementation of setup_test is |
| called. |
| """ |
| self.current_test_name = test_name |
| |
| # Skip the test if the consecutive test case failure limit is reached. |
| if self.consecutive_failures == self.consecutive_failure_limit: |
| raise signals.TestError('Consecutive test failure') |
| |
| return self.setup_test() |
| |
| def setup_test(self): |
| """Setup function that will be called every time before executing each |
| test case in the test class. |
| |
| To signal setup failure, return False or raise an exception. If |
| exceptions were raised, the stack trace would appear in log, but the |
| exceptions would not propagate to upper levels. |
| |
| Implementation is optional. |
| """ |
| return True |
| |
| def _teardown_test(self, test_name): |
| """Proxy function to guarantee the base implementation of teardown_test |
| is called. |
| """ |
| self.log.debug('Tearing down test %s' % test_name) |
| self.teardown_test() |
| |
| def _on_fail(self, record): |
| """Proxy function to guarantee the base implementation of on_fail is |
| called. |
| |
| Args: |
| record: The records.TestResultRecord object for the failed test |
| case. |
| """ |
| self.consecutive_failures += 1 |
| if record.details: |
| self.log.error(record.details) |
| self.log.info(RESULT_LINE_TEMPLATE, record.test_name, record.result) |
| self.on_fail(record.test_name, record.begin_time) |
| |
| def on_fail(self, test_name, begin_time): |
| """A function that is executed upon a test case failure. |
| |
| User implementation is optional. |
| |
| Args: |
| test_name: Name of the test that triggered this function. |
| begin_time: Logline format timestamp taken when the test started. |
| """ |
| |
| def _on_pass(self, record): |
| """Proxy function to guarantee the base implementation of on_pass is |
| called. |
| |
| Args: |
| record: The records.TestResultRecord object for the passed test |
| case. |
| """ |
| self.consecutive_failures = 0 |
| msg = record.details |
| if msg: |
| self.log.info(msg) |
| self.log.info(RESULT_LINE_TEMPLATE, record.test_name, record.result) |
| self.on_pass(record.test_name, record.begin_time) |
| |
| def on_pass(self, test_name, begin_time): |
| """A function that is executed upon a test case passing. |
| |
| Implementation is optional. |
| |
| Args: |
| test_name: Name of the test that triggered this function. |
| begin_time: Logline format timestamp taken when the test started. |
| """ |
| |
| def _on_skip(self, record): |
| """Proxy function to guarantee the base implementation of on_skip is |
| called. |
| |
| Args: |
| record: The records.TestResultRecord object for the skipped test |
| case. |
| """ |
| self.log.info(RESULT_LINE_TEMPLATE, record.test_name, record.result) |
| self.log.info("Reason to skip: %s", record.details) |
| self.on_skip(record.test_name, record.begin_time) |
| |
| def on_skip(self, test_name, begin_time): |
| """A function that is executed upon a test case being skipped. |
| |
| Implementation is optional. |
| |
| Args: |
| test_name: Name of the test that triggered this function. |
| begin_time: Logline format timestamp taken when the test started. |
| """ |
| |
| def _on_exception(self, record): |
| """Proxy function to guarantee the base implementation of on_exception |
| is called. |
| |
| Args: |
| record: The records.TestResultRecord object for the failed test |
| case. |
| """ |
| self.log.exception(record.details) |
| self.on_exception(record.test_name, record.begin_time) |
| |
| def on_exception(self, test_name, begin_time): |
| """A function that is executed upon an unhandled exception from a test |
| case. |
| |
| Implementation is optional. |
| |
| Args: |
| test_name: Name of the test that triggered this function. |
| begin_time: Logline format timestamp taken when the test started. |
| """ |
| |
| def _exec_procedure_func(self, func, tr_record): |
| """Executes a procedure function like on_pass, on_fail etc. |
| |
| This function will alternate the 'Result' of the test's record if |
| exceptions happened when executing the procedure function. |
| |
| This will let signals.TestAbortAll through so abort_all works in all |
| procedure functions. |
| |
| Args: |
| func: The procedure function to be executed. |
| tr_record: The TestResultRecord object associated with the test |
| case executed. |
| """ |
| try: |
| func(tr_record) |
| except signals.TestAbortAll: |
| raise |
| except Exception as e: |
| self.log.exception("Exception happened when executing %s for %s.", |
| func.__name__, self.current_test_name) |
| tr_record.add_error(func.__name__, e) |
| |
| def exec_one_testcase(self, test_name, test_func, args, **kwargs): |
| """Executes one test case and update test results. |
| |
| Executes one test case, create a records.TestResultRecord object with |
| the execution information, and add the record to the test class's test |
| results. |
| |
| Args: |
| test_name: Name of the test. |
| test_func: The test function. |
| args: A tuple of params. |
| kwargs: Extra kwargs. |
| """ |
| class_name = self.__class__.__name__ |
| tr_record = records.TestResultRecord(test_name, class_name) |
| tr_record.test_begin() |
| self.begin_time = int(tr_record.begin_time) |
| self.log_begin_time = tr_record.log_begin_time |
| self.test_name = tr_record.test_name |
| event_bus.post(TestCaseBeginEvent(self, self.test_name)) |
| self.log.info("%s %s", TEST_CASE_TOKEN, test_name) |
| |
| # Enable test retry if specified in the ACTS config |
| retry_tests = self.user_params.get('retry_tests', []) |
| full_test_name = '%s.%s' % (class_name, self.test_name) |
| if any(name in retry_tests for name in [class_name, full_test_name]): |
| test_func = self.get_func_with_retry(test_func) |
| |
| verdict = None |
| test_signal = None |
| try: |
| try: |
| ret = self._setup_test(self.test_name) |
| asserts.assert_true(ret is not False, |
| "Setup for %s failed." % test_name) |
| if args or kwargs: |
| verdict = test_func(*args, **kwargs) |
| else: |
| verdict = test_func() |
| |
| finally: |
| try: |
| self._teardown_test(self.test_name) |
| except signals.TestAbortAll: |
| raise |
| except Exception as e: |
| self.log.error(traceback.format_exc()) |
| tr_record.add_error("teardown_test", e) |
| self._exec_procedure_func(self._on_exception, tr_record) |
| except (signals.TestFailure, AssertionError) as e: |
| test_signal = e |
| if self.user_params.get( |
| keys.Config.key_test_failure_tracebacks.value, False): |
| self.log.exception(e) |
| tr_record.test_fail(e) |
| self._exec_procedure_func(self._on_fail, tr_record) |
| except signals.TestSkip as e: |
| # Test skipped. |
| test_signal = e |
| tr_record.test_skip(e) |
| self._exec_procedure_func(self._on_skip, tr_record) |
| except (signals.TestAbortClass, signals.TestAbortAll) as e: |
| # Abort signals, pass along. |
| test_signal = e |
| tr_record.test_fail(e) |
| self._exec_procedure_func(self._on_fail, tr_record) |
| raise e |
| except signals.TestPass as e: |
| # Explicit test pass. |
| test_signal = e |
| tr_record.test_pass(e) |
| self._exec_procedure_func(self._on_pass, tr_record) |
| except error.ActsError as e: |
| test_signal = e |
| tr_record.test_error(e) |
| self.log.error( |
| 'BaseTest execute_one_test_case error: %s' % e.message) |
| except Exception as e: |
| test_signal = e |
| self.log.error(traceback.format_exc()) |
| # Exception happened during test. |
| tr_record.test_error(e) |
| self._exec_procedure_func(self._on_exception, tr_record) |
| self._exec_procedure_func(self._on_fail, tr_record) |
| else: |
| if verdict or (verdict is None): |
| # Test passed. |
| tr_record.test_pass() |
| self._exec_procedure_func(self._on_pass, tr_record) |
| return |
| tr_record.test_fail() |
| self._exec_procedure_func(self._on_fail, tr_record) |
| finally: |
| self.results.add_record(tr_record) |
| self.summary_writer.dump( |
| tr_record.to_dict(), records.TestSummaryEntryType.RECORD) |
| self.current_test_name = None |
| event_bus.post(TestCaseEndEvent(self, self.test_name, test_signal)) |
| |
| def get_func_with_retry(self, func, attempts=2): |
| """Returns a wrapped test method that re-runs after failure. Return test |
| result upon success. If attempt limit reached, collect all failure |
| messages and raise a TestFailure signal. |
| |
| Params: |
| func: The test method |
| attempts: Number of attempts to run test |
| |
| Returns: result of the test method |
| """ |
| def wrapper(*args, **kwargs): |
| error_msgs = [] |
| extras = {} |
| retry = False |
| for i in range(attempts): |
| try: |
| if retry: |
| self.teardown_test() |
| self.setup_test() |
| return func(*args, **kwargs) |
| except signals.TestFailure as e: |
| retry = True |
| msg = 'Failure on attempt %d: %s' % (i+1, e.details) |
| self.log.warning(msg) |
| error_msgs.append(msg) |
| if e.extras: |
| extras['Attempt %d' % (i+1)] = e.extras |
| raise signals.TestFailure('\n'.join(error_msgs), extras) |
| |
| return wrapper |
| |
| def run_generated_testcases(self, |
| test_func, |
| settings, |
| args=None, |
| kwargs=None, |
| tag="", |
| name_func=None, |
| format_args=False): |
| """Runs generated test cases. |
| |
| Generated test cases are not written down as functions, but as a list |
| of parameter sets. This way we reduce code repetition and improve |
| test case scalability. |
| |
| Args: |
| test_func: The common logic shared by all these generated test |
| cases. This function should take at least one argument, |
| which is a parameter set. |
| settings: A list of strings representing parameter sets. These are |
| usually json strings that get loaded in the test_func. |
| args: Iterable of additional position args to be passed to |
| test_func. |
| kwargs: Dict of additional keyword args to be passed to test_func |
| tag: Name of this group of generated test cases. Ignored if |
| name_func is provided and operates properly. |
| name_func: A function that takes a test setting and generates a |
| proper test name. The test name should be shorter than |
| utils.MAX_FILENAME_LEN. Names over the limit will be |
| truncated. |
| format_args: If True, args will be appended as the first argument |
| in the args list passed to test_func. |
| |
| Returns: |
| A list of settings that did not pass. |
| """ |
| args = args or () |
| kwargs = kwargs or {} |
| failed_settings = [] |
| |
| for setting in settings: |
| test_name = "{} {}".format(tag, setting) |
| |
| if name_func: |
| try: |
| test_name = name_func(setting, *args, **kwargs) |
| except: |
| self.log.exception(("Failed to get test name from " |
| "test_func. Fall back to default %s"), |
| test_name) |
| |
| self.results.requested.append(test_name) |
| |
| if len(test_name) > utils.MAX_FILENAME_LEN: |
| test_name = test_name[:utils.MAX_FILENAME_LEN] |
| |
| previous_success_cnt = len(self.results.passed) |
| |
| if format_args: |
| self.exec_one_testcase(test_name, test_func, |
| args + (setting,), **kwargs) |
| else: |
| self.exec_one_testcase(test_name, test_func, |
| (setting,) + args, **kwargs) |
| |
| if len(self.results.passed) - previous_success_cnt != 1: |
| failed_settings.append(setting) |
| |
| return failed_settings |
| |
| def _exec_func(self, func, *args): |
| """Executes a function with exception safeguard. |
| |
| This will let signals.TestAbortAll through so abort_all works in all |
| procedure functions. |
| |
| Args: |
| func: Function to be executed. |
| args: Arguments to be passed to the function. |
| |
| Returns: |
| Whatever the function returns, or False if unhandled exception |
| occured. |
| """ |
| try: |
| return func(*args) |
| except signals.TestAbortAll: |
| raise |
| except: |
| self.log.exception("Exception happened when executing %s in %s.", |
| func.__name__, self.TAG) |
| return False |
| |
| def _get_all_test_names(self): |
| """Finds all the function names that match the test case naming |
| convention in this class. |
| |
| Returns: |
| A list of strings, each is a test case name. |
| """ |
| test_names = [] |
| for name in dir(self): |
| if name.startswith("test_"): |
| test_names.append(name) |
| return test_names |
| |
| def _get_test_funcs(self, test_names): |
| """Obtain the actual functions of test cases based on test names. |
| |
| Args: |
| test_names: A list of strings, each string is a test case name. |
| |
| Returns: |
| A list of tuples of (string, function). String is the test case |
| name, function is the actual test case function. |
| |
| Raises: |
| Error is raised if the test name does not follow |
| naming convention "test_*". This can only be caused by user input |
| here. |
| """ |
| test_funcs = [] |
| for test_name in test_names: |
| test_funcs.append(self._get_test_func(test_name)) |
| |
| return test_funcs |
| |
| def _get_test_func(self, test_name): |
| """Obtain the actual function of test cases based on the test name. |
| |
| Args: |
| test_name: String, The name of the test. |
| |
| Returns: |
| A tuples of (string, function). String is the test case |
| name, function is the actual test case function. |
| |
| Raises: |
| Error is raised if the test name does not follow |
| naming convention "test_*". This can only be caused by user input |
| here. |
| """ |
| if not test_name.startswith("test_"): |
| raise Error(("Test case name %s does not follow naming " |
| "convention test_*, abort.") % test_name) |
| try: |
| return test_name, getattr(self, test_name) |
| except: |
| |
| def test_skip_func(*args, **kwargs): |
| raise signals.TestSkip("Test %s does not exist" % test_name) |
| |
| self.log.info("Test case %s not found in %s.", test_name, self.TAG) |
| return test_name, test_skip_func |
| |
| def _block_all_test_cases(self, tests, reason='Failed class setup'): |
| """ |
| Block all passed in test cases. |
| Args: |
| tests: The tests to block. |
| reason: Message describing the reason that the tests are blocked. |
| Default is 'Failed class setup' |
| """ |
| for test_name, test_func in tests: |
| signal = signals.TestError(reason) |
| record = records.TestResultRecord(test_name, self.TAG) |
| record.test_begin() |
| if hasattr(test_func, 'gather'): |
| signal.extras = test_func.gather() |
| record.test_error(signal) |
| self.results.add_record(record) |
| self.summary_writer.dump( |
| record.to_dict(), records.TestSummaryEntryType.RECORD) |
| self._on_skip(record) |
| |
| def run(self, test_names=None, test_case_iterations=1): |
| """Runs test cases within a test class by the order they appear in the |
| execution list. |
| |
| One of these test cases lists will be executed, shown here in priority |
| order: |
| 1. The test_names list, which is passed from cmd line. |
| 2. The self.tests list defined in test class. Invalid names are |
| ignored. |
| 3. All function that matches test case naming convention in the test |
| class. |
| |
| Args: |
| test_names: A list of string that are test case names/patterns |
| requested in cmd line. |
| |
| Returns: |
| The test results object of this class. |
| """ |
| self.register_test_class_event_subscriptions() |
| self.log.info("==========> %s <==========", self.TAG) |
| # Devise the actual test cases to run in the test class. |
| if self.tests: |
| # Specified by run list in class. |
| valid_tests = list(self.tests) |
| else: |
| # No test case specified by user, execute all in the test class |
| valid_tests = self._get_all_test_names() |
| if test_names: |
| # Match test cases with any of the user-specified patterns |
| matches = [] |
| for test_name in test_names: |
| for valid_test in valid_tests: |
| if (fnmatch.fnmatch(valid_test, test_name) |
| and valid_test not in matches): |
| matches.append(valid_test) |
| else: |
| matches = valid_tests |
| self.results.requested = matches |
| self.summary_writer.dump(self.results.requested_test_names_dict(), |
| records.TestSummaryEntryType.TEST_NAME_LIST) |
| tests = self._get_test_funcs(matches) |
| |
| # Setup for the class. |
| setup_fail = False |
| try: |
| if self._setup_class() is False: |
| self.log.error("Failed to setup %s.", self.TAG) |
| self._block_all_test_cases(tests) |
| setup_fail = True |
| except signals.TestAbortClass: |
| setup_fail = True |
| except Exception as e: |
| self.log.exception("Failed to setup %s.", self.TAG) |
| self._block_all_test_cases(tests) |
| setup_fail = True |
| if setup_fail: |
| self._exec_func(self._teardown_class) |
| self.log.info("Summary for test class %s: %s", self.TAG, |
| self.results.summary_str()) |
| return self.results |
| |
| # Run tests in order. |
| try: |
| for test_name, test_func in tests: |
| for _ in range(test_case_iterations): |
| self.exec_one_testcase(test_name, test_func, self.cli_args) |
| return self.results |
| except signals.TestAbortClass: |
| return self.results |
| except signals.TestAbortAll as e: |
| # Piggy-back test results on this exception object so we don't lose |
| # results from this test class. |
| setattr(e, "results", self.results) |
| raise e |
| finally: |
| self._exec_func(self._teardown_class) |
| self.log.info("Summary for test class %s: %s", self.TAG, |
| self.results.summary_str()) |
| |
| def _ad_take_bugreport(self, ad, test_name, begin_time): |
| for i in range(3): |
| try: |
| ad.take_bug_report(test_name, begin_time) |
| return True |
| except Exception as e: |
| ad.log.error("bugreport attempt %s error: %s", i + 1, e) |
| |
| def _ad_take_extra_logs(self, ad, test_name, begin_time): |
| result = True |
| if getattr(ad, "qxdm_log", False): |
| # Gather qxdm log modified 3 minutes earlier than test start time |
| if begin_time: |
| qxdm_begin_time = begin_time - 1000 * 60 * 3 |
| else: |
| qxdm_begin_time = None |
| try: |
| ad.get_qxdm_logs(test_name, qxdm_begin_time) |
| except Exception as e: |
| ad.log.error("Failed to get QXDM log for %s with error %s", |
| test_name, e) |
| result = False |
| |
| try: |
| ad.check_crash_report(test_name, begin_time, log_crash_report=True) |
| except Exception as e: |
| ad.log.error("Failed to check crash report for %s with error %s", |
| test_name, e) |
| result = False |
| return result |
| |
| def _skip_bug_report(self, test_name): |
| """A function to check whether we should skip creating a bug report. |
| |
| Args: |
| test_name: The test case name |
| |
| Returns: True if bug report is to be skipped. |
| """ |
| if "no_bug_report_on_fail" in self.user_params: |
| return True |
| |
| # If the current test class or test case is found in the set of |
| # problematic tests, we skip bugreport and other failure artifact |
| # creation. |
| class_name = self.__class__.__name__ |
| quiet_tests = self.user_params.get('quiet_tests', []) |
| if class_name in quiet_tests: |
| self.log.info( |
| "Skipping bug report, as directed for this test class.") |
| return True |
| full_test_name = '%s.%s' % (class_name, test_name) |
| if full_test_name in quiet_tests: |
| self.log.info( |
| "Skipping bug report, as directed for this test case.") |
| return True |
| |
| # Once we hit a certain log path size, it's not going to get smaller. |
| # We cache the result so we don't have to keep doing directory walks. |
| if self.size_limit_reached: |
| return True |
| try: |
| max_log_size = int( |
| self.user_params.get("soft_output_size_limit") or "invalid") |
| log_path = getattr(logging, "log_path", None) |
| if log_path: |
| curr_log_size = utils.get_directory_size(log_path) |
| if curr_log_size > max_log_size: |
| self.log.info( |
| "Skipping bug report, as we've reached the size limit." |
| ) |
| self.size_limit_reached = True |
| return True |
| except ValueError: |
| pass |
| return False |
| |
| def _take_bug_report(self, test_name, begin_time): |
| if self._skip_bug_report(test_name): |
| return |
| |
| executor = ThreadPoolExecutor(max_workers=10) |
| for ad in getattr(self, 'android_devices', []): |
| executor.submit(self._ad_take_bugreport, ad, test_name, begin_time) |
| executor.submit(self._ad_take_extra_logs, ad, test_name, |
| begin_time) |
| executor.shutdown() |
| |
| def _reboot_device(self, ad): |
| ad.log.info("Rebooting device.") |
| ad = ad.reboot() |
| |
| def _cleanup_logger_sessions(self): |
| for (mylogger, session) in self.logger_sessions: |
| self.log.info("Resetting a diagnostic session %s, %s", mylogger, |
| session) |
| mylogger.reset() |
| self.logger_sessions = [] |
| |
| def _pull_diag_logs(self, test_name, begin_time): |
| for (mylogger, session) in self.logger_sessions: |
| self.log.info("Pulling diagnostic session %s", mylogger) |
| mylogger.stop(session) |
| diag_path = os.path.join( |
| self.log_path, logger.epoch_to_log_line_timestamp(begin_time)) |
| utils.create_dir(diag_path) |
| mylogger.pull(session, diag_path) |
| |
| def register_test_class_event_subscriptions(self): |
| self.class_subscriptions = subscription_bundle.create_from_instance( |
| self) |
| self.class_subscriptions.register() |
| |
| def unregister_test_class_event_subscriptions(self): |
| for package in self.all_subscriptions: |
| package.unregister() |
