Merge "Adding LTE hotspot test suite for baselines"
diff --git a/acts/framework/acts/base_test.py b/acts/framework/acts/base_test.py
index d8a08e1..8df1a23 100755
--- a/acts/framework/acts/base_test.py
+++ b/acts/framework/acts/base_test.py
@@ -13,7 +13,7 @@
# 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 copy
+
import fnmatch
import importlib
import logging
@@ -38,6 +38,7 @@
from acts.event.event import TestClassEndEvent
from acts.event.subscription_bundle import SubscriptionBundle
+from mobly import controller_manager
# Macro strings for test result reporting
TEST_CASE_TOKEN = "[Test Case]"
@@ -59,7 +60,7 @@
event.test_case_name))
except error.ActsError as e:
- test_instance.results.errors.append(e)
+ test_instance.results.error.append(e)
test_instance.log.error('BaseTest setup_test error: %s' % e.message)
except Exception as e:
@@ -80,7 +81,7 @@
event.test_case_name))
except error.ActsError as e:
- test_instance.results.errors.append(e)
+ test_instance.results.error.append(e)
test_instance.log.error('BaseTest teardown_test error: %s' % e.message)
except Exception as e:
@@ -123,8 +124,6 @@
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.
- controller_registry: A dictionary that holds the controller objects used
- in a test run.
"""
TAG = None
@@ -150,7 +149,11 @@
'consecutive_failure_limit', -1)
self.size_limit_reached = False
- self.controller_registry = {}
+ # 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():
@@ -238,32 +241,6 @@
return builtin_controllers
@staticmethod
- def verify_controller_module(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 ACTS
- controller interface, or one of the required members is null.
- """
- required_attributes = ("create", "destroy",
- "ACTS_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__))
-
- @staticmethod
def get_module_reference_name(a_module):
"""Returns the module's reference name.
@@ -285,7 +262,8 @@
controller_module,
required=True,
builtin=False):
- """Registers an ACTS controller module for a test run.
+ """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
@@ -353,41 +331,18 @@
the controller module has already been registered or any other error
occurred in the registration process.
"""
- BaseTestClass.verify_controller_module(controller_module)
module_ref_name = self.get_module_reference_name(controller_module)
- if controller_module in self.controller_registry:
- raise signals.ControllerError(
- "Controller module %s has already been registered. It can not "
- "be registered again." % module_ref_name)
- # Create controller objects.
+ # Substitute Mobly controller's module config name with the ACTS one
module_config_name = controller_module.ACTS_CONTROLLER_CONFIG_NAME
- if module_config_name not in self.testbed_configs:
- if required:
- raise signals.ControllerError(
- "No corresponding config found for %s" %
- module_config_name)
- else:
- self.log.warning(
- "No corresponding config found for optional controller %s",
- module_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
- 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)
- controllers = controller_module.create(controller_config)
- except:
- self.log.exception(
- "Failed to initialize objects for controller %s, abort!",
- module_config_name)
- raise
- if not isinstance(controllers, list):
- raise signals.ControllerError(
- "Controller module %s did not return a list of objects, abort."
- % module_ref_name)
- self.controller_registry[controller_module] = controllers
+
# Collect controller information and write to test result.
# Implementation of "get_info" is optional for a controller module.
if hasattr(controller_module, "get_info"):
@@ -402,31 +357,35 @@
if builtin:
setattr(self, module_ref_name, controllers)
- self.log.debug("Found %d objects for controller %s", len(controllers),
- module_config_name)
return controllers
def unregister_controllers(self):
- """Destroy controller objects and clear internal registry.
+ """Destroy controller objects and clear internal registry. Invokes
+ Mobly's controller manager's unregister_controllers.
- This will be called at the end of each TestRunner.run call.
+ This will be called upon test class teardown.
"""
- for controller_module, controllers in self.controller_registry.items():
- name = self.get_module_reference_name(controller_module)
+ 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(
- controllers)
+ 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)
- try:
- self.log.debug('Destroying %s.', name)
- controller_module.destroy(controllers)
- except:
- self.log.exception("Exception occurred destroying %s.", name)
- self.controller_registry = {}
+ self._controller_manager.unregister_controllers()
+
+ def _record_controller_info(self):
+ """Collect controller information and write to summary file."""
+ for record in self._controller_manager.get_controller_info_records():
+ self.summary_writer.dump(
+ record.to_dict(), records.TestSummaryEntryType.CONTROLLER_INFO)
def _setup_class(self):
"""Proxy function to guarantee the base implementation of setup_class
@@ -451,6 +410,7 @@
is called.
"""
self.teardown_class()
+ self._record_controller_info()
self.unregister_controllers()
event_bus.post(TestClassEndEvent(self, self.results))
@@ -467,9 +427,9 @@
"""
self.current_test_name = test_name
- # Block the test if the consecutive test case failure limit is reached.
+ # Skip the test if the consecutive test case failure limit is reached.
if self.consecutive_failures == self.consecutive_failure_limit:
- raise signals.TestBlocked('Consecutive test failure')
+ raise signals.TestError('Consecutive test failure')
return self.setup_test()
@@ -570,26 +530,6 @@
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.
- """
- self.log.info(RESULT_LINE_TEMPLATE, record.test_name, record.result)
- self.log.info("Reason to block: %s", record.details)
- self.on_blocked(record.test_name, record.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.
@@ -691,11 +631,6 @@
self.log.exception(e)
tr_record.test_fail(e)
self._exec_procedure_func(self._on_fail, tr_record)
- except signals.TestBlocked as e:
- # Test blocked.
- test_signal = e
- tr_record.test_blocked(e)
- self._exec_procedure_func(self._on_blocked, tr_record)
except signals.TestSkip as e:
# Test skipped.
test_signal = e
@@ -714,14 +649,14 @@
self._exec_procedure_func(self._on_pass, tr_record)
except error.ActsError as e:
test_signal = e
- self.results.errors.append(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_unknown(e)
+ tr_record.test_error(e)
self._exec_procedure_func(self._on_exception, tr_record)
self._exec_procedure_func(self._on_fail, tr_record)
else:
@@ -734,6 +669,8 @@
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))
@@ -932,14 +869,16 @@
Default is 'Failed class setup'
"""
for test_name, test_func in tests:
- signal = signals.TestBlocked(reason)
+ 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_blocked(signal)
+ record.test_error(signal)
self.results.add_record(record)
- self._on_blocked(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
@@ -980,8 +919,10 @@
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)
- # A TestResultRecord used for when setup_class fails.
+
# Setup for the class.
setup_fail = False
try:
diff --git a/acts/framework/acts/controllers/android_device.py b/acts/framework/acts/controllers/android_device.py
index 96798c7..a774de4 100755
--- a/acts/framework/acts/controllers/android_device.py
+++ b/acts/framework/acts/controllers/android_device.py
@@ -709,8 +709,7 @@
end_time: Epoch time of the ending of the time period, default None
dest_path: Destination path of the excerpt file.
"""
- log_begin_time = acts_logger.normalize_log_line_timestamp(
- acts_logger.epoch_to_log_line_timestamp(begin_time))
+ log_begin_time = acts_logger.epoch_to_log_line_timestamp(begin_time)
if end_time is None:
log_end_time = acts_logger.get_log_line_timestamp()
else:
@@ -723,7 +722,9 @@
return
adb_excerpt_dir = os.path.join(self.log_path, dest_path)
utils.create_dir(adb_excerpt_dir)
- out_name = '%s,%s.txt' % (log_begin_time, self.serial)
+ out_name = '%s,%s.txt' % (
+ acts_logger.normalize_log_line_timestamp(log_begin_time),
+ self.serial)
tag_len = utils.MAX_FILENAME_LEN - len(out_name)
out_name = '%s,%s' % (tag[:tag_len], out_name)
adb_excerpt_path = os.path.join(adb_excerpt_dir, out_name)
diff --git a/acts/framework/acts/keys.py b/acts/framework/acts/keys.py
index 6ae23c7..2c3c6ff 100644
--- a/acts/framework/acts/keys.py
+++ b/acts/framework/acts/keys.py
@@ -60,6 +60,7 @@
ikey_testbed_name = 'testbed_name'
ikey_logger = 'log'
ikey_logpath = 'log_path'
+ ikey_summary_writer = 'summary_writer'
ikey_cli_args = 'cli_args'
# module name of controllers packaged in ACTS.
m_key_monsoon = 'monsoon'
diff --git a/acts/framework/acts/libs/yaml_writer.py b/acts/framework/acts/libs/yaml_writer.py
new file mode 100644
index 0000000..f1f0831
--- /dev/null
+++ b/acts/framework/acts/libs/yaml_writer.py
@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+#
+# Copyright 2019 - 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 collections
+import yaml
+
+# Allow yaml to dump OrderedDict
+yaml.add_representer(collections.OrderedDict,
+ lambda dumper, data: dumper.represent_dict(data),
+ Dumper=yaml.SafeDumper)
+
+_DUMP_KWARGS = dict(explicit_start=True, allow_unicode=True, indent=4)
+if yaml.__version__ >= '5.1':
+ _DUMP_KWARGS.update(sort_keys=False)
+
+
+def safe_dump(content, file):
+ """Calls yaml.safe_dump to write content to the file, with additional
+ parameters from _DUMP_KWARGS."""
+ yaml.safe_dump(content, file, **_DUMP_KWARGS)
diff --git a/acts/framework/acts/records.py b/acts/framework/acts/records.py
index a3795b8..575ee60 100644
--- a/acts/framework/acts/records.py
+++ b/acts/framework/acts/records.py
@@ -18,40 +18,62 @@
import collections
import copy
+import io
import json
import logging
from acts import logger
-from acts import utils
+from acts.libs import yaml_writer
+
+from mobly.records import ExceptionRecord
+from mobly.records import OUTPUT_FILE_SUMMARY
+from mobly.records import TestResultEnums as MoblyTestResultEnums
+from mobly.records import TestResultRecord as MoblyTestResultRecord
+from mobly.records import TestResult as MoblyTestResult
+from mobly.records import TestSummaryEntryType
+from mobly.records import TestSummaryWriter as MoblyTestSummaryWriter
-class TestResultEnums(object):
- """Enums used for TestResultRecord class.
+class TestSummaryWriter(MoblyTestSummaryWriter):
+ """Writes test results to a summary file in real time. Inherits from Mobly's
+ TestSummaryWriter.
+ """
+
+ def dump(self, content, entry_type):
+ """Update Mobly's implementation of dump to work on OrderedDict.
+
+ See MoblyTestSummaryWriter.dump for documentation.
+ """
+ new_content = collections.OrderedDict(copy.deepcopy(content))
+ new_content['Type'] = entry_type.value
+ new_content.move_to_end('Type', last=False)
+ # Both user code and Mobly code can trigger this dump, hence the lock.
+ with self._lock:
+ # For Python3, setting the encoding on yaml.safe_dump does not work
+ # because Python3 file descriptors set an encoding by default, which
+ # PyYAML uses instead of the encoding on yaml.safe_dump. So, the
+ # encoding has to be set on the open call instead.
+ with io.open(self._path, 'a', encoding='utf-8') as f:
+ # Use safe_dump here to avoid language-specific tags in final
+ # output.
+ yaml_writer.safe_dump(new_content, f)
+
+
+class TestResultEnums(MoblyTestResultEnums):
+ """Enums used for TestResultRecord class. Inherits from Mobly's
+ TestResultEnums.
Includes the tokens to mark test result with, and the string names for each
field in TestResultRecord.
"""
- RECORD_NAME = "Test Name"
- RECORD_CLASS = "Test Class"
- RECORD_BEGIN_TIME = "Begin Time"
- RECORD_END_TIME = "End Time"
RECORD_LOG_BEGIN_TIME = "Log Begin Time"
RECORD_LOG_END_TIME = "Log End Time"
- RECORD_RESULT = "Result"
- RECORD_UID = "UID"
- RECORD_EXTRAS = "Extras"
- RECORD_ADDITIONAL_ERRORS = "Extra Errors"
- RECORD_DETAILS = "Details"
- TEST_RESULT_PASS = "PASS"
- TEST_RESULT_FAIL = "FAIL"
- TEST_RESULT_SKIP = "SKIP"
- TEST_RESULT_BLOCKED = "BLOCKED"
- TEST_RESULT_UNKNOWN = "UNKNOWN"
-class TestResultRecord(object):
- """A record that holds the information of a test case execution.
+class TestResultRecord(MoblyTestResultRecord):
+ """A record that holds the information of a test case execution. This class
+ inherits from Mobly's TestResultRecord class.
Attributes:
test_name: A string representing the name of the test case.
@@ -64,24 +86,16 @@
"""
def __init__(self, t_name, t_class=None):
- self.test_name = t_name
- self.test_class = t_class
- self.begin_time = None
- self.end_time = None
+ super().__init__(t_name, t_class)
self.log_begin_time = None
self.log_end_time = None
- self.uid = None
- self.result = None
- self.extras = None
- self.details = None
- self.additional_errors = {}
def test_begin(self):
"""Call this when the test case it records begins execution.
Sets the begin_time of this record.
"""
- self.begin_time = utils.get_current_epoch_time()
+ super().test_begin()
self.log_begin_time = logger.epoch_to_log_line_timestamp(
self.begin_time)
@@ -94,86 +108,10 @@
be any exception instance or of any subclass of
acts.signals.TestSignal.
"""
- self.end_time = utils.get_current_epoch_time()
- self.log_end_time = logger.epoch_to_log_line_timestamp(self.end_time)
- self.result = result
- if self.additional_errors:
- self.result = TestResultEnums.TEST_RESULT_UNKNOWN
- if hasattr(e, 'details'):
- self.details = e.details
- elif e:
- self.details = str(e)
- if hasattr(e, 'extras'):
- self.extras = e.extras
-
- def test_pass(self, e=None):
- """To mark the test as passed in this record.
-
- Args:
- e: An instance of acts.signals.TestPass.
- """
- self._test_end(TestResultEnums.TEST_RESULT_PASS, e)
-
- def test_fail(self, e=None):
- """To mark the test as failed in this record.
-
- Only test_fail does instance check because we want "assert xxx" to also
- fail the test same way assert_true does.
-
- Args:
- e: An exception object. It can be an instance of AssertionError or
- acts.base_test.TestFailure.
- """
- self._test_end(TestResultEnums.TEST_RESULT_FAIL, e)
-
- def test_skip(self, e=None):
- """To mark the test as skipped in this record.
-
- Args:
- e: An instance of acts.signals.TestSkip.
- """
- self._test_end(TestResultEnums.TEST_RESULT_SKIP, e)
-
- def test_blocked(self, e=None):
- """To mark the test as blocked in this record.
-
- Args:
- e: An instance of acts.signals.Test
- """
- self._test_end(TestResultEnums.TEST_RESULT_BLOCKED, e)
-
- def test_unknown(self, e=None):
- """To mark the test as unknown in this record.
-
- Args:
- e: An exception object.
- """
- self._test_end(TestResultEnums.TEST_RESULT_UNKNOWN, e)
-
- def add_error(self, tag, e):
- """Add extra error happened during a test mark the test result as
- UNKNOWN.
-
- If an error is added the test record, the record's result is equivalent
- to the case where an uncaught exception happened.
-
- Args:
- tag: A string describing where this error came from, e.g. 'on_pass'.
- e: An exception object.
- """
- self.result = TestResultEnums.TEST_RESULT_UNKNOWN
- self.additional_errors[tag] = str(e)
-
- def __str__(self):
- d = self.to_dict()
- l = ["%s = %s" % (k, v) for k, v in d.items()]
- s = ', '.join(l)
- return s
-
- def __repr__(self):
- """This returns a short string representation of the test record."""
- t = utils.epoch_to_human_time(self.begin_time)
- return "%s %s %s" % (t, self.test_name, self.result)
+ super()._test_end(result, e)
+ if self.end_time:
+ self.log_end_time = logger.epoch_to_log_line_timestamp(
+ self.end_time)
def to_dict(self):
"""Gets a dictionary representing the content of this class.
@@ -192,7 +130,11 @@
d[TestResultEnums.RECORD_UID] = self.uid
d[TestResultEnums.RECORD_EXTRAS] = self.extras
d[TestResultEnums.RECORD_DETAILS] = self.details
- d[TestResultEnums.RECORD_ADDITIONAL_ERRORS] = self.additional_errors
+ d[TestResultEnums.RECORD_EXTRA_ERRORS] = {
+ key: value.to_dict()
+ for (key, value) in self.extra_errors.items()
+ }
+ d[TestResultEnums.RECORD_STACKTRACE] = self.stacktrace
return d
def json_str(self):
@@ -212,8 +154,9 @@
return json.dumps(self.to_dict())
-class TestResult(object):
- """A class that contains metrics of a test run.
+class TestResult(MoblyTestResult):
+ """A class that contains metrics of a test run. This class inherits from
+ Mobly's TestResult class.
This class is essentially a container of TestResultRecord objects.
@@ -224,21 +167,12 @@
self.executed: A list of records for tests that were actually executed.
self.passed: A list of records for tests passed.
self.skipped: A list of records for tests skipped.
- self.unknown: A list of records for tests with unknown result token.
"""
def __init__(self):
- self.requested = []
- self.failed = []
- self.executed = []
- self.passed = []
- self.skipped = []
- self.blocked = []
- self.unknown = []
+ super().__init__()
self.controller_info = {}
- self.post_run_data = {}
self.extras = {}
- self.errors = []
def __add__(self, r):
"""Overrides '+' operator for TestResult class.
@@ -286,36 +220,6 @@
info = str(info)
self.extras[name] = info
- def add_record(self, record):
- """Adds a test record to test result.
-
- A record is considered executed once it's added to the test result.
-
- Args:
- record: A test record object to add.
- """
- if record.result == TestResultEnums.TEST_RESULT_FAIL:
- self.executed.append(record)
- self.failed.append(record)
- elif record.result == TestResultEnums.TEST_RESULT_SKIP:
- self.skipped.append(record)
- elif record.result == TestResultEnums.TEST_RESULT_PASS:
- self.executed.append(record)
- self.passed.append(record)
- elif record.result == TestResultEnums.TEST_RESULT_BLOCKED:
- self.blocked.append(record)
- else:
- self.executed.append(record)
- self.unknown.append(record)
-
- @property
- def is_all_pass(self):
- """True if no tests failed or threw errors, False otherwise."""
- num_of_failures = (len(self.failed) +
- len(self.unknown) +
- len(self.blocked))
- return num_of_failures == 0
-
def json_str(self):
"""Converts this test result to a string in json format.
@@ -337,7 +241,7 @@
d["Results"] = [record.to_dict() for record in self.executed]
d["Summary"] = self.summary_dict()
d["Extras"] = self.extras
- d["Errors"] = self.errors_list()
+ d["Error"] = self.errors_list()
json_str = json.dumps(d, indent=4)
return json_str
@@ -357,29 +261,12 @@
msg = ", ".join(l)
return msg
- def summary_dict(self):
- """Gets a dictionary that summarizes the stats of this test result.
-
- The summary provides the counts of how many test cases fall into each
- category, like "Passed", "Failed" etc.
-
- Returns:
- A dictionary with the stats of this test result.
- """
- d = collections.OrderedDict()
- d["ControllerInfo"] = self.controller_info
- d["Requested"] = len(self.requested)
- d["Executed"] = len(self.executed)
- d["Passed"] = len(self.passed)
- d["Failed"] = len(self.failed)
- d["Skipped"] = len(self.skipped)
- d["Blocked"] = len(self.blocked)
- d["Unknown"] = len(self.unknown)
- d["Errors"] = len(self.errors)
- return d
-
def errors_list(self):
l = list()
- for e in self.errors:
- l.append({"Error Code": e.error_code, "Message": e.message})
+ for record in self.error:
+ record_dict = record.to_dict()
+ l.append({k: record_dict[k] for k in [
+ TestResultEnums.RECORD_NAME,
+ TestResultEnums.RECORD_DETAILS,
+ TestResultEnums.RECORD_EXTRA_ERRORS]})
return l
diff --git a/acts/framework/acts/signals.py b/acts/framework/acts/signals.py
index e116984..67c22cd 100644
--- a/acts/framework/acts/signals.py
+++ b/acts/framework/acts/signals.py
@@ -17,7 +17,3 @@
"""
from mobly.signals import *
-
-
-class TestBlocked(TestSkip):
- """Raised when a test has been blocked from running."""
diff --git a/acts/framework/acts/test_runner.py b/acts/framework/acts/test_runner.py
index 1f127c8..1adfc91 100644
--- a/acts/framework/acts/test_runner.py
+++ b/acts/framework/acts/test_runner.py
@@ -106,6 +106,8 @@
self.log_path: A string representing the path of the dir under which
all logs from this test run should be written.
self.log: The logger object used throughout this test run.
+ self.summary_writer: The TestSummaryWriter object used to stream test
+ results to a file.
self.test_classes: A dictionary where we can look up the test classes
by name to instantiate. Supports unix shell style
wildcards.
@@ -131,6 +133,8 @@
self.log_path = os.path.abspath(l_path)
logger.setup_test_logger(self.log_path, self.testbed_name)
self.log = logging.getLogger()
+ self.summary_writer = records.TestSummaryWriter(
+ os.path.join(self.log_path, records.OUTPUT_FILE_SUMMARY))
if self.test_configs.get(keys.Config.key_random.value):
test_case_iterations = self.test_configs.get(
keys.Config.key_test_case_iterations.value, 10)
@@ -214,6 +218,8 @@
# Unpack other params.
self.test_run_info[keys.Config.ikey_logpath.value] = self.log_path
self.test_run_info[keys.Config.ikey_logger.value] = self.log
+ self.test_run_info[
+ keys.Config.ikey_summary_writer.value] = self.summary_writer
cli_args = test_configs.get(keys.Config.ikey_cli_args.value)
self.test_run_info[keys.Config.ikey_cli_args.value] = cli_args
user_param_pairs = []
@@ -290,7 +296,7 @@
cls_result = test_cls_instance.run(test_cases,
test_case_iterations)
self.results += cls_result
- self._write_results_json_str()
+ self._write_results_to_file()
except signals.TestAbortAll as e:
self.results += e.results
raise e
@@ -350,19 +356,20 @@
if self.running:
msg = "\nSummary for test run %s: %s\n" % (
self.id, self.results.summary_str())
- self._write_results_json_str()
+ self._write_results_to_file()
self.log.info(msg.strip())
logger.kill_test_logger(self.log)
self.running = False
- def _write_results_json_str(self):
- """Writes out a json file with the test result info for easy parsing.
-
- TODO(angli): This should be replaced by standard log record mechanism.
- """
+ def _write_results_to_file(self):
+ """Writes test results to file(s) in a serializable format."""
+ # Old JSON format
path = os.path.join(self.log_path, "test_run_summary.json")
with open(path, 'w') as f:
f.write(self.results.json_str())
+ # New YAML format
+ self.summary_writer.dump(
+ self.results.summary_dict(), records.TestSummaryEntryType.SUMMARY)
def dump_config(self):
"""Writes the test config to a JSON file under self.log_path"""
diff --git a/acts/framework/acts/test_utils/tel/TelephonyBaseTest.py b/acts/framework/acts/test_utils/tel/TelephonyBaseTest.py
index 0746d8b..5ee4721 100644
--- a/acts/framework/acts/test_utils/tel/TelephonyBaseTest.py
+++ b/acts/framework/acts/test_utils/tel/TelephonyBaseTest.py
@@ -443,9 +443,6 @@
def on_fail(self, test_name, begin_time):
self._take_bug_report(test_name, begin_time)
- def on_blocked(self, test_name, begin_time):
- self.on_fail(test_name, begin_time)
-
def _ad_take_extra_logs(self, ad, test_name, begin_time):
ad.adb.wait_for_device()
result = True
@@ -509,7 +506,7 @@
def _block_all_test_cases(self, tests, reason='Failed class setup'):
"""Over-write _block_all_test_cases in BaseTestClass."""
for (i, (test_name, test_func)) in enumerate(tests):
- signal = signals.TestBlocked(reason)
+ signal = signals.TestFailure(reason)
record = records.TestResultRecord(test_name, self.TAG)
record.test_begin()
# mark all test cases as FAIL
diff --git a/acts/framework/acts/test_utils/tel/tel_defines.py b/acts/framework/acts/test_utils/tel/tel_defines.py
index 222d77b..3aefe98 100644
--- a/acts/framework/acts/test_utils/tel/tel_defines.py
+++ b/acts/framework/acts/test_utils/tel/tel_defines.py
@@ -143,6 +143,9 @@
# has sufficient time to reconfigure based on new network
WAIT_TIME_BETWEEN_REG_AND_CALL = 5
+# Wait time for data pdn to be up on CBRS
+WAIT_TIME_FOR_CBRS_DATA_SWITCH = 30
+
# Time to wait for 1xrtt voice attach check
# After DUT voice network type report 1xrtt (from unknown), it need to wait for
# several seconds before the DUT can receive incoming call.
diff --git a/acts/framework/setup.py b/acts/framework/setup.py
index 936f0a7..0619706 100755
--- a/acts/framework/setup.py
+++ b/acts/framework/setup.py
@@ -30,6 +30,7 @@
'mock<=1.0.1',
'numpy',
'pyserial',
+ 'pyyaml>=5.1',
'shellescape>=3.4.1',
'protobuf',
'requests',
diff --git a/acts/framework/tests/acts_base_class_test.py b/acts/framework/tests/acts_base_class_test.py
index 37b5c53..b402a47 100755
--- a/acts/framework/tests/acts_base_class_test.py
+++ b/acts/framework/tests/acts_base_class_test.py
@@ -52,6 +52,7 @@
'reporter': mock.MagicMock(),
'log': mock.MagicMock(),
'log_path': self.tmp_dir,
+ 'summary_writer': mock.MagicMock(),
'cli_args': None,
'user_params': {
'some_param': 'hahaha'
@@ -80,7 +81,7 @@
class MockBaseTest(base_test.BaseTestClass):
def __init__(self, controllers):
super(MockBaseTest, self).__init__(controllers)
- self.tests = ('test_something', )
+ self.tests = ('test_something',)
def test_something(self):
pass
@@ -98,7 +99,7 @@
class MockBaseTest(base_test.BaseTestClass):
def __init__(self, controllers):
super(MockBaseTest, self).__init__(controllers)
- self.tests = ('not_a_test_something', )
+ self.tests = ('not_a_test_something',)
def not_a_test_something(self):
pass
@@ -166,7 +167,7 @@
class MockBaseTest(base_test.BaseTestClass):
def __init__(self, controllers):
super(MockBaseTest, self).__init__(controllers)
- self.tests = ('test_something', )
+ self.tests = ('test_something',)
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
@@ -184,16 +185,16 @@
# This should not execute because setup_class failed.
never_call()
- def on_blocked(self, test_name, begin_time):
+ def on_skip(self, test_name, begin_time):
call_check('haha')
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
- actual_record = bt_cls.results.blocked[0]
+ actual_record = bt_cls.results.error[0]
self.assertEqual(actual_record.test_name, 'test_something')
expected_summary = {
- 'Blocked': 1, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 0,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 0
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
call_check.assert_called_once_with('haha')
@@ -209,13 +210,13 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run(test_names=['test_something'])
- actual_record = bt_cls.results.unknown[0]
+ actual_record = bt_cls.results.error[0]
self.assertEqual(actual_record.test_name, self.mock_test_name)
self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertIsNone(actual_record.extras)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -235,8 +236,8 @@
self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertIsNone(actual_record.extras)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 1, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 0
+ 'Error': 0, 'Executed': 1,
+ 'Failed': 1, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -257,8 +258,8 @@
self.assertEqual(actual_record.details, expected_msg)
self.assertIsNone(actual_record.extras)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 1, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 0
+ 'Error': 0, 'Executed': 1,
+ 'Failed': 1, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -273,8 +274,8 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run(test_names=['test_something'])
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 1, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -288,13 +289,13 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
- actual_record = bt_cls.results.unknown[0]
+ actual_record = bt_cls.results.error[0]
self.assertEqual(actual_record.test_name, self.mock_test_name)
- self.assertIsNone(actual_record.details)
+ self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertIsNone(actual_record.extras)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -308,15 +309,13 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
- actual_record = bt_cls.results.unknown[0]
+ actual_record = bt_cls.results.error[0]
self.assertEqual(actual_record.test_name, self.mock_test_name)
- self.assertIsNone(actual_record.details)
+ self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertIsNone(actual_record.extras)
- expected_extra_error = {'teardown_test': MSG_EXPECTED_EXCEPTION}
- self.assertEqual(actual_record.additional_errors, expected_extra_error)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -338,8 +337,8 @@
self.assertIsNone(actual_record.details)
self.assertIsNone(actual_record.extras)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 1, 'Requested': 1, 'Skipped': 0, 'Unknown': 0
+ 'Error': 0, 'Executed': 1,
+ 'Failed': 0, 'Passed': 1, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -358,14 +357,14 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
- actual_record = bt_cls.results.unknown[0]
+ actual_record = bt_cls.results.error[0]
my_mock.assert_called_once_with('teardown_test')
self.assertEqual(actual_record.test_name, self.mock_test_name)
self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertIsNone(actual_record.extras)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -381,14 +380,14 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
- actual_record = bt_cls.results.unknown[0]
+ actual_record = bt_cls.results.error[0]
my_mock.assert_called_once_with('teardown_test')
self.assertEqual(actual_record.test_name, self.mock_test_name)
self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertIsNone(actual_record.extras)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -408,13 +407,13 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
my_mock.assert_called_once_with('on_exception')
- actual_record = bt_cls.results.unknown[0]
+ actual_record = bt_cls.results.error[0]
self.assertEqual(actual_record.test_name, self.mock_test_name)
- self.assertIsNone(actual_record.details)
+ self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertIsNone(actual_record.extras)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -436,8 +435,8 @@
self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertIsNone(actual_record.extras)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 1, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 0
+ 'Error': 0, 'Executed': 1,
+ 'Failed': 1, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -457,13 +456,13 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
my_mock.assert_called_once_with('on_fail')
- actual_record = bt_cls.results.unknown[0]
+ actual_record = bt_cls.results.error[0]
self.assertEqual(actual_record.test_name, self.mock_test_name)
self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertIsNone(actual_record.extras)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -489,29 +488,30 @@
'Setup for test_something failed.')
self.assertIsNone(actual_record.extras)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 1, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 0
+ 'Error': 0, 'Executed': 1,
+ 'Failed': 1, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
def test_failure_to_call_procedure_function_is_recorded(self):
class MockBaseTest(base_test.BaseTestClass):
- def on_fail(self):
+ # Wrong method signature; will raise exception
+ def on_pass(self):
pass
def test_something(self):
- asserts.assert_true(False, MSG_EXPECTED_EXCEPTION)
+ asserts.explicit_pass(MSG_EXPECTED_EXCEPTION)
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
- actual_record = bt_cls.results.unknown[0]
- self.assertIn('_on_fail', actual_record.additional_errors)
+ actual_record = bt_cls.results.error[0]
+ self.assertIn('_on_pass', actual_record.extra_errors)
self.assertEqual(actual_record.test_name, self.mock_test_name)
self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertIsNone(actual_record.extras)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -527,15 +527,13 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
- actual_record = bt_cls.results.unknown[0]
- expected_extra_error = {'_on_pass': expected_msg}
- self.assertEqual(actual_record.additional_errors, expected_extra_error)
+ actual_record = bt_cls.results.error[0]
self.assertEqual(actual_record.test_name, self.mock_test_name)
self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertIsNone(actual_record.extras)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -549,15 +547,15 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
- actual_record = bt_cls.results.unknown[0]
+ actual_record = bt_cls.results.error[0]
self.assertEqual(actual_record.test_name, self.mock_test_name)
self.assertEqual(actual_record.details, 'Test Body Exception.')
self.assertIsNone(actual_record.extras)
- self.assertEqual(actual_record.additional_errors['teardown_test'],
- 'Details=This is an expected exception., Extras=None')
+ self.assertEqual(actual_record.extra_errors['teardown_test'].details,
+ 'This is an expected exception.')
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -574,15 +572,15 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
- actual_record = bt_cls.results.unknown[0]
+ actual_record = bt_cls.results.error[0]
self.assertEqual(actual_record.test_name, self.mock_test_name)
self.assertEqual(actual_record.details, 'Test Passed!')
self.assertIsNone(actual_record.extras)
- self.assertEqual(actual_record.additional_errors['teardown_test'],
- 'Details=This is an expected exception., Extras=None')
+ self.assertEqual(actual_record.extra_errors['teardown_test'].details,
+ 'This is an expected exception.')
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -597,16 +595,13 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
- actual_record = bt_cls.results.unknown[0]
+ actual_record = bt_cls.results.error[0]
self.assertEqual(actual_record.test_name, self.mock_test_name)
self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertEqual(actual_record.extras, MOCK_EXTRA)
- self.assertEqual(actual_record.additional_errors, {
- '_on_pass': MSG_EXPECTED_EXCEPTION
- })
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 1, 'Executed': 1,
+ 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -620,17 +615,14 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
- actual_record = bt_cls.results.unknown[0]
- self.assertEqual(bt_cls.results.failed, [])
+ actual_record = bt_cls.results.failed[0]
+ self.assertEqual(bt_cls.results.error, [])
self.assertEqual(actual_record.test_name, self.mock_test_name)
self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertEqual(actual_record.extras, MOCK_EXTRA)
- self.assertEqual(actual_record.additional_errors, {
- '_on_fail': MSG_EXPECTED_EXCEPTION
- })
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 1,
- 'Failed': 0, 'Passed': 0, 'Requested': 1, 'Skipped': 0, 'Unknown': 1
+ 'Error': 0, 'Executed': 1,
+ 'Failed': 1, 'Passed': 0, 'Requested': 1, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -652,8 +644,8 @@
self.assertEqual(bt_cls.results.failed[0].details,
MSG_EXPECTED_EXCEPTION)
expected_summary = {
- 'Blocked': 0, 'ControllerInfo': {}, 'Errors': 0, 'Executed': 2,
- 'Failed': 1, 'Passed': 1, 'Requested': 3, 'Skipped': 0, 'Unknown': 0
+ 'Error': 0, 'Executed': 2,
+ 'Failed': 1, 'Passed': 1, 'Requested': 3, 'Skipped': 0
}
self.assertEqual(bt_cls.results.summary_dict(), expected_summary)
@@ -665,7 +657,7 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run(test_names=['test_func'])
- actual_record = bt_cls.results.unknown[0]
+ actual_record = bt_cls.results.error[0]
self.assertEqual(actual_record.test_name, 'test_func')
self.assertEqual(actual_record.details, MSG_EXPECTED_EXCEPTION)
self.assertIsNone(actual_record.extras)
@@ -811,7 +803,7 @@
bt_cls = MockBaseTest(self.mock_test_cls_configs)
bt_cls.run()
- actual_record = bt_cls.results.unknown[0]
+ actual_record = bt_cls.results.error[0]
self.assertEqual(actual_record.test_name, 'test_func')
self.assertEqual(actual_record.details, MSG_UNEXPECTED_EXCEPTION)
self.assertIsNone(actual_record.extras)
@@ -986,8 +978,9 @@
base_cls = base_test.BaseTestClass(self.mock_test_cls_configs)
base_cls.register_controller(mock_controller)
registered_name = 'mock_controller'
- self.assertTrue(mock_controller in base_cls.controller_registry)
- mock_ctrlrs = base_cls.controller_registry[mock_controller]
+ controller_objects = base_cls._controller_manager._controller_objects
+ self.assertTrue(registered_name in controller_objects)
+ mock_ctrlrs = controller_objects[registered_name]
self.assertEqual(mock_ctrlrs[0].magic, 'magic1')
self.assertEqual(mock_ctrlrs[1].magic, 'magic2')
expected_msg = 'Controller module .* has already been registered.'
@@ -1027,7 +1020,8 @@
base_cls = base_test.BaseTestClass(self.mock_test_cls_configs)
base_cls.register_controller(mock_controller, builtin=True)
self.assertTrue(hasattr(base_cls, mock_ref_name))
- self.assertTrue(mock_controller in base_cls.controller_registry)
+ self.assertTrue(mock_controller.__name__ in
+ base_cls._controller_manager._controller_objects)
mock_ctrlrs = getattr(base_cls, mock_ctrlr_ref_name)
self.assertEqual(mock_ctrlrs[0].magic, 'magic1')
self.assertEqual(mock_ctrlrs[1].magic, 'magic2')
@@ -1065,31 +1059,6 @@
self.assertEqual(magic_devices[0].magic, 'magic1')
self.assertEqual(magic_devices[1].magic, 'magic2')
- def test_verify_controller_module(self):
- base_test.BaseTestClass.verify_controller_module(mock_controller)
-
- def test_verify_controller_module_null_attr(self):
- tmp = mock_controller.ACTS_CONTROLLER_CONFIG_NAME
- mock_controller.ACTS_CONTROLLER_CONFIG_NAME = None
- msg = 'Controller interface .* in .* cannot be null.'
- try:
- with self.assertRaisesRegexp(signals.ControllerError, msg):
- base_test.BaseTestClass.verify_controller_module(
- mock_controller)
- finally:
- mock_controller.ACTS_CONTROLLER_CONFIG_NAME = tmp
-
- def test_verify_controller_module_missing_attr(self):
- tmp = mock_controller.ACTS_CONTROLLER_CONFIG_NAME
- delattr(mock_controller, 'ACTS_CONTROLLER_CONFIG_NAME')
- msg = 'Module .* missing required controller module attribute'
- try:
- with self.assertRaisesRegexp(signals.ControllerError, msg):
- base_test.BaseTestClass.verify_controller_module(
- mock_controller)
- finally:
- setattr(mock_controller, 'ACTS_CONTROLLER_CONFIG_NAME', tmp)
-
if __name__ == '__main__':
unittest.main()
diff --git a/acts/framework/tests/acts_records_test.py b/acts/framework/tests/acts_records_test.py
index ee59258..a744e1a 100755
--- a/acts/framework/tests/acts_records_test.py
+++ b/acts/framework/tests/acts_records_test.py
@@ -53,7 +53,8 @@
d[records.TestResultEnums.RECORD_LOG_END_TIME] = record.log_end_time
d[records.TestResultEnums.RECORD_UID] = None
d[records.TestResultEnums.RECORD_CLASS] = None
- d[records.TestResultEnums.RECORD_ADDITIONAL_ERRORS] = {}
+ d[records.TestResultEnums.RECORD_EXTRA_ERRORS] = {}
+ d[records.TestResultEnums.RECORD_STACKTRACE] = record.stacktrace
actual_d = record.to_dict()
self.assertDictEqual(actual_d, d)
# Verify that these code paths do not cause crashes and yield non-empty
@@ -214,7 +215,7 @@
record1.test_fail(s)
record2 = records.TestResultRecord(self.tn)
record2.test_begin()
- record2.test_unknown(s)
+ record2.test_error(s)
tr = records.TestResult()
tr.add_record(record1)
tr.add_record(record2)
diff --git a/acts/framework/tests/libs/version_selector_test.py b/acts/framework/tests/libs/version_selector_test.py
index 54aa78b..53c6352 100755
--- a/acts/framework/tests/libs/version_selector_test.py
+++ b/acts/framework/tests/libs/version_selector_test.py
@@ -21,6 +21,7 @@
sys.path[0] = os.path.join(sys.path[0], '../')
import unittest
import logging
+import mock
from acts import base_test
from acts.libs import version_selector
@@ -101,6 +102,7 @@
"""Tests that VersionedTestClass (above) can be called with
test_tracker_info."""
test_class = VersionedTestClass({'log': logging.getLogger(),
+ 'summary_writer': mock.MagicMock(),
'cli_args': []})
test_class.run(['test_1', 'test_2'], 1)
diff --git a/acts/tests/google/fuchsia/wlan/WlanPhyCompliance11NTest.py b/acts/tests/google/fuchsia/wlan/WlanPhyCompliance11NTest.py
index 2a595b3..0af980f 100644
--- a/acts/tests/google/fuchsia/wlan/WlanPhyCompliance11NTest.py
+++ b/acts/tests/google/fuchsia/wlan/WlanPhyCompliance11NTest.py
@@ -116,6 +116,8 @@
Args:
ap_settings: A dictionary of hostapd constant n_capabilities.
"""
+ security_profile = None
+ password = None
temp_n_capabilities = list(ap_settings['n_capabilities'])
n_capabilities = []
for n_capability in temp_n_capabilities:
@@ -155,8 +157,7 @@
password=rand_ascii_str(20),
wpa_cipher='CCMP',
wpa2_cipher='CCMP')
- else:
- security_profile = None
+ password = security_profile.password
validate_setup_ap_and_associate(
access_point=self.access_point,
@@ -169,7 +170,7 @@
force_wmm=True,
ssid=utils.rand_ascii_str(20),
security=security_profile,
- password=security_profile.password
+ password=password
)
diff --git a/acts/tests/google/tel/live/TelLiveCBRSTest.py b/acts/tests/google/tel/live/TelLiveCBRSTest.py
new file mode 100644
index 0000000..6454db0
--- /dev/null
+++ b/acts/tests/google/tel/live/TelLiveCBRSTest.py
@@ -0,0 +1,540 @@
+#!/usr/bin/env python3.4
+#
+# Copyright 2019 - Google
+#
+# 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.
+"""
+ Test Script for CBRS devices
+"""
+
+import time
+import collections
+
+from acts.test_decorators import test_tracker_info
+from acts.test_utils.tel.TelephonyBaseTest import TelephonyBaseTest
+from acts.test_utils.tel.tel_defines import DIRECTION_MOBILE_ORIGINATED
+from acts.test_utils.tel.tel_defines import DIRECTION_MOBILE_TERMINATED
+from acts.test_utils.tel.tel_defines import WAIT_TIME_BETWEEN_REG_AND_CALL
+from acts.test_utils.tel.tel_defines import WAIT_TIME_IN_CALL
+from acts.test_utils.tel.tel_defines import WAIT_TIME_FOR_CBRS_DATA_SWITCH
+from acts.test_utils.tel.tel_test_utils import get_phone_number
+from acts.test_utils.tel.tel_test_utils import hangup_call
+from acts.test_utils.tel.tel_test_utils import hangup_call_by_adb
+from acts.test_utils.tel.tel_test_utils import initiate_call
+from acts.test_utils.tel.tel_test_utils import is_phone_not_in_call
+from acts.test_utils.tel.tel_test_utils import wait_and_answer_call
+from acts.test_utils.tel.tel_test_utils import is_phone_in_call
+from acts.test_utils.tel.tel_test_utils import start_qxdm_loggers
+from acts.test_utils.tel.tel_test_utils import test_data_browsing_success_using_sl4a
+from acts.test_utils.tel.tel_test_utils import test_data_browsing_failure_using_sl4a
+from acts.test_utils.tel.tel_test_utils import get_operator_name
+from acts.test_utils.tel.tel_test_utils import is_current_data_on_cbrs
+from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_3g
+from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_2g
+from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_csfb
+from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_iwlan
+from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_not_iwlan
+from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_volte
+from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_general
+from acts.test_utils.tel.tel_voice_utils import phone_setup_volte_for_subscription
+from acts.test_utils.tel.tel_voice_utils import phone_setup_cdma
+from acts.test_utils.tel.tel_voice_utils import phone_setup_voice_2g
+from acts.test_utils.tel.tel_voice_utils import phone_idle_3g
+from acts.test_utils.tel.tel_voice_utils import phone_idle_2g
+from acts.test_utils.tel.tel_voice_utils import phone_idle_csfb
+from acts.test_utils.tel.tel_voice_utils import phone_idle_iwlan
+from acts.test_utils.tel.tel_voice_utils import phone_idle_not_iwlan
+from acts.test_utils.tel.tel_voice_utils import phone_idle_volte
+from acts.test_utils.tel.tel_subscription_utils import get_subid_from_slot_index
+from acts.test_utils.tel.tel_subscription_utils import get_operatorname_from_slot_index
+from acts.utils import get_current_epoch_time
+
+
+class TelLiveCBRSTest(TelephonyBaseTest):
+ def __init__(self, controllers):
+ TelephonyBaseTest.__init__(self, controllers)
+ self.number_of_devices = 2
+ self.stress_test_number = self.get_stress_test_number()
+ self.message_lengths = (50, 160, 180)
+ self.long_message_lengths = (800, 1600)
+ self.cbrs_subid = None
+ self.default_data_subid = None
+
+
+ def on_fail(self, test_name, begin_time):
+ if test_name.startswith('test_stress'):
+ return
+ super().on_fail(test_name, begin_time)
+
+
+ def _cbrs_call_sequence(self, ads, mo_mt,
+ cbrs_phone_setup_func,
+ verify_cbrs_initial_idle_func,
+ verify_data_initial_func,
+ verify_cbrs_in_call_state_func,
+ verify_data_in_call_func,
+ incall_cbrs_setting_check_func,
+ verify_data_final_func,
+ verify_cbrs_final_func,
+ expected_result):
+ """_cbrs_call_sequence
+
+ Args:
+ ads: list of android devices. This list should have 2 ad.
+ mo_mt: indicating this call sequence is MO or MT.
+ Valid input: DIRECTION_MOBILE_ORIGINATED and
+ DIRECTION_MOBILE_TERMINATED.
+
+ Returns:
+ if expected_result is True,
+ Return True if call sequence finish without exception.
+ if expected_result is string,
+ Return True if expected exception happened. Otherwise False.
+
+ """
+
+ class _CBRSCallSequenceException(Exception):
+ pass
+
+ if (len(ads) != 2) or (mo_mt not in [
+ DIRECTION_MOBILE_ORIGINATED, DIRECTION_MOBILE_TERMINATED
+ ]):
+ self.log.error("Invalid parameters.")
+ return False
+
+ # Fetch CBRS sub_id and default
+ slot_dict = {0: {}, 1: {}}
+ for slot in (0, 1):
+ slot_dict[slot]['sub_id'] = get_subid_from_slot_index(
+ ads[0].log, ads[0], slot)
+ slot_dict[slot]['operator'] = get_operatorname_from_slot_index(
+ ads[0], slot)
+ if "Google" in slot_dict[slot]['operator']:
+ self.cbrs_subid = slot_dict[slot]['sub_id']
+ else:
+ self.default_data_subid = slot_dict[slot]['sub_id']
+ ads[0].log.info("Slot %d - Sub %s - %s", slot,
+ slot_dict[slot]['sub_id'],
+ slot_dict[slot]['operator'])
+
+ if mo_mt == DIRECTION_MOBILE_ORIGINATED:
+ ad_caller = ads[0]
+ ad_callee = ads[1]
+ caller_number = get_phone_number(self.log, ad_caller)
+ callee_number = get_phone_number(self.log, ad_callee)
+ mo_operator = get_operator_name(ads[0].log, ads[0])
+ mt_operator = get_operator_name(ads[1].log, ads[1])
+ else:
+ ad_caller = ads[1]
+ ad_callee = ads[0]
+ caller_number = get_phone_number(self.log, ad_caller)
+ callee_number = get_phone_number(self.log, ad_callee)
+ mt_operator = get_operator_name(ads[0].log, ads[0])
+ mo_operator = get_operator_name(ads[1].log, ads[1])
+
+ self.log.info("-->Begin cbrs_call_sequence: %s to %s<--",
+ caller_number, callee_number)
+ self.log.info("--> %s to %s <--", mo_operator, mt_operator)
+
+ try:
+ # Setup
+ if cbrs_phone_setup_func and not cbrs_phone_setup_func():
+ raise _CBRSCallSequenceException("cbrs_phone_setup_func fail.")
+ if not phone_setup_voice_general(self.log, ads[1]):
+ raise _CBRSCallSequenceException(
+ "phone_setup_voice_general fail.")
+ time.sleep(WAIT_TIME_BETWEEN_REG_AND_CALL)
+
+ # Ensure idle status correct
+ if verify_cbrs_initial_idle_func and not \
+ verify_cbrs_initial_idle_func():
+ raise _CBRSCallSequenceException(
+ "verify_cbrs_initial_idle_func fail.")
+
+ # Ensure data checks are performed
+ if verify_data_initial_func and not \
+ verify_data_initial_func():
+ raise _CBRSCallSequenceException(
+ "verify_data_initial_func fail.")
+
+ # Make MO/MT call.
+ if not initiate_call(self.log, ad_caller, callee_number):
+ raise _CBRSCallSequenceException("initiate_call fail.")
+ if not wait_and_answer_call(self.log, ad_callee, caller_number):
+ raise _CBRSCallSequenceException("wait_and_answer_call fail.")
+ time.sleep(WAIT_TIME_FOR_CBRS_DATA_SWITCH)
+
+ # Check state, wait 30 seconds, check again.
+ if (verify_cbrs_in_call_state_func and not
+ verify_cbrs_in_call_state_func()):
+ raise _CBRSCallSequenceException(
+ "verify_cbrs_in_call_state_func fail.")
+
+ if is_phone_not_in_call(self.log, ads[1]):
+ raise _CBRSCallSequenceException("PhoneB not in call.")
+
+ # Ensure data checks are performed
+ if verify_data_in_call_func and not \
+ verify_data_in_call_func():
+ raise _CBRSCallSequenceException(
+ "verify_data_in_call_func fail.")
+
+ time.sleep(WAIT_TIME_IN_CALL)
+
+ if (verify_cbrs_in_call_state_func and not
+ verify_cbrs_in_call_state_func()):
+ raise _CBRSCallSequenceException(
+ "verify_cbrs_in_call_state_func fail after 30 seconds.")
+ if is_phone_not_in_call(self.log, ads[1]):
+ raise _CBRSCallSequenceException(
+ "PhoneB not in call after 30 seconds.")
+
+ # in call change setting and check
+ if (incall_cbrs_setting_check_func and not
+ incall_cbrs_setting_check_func()):
+ raise _CBRSCallSequenceException(
+ "incall_cbrs_setting_check_func fail.")
+
+ # Hangup call
+ if is_phone_in_call(self.log, ads[0]):
+ if not hangup_call(self.log, ads[0]):
+ raise _CBRSCallSequenceException("hangup_call fail.")
+ else:
+ if incall_cbrs_setting_check_func is None:
+ raise _CBRSCallSequenceException("Unexpected call drop.")
+
+ time.sleep(WAIT_TIME_FOR_CBRS_DATA_SWITCH)
+
+ # Ensure data checks are performed
+ if verify_data_final_func and not \
+ verify_data_final_func():
+ raise _CBRSCallSequenceException(
+ "verify_data_final_func fail.")
+
+ # Ensure data checks are performed
+ if verify_cbrs_final_func and not \
+ verify_cbrs_final_func():
+ raise _CBRSCallSequenceException(
+ "verify_cbrs_final_func fail.")
+
+ except _CBRSCallSequenceException as e:
+ if str(e) == expected_result:
+ self.log.info("Expected exception: <%s>, return True.", e)
+ return True
+ else:
+ self.log.info("Unexpected exception: <%s>, return False.", e)
+ return False
+ finally:
+ for ad in ads:
+ if ad.droid.telecomIsInCall():
+ hangup_call_by_adb(ad)
+ self.log.info("cbrs_call_sequence finished, return %s",
+ expected_result is True)
+ return (expected_result is True)
+
+
+ def _phone_idle_iwlan(self):
+ return phone_idle_iwlan(self.log, self.android_devices[0])
+
+ def _phone_idle_not_iwlan(self):
+ return phone_idle_not_iwlan(self.log, self.android_devices[0])
+
+ def _phone_idle_volte(self):
+ return phone_idle_volte(self.log, self.android_devices[0])
+
+ def _phone_idle_csfb(self):
+ return phone_idle_csfb(self.log, self.android_devices[0])
+
+ def _phone_idle_3g(self):
+ return phone_idle_3g(self.log, self.android_devices[0])
+
+ def _phone_idle_2g(self):
+ return phone_idle_2g(self.log, self.android_devices[0])
+
+ def _is_phone_in_call_iwlan(self):
+ return is_phone_in_call_iwlan(self.log, self.android_devices[0])
+
+ def _is_phone_in_call_not_iwlan(self):
+ return is_phone_in_call_not_iwlan(self.log, self.android_devices[0])
+
+ def _is_phone_not_in_call(self):
+ if is_phone_in_call(self.log, self.android_devices[0]):
+ self.log.info("{} in call.".format(self.android_devices[0].serial))
+ return False
+ self.log.info("{} not in call.".format(self.android_devices[0].serial))
+ return True
+
+ def _is_phone_in_call_volte(self):
+ return is_phone_in_call_volte(self.log, self.android_devices[0])
+
+ def _is_phone_in_call_3g(self):
+ return is_phone_in_call_3g(self.log, self.android_devices[0])
+
+ def _is_phone_in_call_2g(self):
+ return is_phone_in_call_2g(self.log, self.android_devices[0])
+
+ def _is_phone_in_call_csfb(self):
+ return is_phone_in_call_csfb(self.log, self.android_devices[0])
+
+ def _is_phone_in_call(self):
+ return is_phone_in_call(self.log, self.android_devices[0])
+
+ def _phone_setup_voice_general(self):
+ return phone_setup_voice_general(self.log, self.android_devices[0])
+
+ def _phone_setup_volte(self):
+ return phone_setup_volte_for_subscription(self.log,
+ self.android_devices[0],
+ self.default_data_subid)
+
+ def _phone_setup_1x(self):
+ return phone_setup_cdma(self.log, self.android_devices[0])
+
+ def _phone_setup_2g(self):
+ return phone_setup_voice_2g(self.log, self.android_devices[0])
+
+
+ def _test_data_browsing_success_using_sl4a(self):
+ return test_data_browsing_success_using_sl4a(self.log,
+ self.android_devices[0])
+
+ def _test_data_browsing_failure_using_sl4a(self):
+ return test_data_browsing_failure_using_sl4a(self.log,
+ self.android_devices[0])
+
+ def _is_current_data_on_cbrs(self):
+ return is_current_data_on_cbrs(self.android_devices[0],
+ self.cbrs_subid)
+
+ def _is_current_data_on_default(self):
+ return not is_current_data_on_cbrs(self.android_devices[0],
+ self.cbrs_subid)
+
+
+ """ Tests Begin """
+
+
+ @test_tracker_info(uuid="f7a3db92-2f1b-4131-99bc-b323dbce812c")
+ @TelephonyBaseTest.tel_test_wrap
+ def test_cbrs_mo_voice_data_concurrency_lte(self):
+ """ Test CBRS Data with MO Voice Call on LTE
+
+ PhoneA should be on LTE with VoLTE enabled
+ Verify Data Browsing works fine on cbrs before call
+ Call from PhoneA to PhoneB, call should succeed
+ Verify Data Browsing works fine on pSIM during call
+ Terminate call
+ Verify Data Browsing works fine on cbrs after call
+
+ Returns:
+ True if pass; False if fail.
+ """
+ ads = [self.android_devices[0], self.android_devices[1]]
+ result = self._cbrs_call_sequence(
+ ads, DIRECTION_MOBILE_ORIGINATED,
+ self._phone_setup_volte, self._is_current_data_on_cbrs,
+ self._test_data_browsing_success_using_sl4a,
+ self._is_phone_in_call_volte,
+ self._is_current_data_on_default,
+ self._test_data_browsing_success_using_sl4a,
+ self._test_data_browsing_success_using_sl4a,
+ self._is_current_data_on_cbrs, True)
+
+ self.log.info("CBRS MO Result: %s", result)
+ return result
+
+
+ @test_tracker_info(uuid="17acce7a-de9c-4540-b2d3-2c98367a0b4e")
+ @TelephonyBaseTest.tel_test_wrap
+ def test_cbrs_mt_voice_data_concurrency_lte(self):
+ """ Test CBRS Data with MT Voice Call on LTE
+
+ PhoneA should be on LTE with VoLTE enabled
+ Verify Data Browsing works fine on cbrs before call
+ Call from PhoneB to PhoneA, call should succeed
+ Verify Data Browsing works fine on pSIM during call
+ Terminate call
+ Verify Data Browsing works fine on cbrs after call
+
+ Returns:
+ True if pass; False if fail.
+ """
+ ads = [self.android_devices[0], self.android_devices[1]]
+ result = self._cbrs_call_sequence(
+ ads, DIRECTION_MOBILE_TERMINATED,
+ self._phone_setup_volte, self._is_current_data_on_cbrs,
+ self._test_data_browsing_success_using_sl4a,
+ self._is_phone_in_call_volte,
+ self._is_current_data_on_default,
+ self._test_data_browsing_success_using_sl4a,
+ self._test_data_browsing_success_using_sl4a,
+ self._is_current_data_on_cbrs, True)
+
+ self.log.info("CBRS MT Result: %s", result)
+ return result
+
+ @test_tracker_info(uuid="dc2608fc-b99d-419b-8989-e1f8cdeb04da")
+ @TelephonyBaseTest.tel_test_wrap
+ def test_cbrs_mo_voice_data_concurrency_1x(self):
+ """ Test CBRS Data with MO Voice Call on 3G
+
+ PhoneA should be on UMTS
+ Verify Data Browsing works fine on cbrs before call
+ Call from PhoneA to PhoneB, call should succeed
+ Verify Data Browsing works fine on pSIM during call
+ Terminate call
+ Verify Data Browsing works fine on cbrs after call
+
+ Returns:
+ True if pass; False if fail.
+ """
+ ads = [self.android_devices[0], self.android_devices[1]]
+ result = self._cbrs_call_sequence(
+ ads, DIRECTION_MOBILE_ORIGINATED,
+ self._phone_setup_1x, self._is_current_data_on_cbrs,
+ self._test_data_browsing_success_using_sl4a,
+ self._is_phone_in_call_3g,
+ self._is_current_data_on_default,
+ self._test_data_browsing_failure_using_sl4a,
+ self._test_data_browsing_success_using_sl4a,
+ self._is_current_data_on_cbrs, True)
+
+ self.log.info("CBRS MO Result: %s", result)
+ return result
+
+
+ @test_tracker_info(uuid="cd3a6613-ca37-43c7-8364-7e4e627ca558")
+ @TelephonyBaseTest.tel_test_wrap
+ def test_cbrs_mt_voice_data_concurrency_1x(self):
+ """ Test CBRS Data with MT Voice Call on 3G
+
+ PhoneA should be on UMTS
+ Verify Data Browsing works fine on cbrs before call
+ Call from PhoneA to PhoneA, call should succeed
+ Verify Data Browsing works fine on pSIM during call
+ Terminate call
+ Verify Data Browsing works fine on cbrs after call
+
+ Returns:
+ True if pass; False if fail.
+ """
+ ads = [self.android_devices[0], self.android_devices[1]]
+ result = self._cbrs_call_sequence(
+ ads, DIRECTION_MOBILE_TERMINATED,
+ self._phone_setup_1x, self._is_current_data_on_cbrs,
+ self._test_data_browsing_success_using_sl4a,
+ self._is_phone_in_call_3g,
+ self._is_current_data_on_default,
+ self._test_data_browsing_failure_using_sl4a,
+ self._test_data_browsing_success_using_sl4a,
+ self._is_current_data_on_cbrs, True)
+
+ self.log.info("CBRS MT Result: %s", result)
+ return result
+
+
+ def _test_stress_cbrs(self, mo_mt):
+ """ Test CBRS/SSIM VoLTE Stress
+
+ mo_mt: indicating this call sequence is MO or MT.
+ Valid input: DIRECTION_MOBILE_ORIGINATED and
+ DIRECTION_MOBILE_TERMINATED.
+
+ Returns:
+ True if pass; False if fail.
+ """
+ if (mo_mt not in [DIRECTION_MOBILE_ORIGINATED,
+ DIRECTION_MOBILE_TERMINATED]):
+ self.log.error("Invalid parameters.")
+ return False
+ ads = [self.android_devices[0], self.android_devices[1]]
+ total_iteration = self.stress_test_number
+ fail_count = collections.defaultdict(int)
+ slot_dict = {0: {}, 1: {}}
+ for slot in (0, 1):
+ slot_dict[slot]['sub_id'] = get_subid_from_slot_index(
+ ads[0].log, ads[0], slot)
+ slot_dict[slot]['operator'] = get_operatorname_from_slot_index(
+ ads[0], slot)
+ if "Google" in slot_dict[slot]['operator']:
+ self.cbrs_subid = slot_dict[slot]['sub_id']
+ else:
+ self.default_data_subid = slot_dict[slot]['sub_id']
+ ads[0].log.info("Slot %d - Sub %s - %s", slot,
+ slot_dict[slot]['sub_id'],
+ slot_dict[slot]['operator'])
+ self.log.info("Total iteration = %d.", total_iteration)
+ current_iteration = 1
+ for i in range(1, total_iteration + 1):
+ msg = "Stress Call Test Iteration: <%s> / <%s>" % (
+ i, total_iteration)
+ begin_time = get_current_epoch_time()
+ self.log.info(msg)
+ start_qxdm_loggers(self.log, self.android_devices, begin_time)
+ iteration_result = self._cbrs_call_sequence(
+ ads, mo_mt,
+ self._phone_setup_volte,
+ self._is_current_data_on_cbrs,
+ self._test_data_browsing_success_using_sl4a,
+ self._is_phone_in_call_volte,
+ self._is_current_data_on_default,
+ self._test_data_browsing_success_using_sl4a,
+ self._test_data_browsing_success_using_sl4a,
+ self._is_current_data_on_cbrs, True)
+ self.log.info("Result: %s", iteration_result)
+ if iteration_result:
+ self.log.info(">----Iteration : %d/%d succeed.----<",
+ i, total_iteration)
+ else:
+ fail_count["cbrs_fail"] += 1
+ self.log.error(">----Iteration : %d/%d failed.----<",
+ i, total_iteration)
+ self._take_bug_report("%s_IterNo_%s" % (self.test_name, i),
+ begin_time)
+ current_iteration += 1
+ test_result = True
+ for failure, count in fail_count.items():
+ if count:
+ self.log.error("%s: %s %s failures in %s iterations",
+ self.test_name, count, failure,
+ total_iteration)
+ test_result = False
+ return test_result
+
+ @test_tracker_info(uuid="860dc00d-5be5-4cdd-aeb1-a89edfa83342")
+ @TelephonyBaseTest.tel_test_wrap
+ def test_stress_cbrs_mt_calls_lte(self):
+ """ Test SSIM to CBRS stress
+
+ Call from PhoneB to PhoneA
+ Perform CBRS Data checks
+ Repeat above steps
+
+ Returns:
+ True if pass; False if fail.
+ """
+ return self._test_stress_cbrs(DIRECTION_MOBILE_TERMINATED)
+
+ @test_tracker_info(uuid="54366c70-c9cb-4eed-bd1c-a37c83d5c0ae")
+ @TelephonyBaseTest.tel_test_wrap
+ def test_stress_cbrs_mo_calls_lte(self):
+ """ Test CBRS to SSIM stress
+
+ Call from PhoneA to PhoneB
+ Perform CBRS Data checks
+ Repeat above steps
+
+ Returns:
+ True if pass; False if fail.
+ """
+ return self._test_stress_cbrs(DIRECTION_MOBILE_ORIGINATED)
\ No newline at end of file
diff --git a/acts/tests/google/tel/live/TelLiveStressTest.py b/acts/tests/google/tel/live/TelLiveStressTest.py
index 77007e6..afd6e34 100644
--- a/acts/tests/google/tel/live/TelLiveStressTest.py
+++ b/acts/tests/google/tel/live/TelLiveStressTest.py
@@ -40,8 +40,6 @@
from acts.test_utils.tel.tel_defines import NETWORK_MODE_CDMA
from acts.test_utils.tel.tel_defines import NETWORK_MODE_GSM_ONLY
from acts.test_utils.tel.tel_defines import NETWORK_MODE_TDSCDMA_GSM_WCDMA
-from acts.test_utils.tel.tel_defines import RAT_LTE
-from acts.test_utils.tel.tel_defines import RAT_UNKNOWN
from acts.test_utils.tel.tel_defines import WAIT_TIME_AFTER_MODE_CHANGE
from acts.test_utils.tel.tel_defines import WFC_MODE_CELLULAR_PREFERRED
from acts.test_utils.tel.tel_defines import WFC_MODE_WIFI_PREFERRED
@@ -52,7 +50,6 @@
from acts.test_utils.tel.tel_test_utils import active_file_download_test
from acts.test_utils.tel.tel_test_utils import is_phone_in_call
from acts.test_utils.tel.tel_test_utils import call_setup_teardown
-from acts.test_utils.tel.tel_test_utils import check_is_wifi_connected
from acts.test_utils.tel.tel_test_utils import ensure_network_generation_for_subscription
from acts.test_utils.tel.tel_test_utils import ensure_wifi_connected
from acts.test_utils.tel.tel_test_utils import extract_test_log
@@ -77,7 +74,7 @@
from acts.test_utils.tel.tel_test_utils import wait_for_call_id_clearing
from acts.test_utils.tel.tel_test_utils import wait_for_data_connection
from acts.test_utils.tel.tel_test_utils import wait_for_in_call_active
-from acts.test_utils.tel.tel_test_utils import wifi_toggle_state
+from acts.test_utils.tel.tel_test_utils import check_call_state_idle_by_adb
from acts.test_utils.tel.tel_test_utils import is_current_data_on_cbrs
from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_3g
from acts.test_utils.tel.tel_voice_utils import is_phone_in_call_2g
@@ -92,11 +89,6 @@
from acts.test_utils.tel.tel_voice_utils import phone_idle_iwlan
from acts.test_utils.tel.tel_voice_utils import phone_idle_volte
from acts.test_utils.tel.tel_voice_utils import get_current_voice_rat
-from acts.test_utils.tel.tel_subscription_utils import get_default_data_sub_id
-from acts.test_utils.tel.tel_subscription_utils import get_outgoing_message_sub_id
-from acts.test_utils.tel.tel_subscription_utils import get_outgoing_voice_sub_id
-from acts.test_utils.tel.tel_subscription_utils import get_incoming_voice_sub_id
-from acts.test_utils.tel.tel_subscription_utils import get_incoming_message_sub_id
from acts.test_utils.tel.tel_subscription_utils import get_subid_from_slot_index
from acts.test_utils.tel.tel_subscription_utils import get_operatorname_from_slot_index
from acts.test_utils.tel.tel_subscription_utils import set_subid_for_data
@@ -698,12 +690,20 @@
self.result_info["CBRS-Data-Pass"] += 1
cbrs_fail = False
else:
- self.result_info["CBRS-Data-Fail"] += 1
- cbrs_fail = True
+ if not check_call_state_idle_by_adb(ad):
+ self.result_info["CBRS-InCall-Pass"] += 1
+ cbrs_fail = False
+ else:
+ self.result_info["CBRS-Data-Fail"] += 1
+ cbrs_fail = True
else:
if is_current_data_on_cbrs(ad, ad.cbrs):
- self.result_info["CBRS-InCall-Fail"] += 1
- cbrs_fail = True
+ if check_call_state_idle_by_adb(ad):
+ self.result_info["CBRS-Data-Pass"] += 1
+ cbrs_fail = False
+ else:
+ self.result_info["CBRS-InCall-Fail"] += 1
+ cbrs_fail = True
else:
self.result_info["CBRS-InCall-Pass"] += 1
cbrs_fail = False
