blob: 9c504fceb0c5b4c7f1bdcce26b6c971bd001f80c [file] [log] [blame]
#!/usr/bin/env python3
# Copyright 2017 - 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
from acts.libs.testtracker.testtracker_results_writer import KEY_EFFORT_NAME
from acts.libs.testtracker.testtracker_results_writer import TestTrackerError
from acts.libs.testtracker.testtracker_results_writer import TestTrackerResultsWriter
from mobly.base_test import BaseTestClass
from acts import signals
def test_info(predicate=None, **keyed_info):
"""Adds info about test.
Extra info to include about the test. This info will be available in the
test output. Note that if a key is given multiple times it will be added
as a list of all values. If multiples of these are stacked their results
will be merged.
# This test will have a variable my_var
@test_info(my_var='THIS IS MY TEST')
def my_test(self):
return False
predicate: A func to call that if false will skip adding this test
info. Function signature is bool(test_obj, args, kwargs)
**keyed_info: The key, value info to include in the extras for this
def test_info_decorator(func):
return _TestInfoDecoratorFunc(func, predicate, keyed_info)
return test_info_decorator
def __select_last(test_signals, _):
return test_signals[-1]
def repeated_test(num_passes, acceptable_failures=0,
"""A decorator that runs a test case multiple times.
This decorator can be used to run a test multiple times and aggregate the
data into a single test result. By setting `result_selector`, the user can
access the returned result of each run, allowing them to average results,
return the median, or gather and return standard deviation values.
This decorator should be used on test cases, and should not be used on
static or class methods. The test case must take in an additional argument,
`attempt_number`, which returns the current attempt number, starting from
Note that any TestSignal intended to abort or skip the test will take
abort or skip immediately.
num_passes: The number of times the test needs to pass to report the
test case as passing.
acceptable_failures: The number of failures accepted. If the failures
exceeds this number, the test will stop repeating. The maximum
number of runs is `num_passes + acceptable_failures`. If the test
does fail, result_selector will still be called.
result_selector: A lambda that takes in the list of TestSignals and
returns the test signal to report the test case as. Note that the
list also contains any uncaught exceptions from the test execution.
def decorator(func):
if not func.__name__.startswith('test_'):
raise ValueError('Tests must start with "test_".')
def test_wrapper(self):
num_failures = 0
num_seen_passes = 0
test_signals_received = []
for i in range(num_passes + acceptable_failures):
func(self, i + 1)
except (signals.TestFailure, signals.TestError,
AssertionError) as signal:
num_failures += 1
except signals.TestPass as signal:
num_seen_passes += 1
except (signals.TestSignal, KeyboardInterrupt):
except Exception as signal:
num_failures += 1
num_seen_passes += 1
'Test iteration %s of %s passed without details.' % (
i, func.__name__)))
if num_failures > acceptable_failures:
elif num_seen_passes == num_passes:
raise result_selector(test_signals_received, self)
return test_wrapper
return decorator
def test_tracker_info(uuid, extra_environment_info=None, predicate=None):
"""Decorator for adding test tracker info to tests results.
Will add test tracker info inside of Extras/test_tracker_info.
# This test will be linked to test tracker uuid abcd
def my_test(self):
return False
uuid: The uuid of the test case in test tracker.
extra_environment_info: Extra info about the test tracker environment.
predicate: A func that if false when called will ignore this info.
def test_tracker_info_decorator(func):
keyed_info = dict(test_tracker_uuid=uuid,
return _TestTrackerInfoDecoratorFunc(func, predicate, keyed_info)
return test_tracker_info_decorator
class _TestInfoDecoratorFunc(object):
"""Object that acts as a function decorator test info."""
def __init__(self, func, predicate, keyed_info):
self.func = func
self.predicate = predicate
self.keyed_info = keyed_info
self.__name__ = func.__name__
self.__doc__ = func.__doc__
self.__module__ = func.__module__
def __get__(self, instance, owner):
"""Called by Python to create a binding for an instance closure.
When called by Python this object will create a special binding for
that instance. That binding will know how to interact with this
specific decorator.
return _TestInfoBinding(self, instance)
def __call__(self, *args, **kwargs):
When called runs the underlying func and then attaches test info
to a signal.
new_signal = self._get_signal_from_func_call(*args, **kwargs)
raise new_signal
def _get_signal_from_func_call(self, *args, **kwargs):
"""Calls the underlying func, then attaches test info to the resulting
signal and raises the signal.
cause = None
result = self.func(*args, **kwargs)
if result or result is None:
new_signal = signals.TestPass('')
new_signal = signals.TestFailure('')
except signals.TestSignal as signal:
new_signal = signal
except Exception as ex:
cause = ex
new_signal = signals.TestError(cause)
if new_signal.extras is None:
new_signal.extras = {}
if not isinstance(new_signal.extras, dict):
raise ValueError('test_info can only append to signal data '
'that has a dict as the extra value.')
gathered_extras = self._gather_local_info(None, *args, **kwargs)
for k, v in gathered_extras.items():
if k not in new_signal.extras:
new_signal.extras[k] = v
if not isinstance(new_signal.extras[k], list):
new_signal.extras[k] = [new_signal.extras[k]]
new_signal.extras[k].insert(0, v)
raise new_signal from cause
def gather(self, *args, **kwargs):
Gathers the info from this decorator without invoking the underlying
function. This will also gather all child info if the underlying func
has that ability.
Returns: A dictionary of info.
if hasattr(self.func, 'gather'):
extras = self.func.gather(*args, **kwargs)
extras = {}
self._gather_local_info(extras, *args, **kwargs)
return extras
def _gather_local_info(self, gather_into, *args, **kwargs):
"""Gathers info from this decorator and ignores children.
gather_into: Gathers into a dictionary that already exists.
Returns: The dictionary with gathered info in it.
if gather_into is None:
extras = {}
extras = gather_into
if not self.predicate or self.predicate(args, kwargs):
for k, v in self.keyed_info.items():
if v and k not in extras:
extras[k] = v
elif v and k in extras:
if not isinstance(extras[k], list):
extras[k] = [extras[k]]
extras[k].insert(0, v)
return extras
class _TestTrackerInfoDecoratorFunc(_TestInfoDecoratorFunc):
Expands on _TestInfoDecoratorFunc by writing gathered test info to a
TestTracker proto file
def __call__(self, *args, **kwargs):
When called runs the underlying func and then attaches test info
to a signal. It then writes the result from the signal to a TestTracker
Result proto file.
self._get_signal_from_func_call(*args, **kwargs)
except signals.TestSignal as new_signal:
if not args or not isinstance(args[0], BaseTestClass):
logging.warning('The decorated object must be an instance of'
'an ACTS/Mobly test class.')
self._write_to_testtracker(args[0], new_signal)
raise new_signal
def _write_to_testtracker(self, test_instance, signal):
"""Write test result from given signal to a TestTracker Result proto
Due to infra contraints on nested structures in userparams, this
expects the test_instance to have user_params defined as follows:
testtracker_properties: A comma-delimited list of
<userparam_name>: testtracker property value.
tt_prop_to_param_names = test_instance.user_params.get(
if not tt_prop_to_param_names:
tt_prop_to_param_names = tt_prop_to_param_names.split(',')
testtracker_properties = {}
for entry in tt_prop_to_param_names:
prop_name, param_name = entry.split('=')
if param_name in test_instance.user_params:
testtracker_properties[prop_name] = (
if (hasattr(test_instance, 'android_devices') and
KEY_EFFORT_NAME not in testtracker_properties):
testtracker_properties[KEY_EFFORT_NAME] = (
writer = TestTrackerResultsWriter(
test_instance.log_path, testtracker_properties)
signal, test_instance.begin_time)
except TestTrackerError:
test_instance.log.exception('TestTracker Error')
class _TestInfoBinding(object):
When Python creates an instance of an object it creates a binding object
for each closure that contains what the instance variable should be when
called. This object is a similar binding for _TestInfoDecoratorFunc.
When Python tries to create a binding of a _TestInfoDecoratorFunc it
will return one of these objects to hold the instance for that closure.
def __init__(self, target, instance):
target: The target for creating a binding to.
instance: The instance to bind the target with.
""" = target
self.instance = instance
self.__name__ = target.__name__
def __call__(self, *args, **kwargs):
When this object is called it will call the target with the bound
return, *args, **kwargs)
def gather(self, *args, **kwargs):
Will gather the target with the bound instance.
return, *args, **kwargs)