| #!/usr/bin/env python3.4 |
| # |
| # 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 os |
| import time |
| import traceback |
| |
| from acts import asserts |
| from acts import keys |
| from acts import logger |
| from acts import records |
| from acts import signals |
| from acts import tracelogger |
| from acts import utils |
| |
| # Macro strings for test result reporting |
| TEST_CASE_TOKEN = "[Test Case]" |
| RESULT_LINE_TEMPLATE = TEST_CASE_TOKEN + " %s %s" |
| |
| |
| class Error(Exception): |
| """Raised for exceptions that occured in BaseTestClass.""" |
| |
| |
| class BaseTestClass(object): |
| """Base class for all test classes to inherit from. |
| |
| 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. |
| 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.tests = [] |
| if not self.TAG: |
| self.TAG = self.__class__.__name__ |
| # Set all the controller objects and params. |
| 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) |
| if 'android_devices' in self.__dict__: |
| for ad in self.android_devices: |
| if ad.droid: |
| utils.set_location_service(ad, False) |
| utils.sync_device_time(ad) |
| |
| def __enter__(self): |
| return self |
| |
| def __exit__(self, *args): |
| self._exec_func(self.clean_up) |
| |
| def unpack_userparams(self, |
| req_param_names=[], |
| opt_param_names=[], |
| **kwargs): |
| """An optional function that unpacks user defined parameters into |
| individual variables. |
| |
| After unpacking, the params can be directly accessed with self.xxx. |
| |
| If a required param is not provided, an exception is raised. If an |
| optional param is not provided, a warning line will be logged. |
| |
| To provide a param, add it in the config file or pass it in as a kwarg. |
| If a param appears in both the config file and kwarg, the value in the |
| config file is used. |
| |
| User params from the config file can also be directly accessed in |
| self.user_params. |
| |
| Args: |
| req_param_names: A list of names of the required user params. |
| opt_param_names: A list of names of the optional user params. |
| **kwargs: Arguments that provide default values. |
| e.g. unpack_userparams(required_list, opt_list, arg_a="hello") |
| self.arg_a will be "hello" unless it is specified again in |
| required_list or opt_list. |
| |
| Raises: |
| Error is raised if a required user params is not provided. |
| """ |
| for k, v in kwargs.items(): |
| if k in self.user_params: |
| v = self.user_params[k] |
| setattr(self, k, v) |
| for name in req_param_names: |
| if hasattr(self, name): |
| continue |
| if name not in self.user_params: |
| raise Error(("Missing required user param '%s' in test " |
| "configuration.") % name) |
| setattr(self, name, self.user_params[name]) |
| for name in opt_param_names: |
| if hasattr(self, name): |
| continue |
| if name in self.user_params: |
| setattr(self, name, self.user_params[name]) |
| else: |
| self.log.warning(("Missing optional user param '%s' in " |
| "configuration, continue."), name) |
| |
| capablity_of_devices = utils.CapablityPerDevice |
| if "additional_energy_info_models" in self.user_params: |
| self.energy_info_models = (capablity_of_devices.energy_info_models |
| + self.additional_energy_info_models) |
| else: |
| self.energy_info_models = capablity_of_devices.energy_info_models |
| self.user_params["energy_info_models"] = self.energy_info_models |
| |
| if "additional_tdls_models" in self.user_params: |
| self.tdls_models = (capablity_of_devices.energy_info_models + |
| self.additional_tdls_models) |
| else: |
| self.tdls_models = capablity_of_devices.energy_info_models |
| self.user_params["tdls_models"] = self.tdls_models |
| |
| def _setup_class(self): |
| """Proxy function to guarantee the base implementation of setup_class |
| is called. |
| """ |
| return self.setup_class() |
| |
| def setup_class(self): |
| """Setup function that will be called before executing any 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. |
| """ |
| |
| def teardown_class(self): |
| """Teardown function that will be called after all the selected test |
| cases in the test class have been executed. |
| |
| Implementation is optional. |
| """ |
| |
| def _setup_test(self, test_name): |
| """Proxy function to guarantee the base implementation of setup_test is |
| called. |
| """ |
| self.current_test_name = test_name |
| try: |
| # Write test start token to adb log if android device is attached. |
| for ad in self.android_devices: |
| ad.droid.logV("%s BEGIN %s" % (TEST_CASE_TOKEN, test_name)) |
| except: |
| pass |
| 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. |
| """ |
| |
| def _teardown_test(self, test_name): |
| """Proxy function to guarantee the base implementation of teardown_test |
| is called. |
| """ |
| try: |
| # Write test end token to adb log if android device is attached. |
| for ad in self.android_devices: |
| ad.droid.logV("%s END %s" % (TEST_CASE_TOKEN, test_name)) |
| except: |
| pass |
| try: |
| self.teardown_test() |
| finally: |
| self.current_test_name = None |
| |
| def teardown_test(self): |
| """Teardown function that will be called every time a test case has |
| been executed. |
| |
| Implementation is optional. |
| """ |
| |
| 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. |
| """ |
| test_name = record.test_name |
| if record.details: |
| self.log.error(record.details) |
| begin_time = logger.epoch_to_log_line_timestamp(record.begin_time) |
| self.log.info(RESULT_LINE_TEMPLATE, test_name, record.result) |
| self.on_fail(test_name, 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. |
| """ |
| test_name = record.test_name |
| begin_time = logger.epoch_to_log_line_timestamp(record.begin_time) |
| msg = record.details |
| if msg: |
| self.log.info(msg) |
| self.log.info(RESULT_LINE_TEMPLATE, test_name, record.result) |
| self.on_pass(test_name, 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. |
| """ |
| test_name = record.test_name |
| begin_time = logger.epoch_to_log_line_timestamp(record.begin_time) |
| self.log.info(RESULT_LINE_TEMPLATE, test_name, record.result) |
| self.log.info("Reason to skip: %s", record.details) |
| self.on_skip(test_name, 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_blocked(self, record): |
| """Proxy function to guarantee the base implementation of on_blocked |
| is called. |
| |
| Args: |
| record: The records.TestResultRecord object for the blocked test |
| case. |
| """ |
| test_name = record.test_name |
| begin_time = logger.epoch_to_log_line_timestamp(record.begin_time) |
| self.log.info(RESULT_LINE_TEMPLATE, test_name, record.result) |
| self.log.info("Reason to block: %s", record.details) |
| self.on_blocked(test_name, begin_time) |
| |
| def on_blocked(self, test_name, begin_time): |
| """A function that is executed upon a test begin skipped. |
| |
| 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. |
| """ |
| test_name = record.test_name |
| self.log.exception(record.details) |
| begin_time = logger.epoch_to_log_line_timestamp(record.begin_time) |
| self.on_exception(test_name, 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. |
| """ |
| is_generate_trigger = False |
| tr_record = records.TestResultRecord(test_name, self.TAG) |
| tr_record.test_begin() |
| self.log.info("%s %s", TEST_CASE_TOKEN, test_name) |
| verdict = None |
| try: |
| try: |
| if hasattr(self, 'android_devices'): |
| for ad in self.android_devices: |
| if not ad.is_adb_logcat_on: |
| ad.start_adb_logcat(cont_logcat_file=True) |
| ret = self._setup_test(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(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: |
| self.log.error(e) |
| tr_record.test_fail(e) |
| self._exec_procedure_func(self._on_fail, tr_record) |
| except signals.TestSkip as e: |
| # Test skipped. |
| 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. |
| tr_record.test_fail(e) |
| raise e |
| except signals.TestPass as e: |
| # Explicit test pass. |
| tr_record.test_pass(e) |
| self._exec_procedure_func(self._on_pass, tr_record) |
| except signals.TestSilent as e: |
| # This is a trigger test for generated tests, suppress reporting. |
| is_generate_trigger = True |
| self.results.requested.remove(test_name) |
| except signals.TestBlocked as e: |
| tr_record.test_blocked(e) |
| self._exec_procedure_func(self._on_blocked, tr_record) |
| except Exception as e: |
| self.log.error(traceback.format_exc()) |
| # Exception happened during test. |
| tr_record.test_unknown(e) |
| self._exec_procedure_func(self._on_exception, tr_record) |
| self._exec_procedure_func(self._on_fail, tr_record) |
| else: |
| # Keep supporting return False for now. |
| # TODO(angli): Deprecate return False support. |
| if verdict or (verdict is None): |
| # Test passed. |
| tr_record.test_pass() |
| self._exec_procedure_func(self._on_pass, tr_record) |
| return |
| # Test failed because it didn't return True. |
| # This should be removed eventually. |
| tr_record.test_fail() |
| self._exec_procedure_func(self._on_fail, tr_record) |
| finally: |
| if not is_generate_trigger: |
| self.results.add_record(tr_record) |
| |
| 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): |
| """ |
| Block all passed in test cases. |
| Args: |
| tests: The tests to block. |
| """ |
| for test_name, test_func in tests: |
| signal = signals.TestBlocked("Failed class setup") |
| record = records.TestResultRecord(test_name, self.TAG) |
| record.test_begin() |
| if hasattr(test_func, 'gather'): |
| signal.extras = test_func.gather() |
| record.test_blocked(signal) |
| self.results.add_record(record) |
| self._on_blocked(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. Invalid names |
| are guarded by cmd line arg parsing. |
| 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 requested in |
| cmd line. |
| |
| Returns: |
| The test results object of this class. |
| """ |
| self.log.info("==========> %s <==========", self.TAG) |
| # Devise the actual test cases to run in the test class. |
| if not test_names: |
| if self.tests: |
| # Specified by run list in class. |
| test_names = list(self.tests) |
| else: |
| # No test case specified by user, execute all in the test class |
| test_names = self._get_all_test_names() |
| self.results.requested = test_names |
| tests = self._get_test_funcs(test_names) |
| # A TestResultRecord used for when setup_class fails. |
| # Setup for the class. |
| try: |
| if self._setup_class() is False: |
| self.log.error("Failed to setup %s.", self.TAG) |
| self._block_all_test_cases(tests) |
| return self.results |
| except Exception as e: |
| self.log.exception("Failed to setup %s.", self.TAG) |
| self._exec_func(self.teardown_class) |
| self._block_all_test_cases(tests) |
| 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 clean_up(self): |
| """A function that is executed upon completion of all tests cases |
| selected in the test class. |
| |
| This function should clean up objects initialized in the constructor by |
| user. |
| """ |
| |
| def _take_bug_report(self, test_name, begin_time): |
| if "no_bug_report_on_fail" in self.user_params: |
| return |
| |
| # magical sleep to ensure the runtime restart or reboot begins |
| time.sleep(1) |
| for ad in self.android_devices: |
| try: |
| ad.adb.wait_for_device() |
| ad.take_bug_report(test_name, begin_time) |
| bugreport_path = os.path.join(ad.log_path, test_name) |
| utils.create_dir(bugreport_path) |
| ad.check_crash_report(True, test_name) |
| if getattr(ad, "qxdm_always_on", False): |
| ad.log.info("Pull QXDM Logs") |
| ad.pull_files(["/data/vendor/radio/diag_logs/logs/"], |
| bugreport_path) |
| except Exception as e: |
| ad.log.error( |
| "Failed to take a bug report for %s with error %s", |
| test_name, e) |
| |
| def _reboot_device(self, ad): |
| ad.log.info("Rebooting device.") |
| ad = ad.reboot() |
| |
| def _cleanup_logger_sessions(self): |
| for (logger, session) in self.logger_sessions: |
| self.log.info("Resetting a diagnostic session %s, %s", logger, |
| session) |
| logger.reset() |
| self.logger_sessions = [] |
| |
| def _pull_diag_logs(self, test_name, begin_time): |
| for (logger, session) in self.logger_sessions: |
| self.log.info("Pulling diagnostic session %s", logger) |
| logger.stop(session) |
| diag_path = os.path.join(self.log_path, begin_time) |
| utils.create_dir(diag_path) |
| logger.pull(session, diag_path) |
