1#!/usr/bin/env python3 2# 3# Copyright 2017 - The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16 17import logging 18 19from acts.libs.testtracker.testtracker_results_writer import KEY_EFFORT_NAME 20from acts.libs.testtracker.testtracker_results_writer import TestTrackerError 21from acts.libs.testtracker.testtracker_results_writer import TestTrackerResultsWriter 22from mobly.base_test import BaseTestClass 23 24from acts import signals 25 26 27def test_info(predicate=None, **keyed_info): 28 """Adds info about test. 29 30 Extra info to include about the test. This info will be available in the 31 test output. Note that if a key is given multiple times it will be added 32 as a list of all values. If multiples of these are stacked their results 33 will be merged. 34 35 Example: 36 # This test will have a variable my_var 37 @test_info(my_var='THIS IS MY TEST') 38 def my_test(self): 39 return False 40 41 Args: 42 predicate: A func to call that if false will skip adding this test 43 info. Function signature is bool(test_obj, args, kwargs) 44 **keyed_info: The key, value info to include in the extras for this 45 test. 46 """ 47 48 def test_info_decorator(func): 49 return _TestInfoDecoratorFunc(func, predicate, keyed_info) 50 51 return test_info_decorator 52 53 54def __select_last(test_signals, _): 55 return test_signals[-1] 56 57 58def repeated_test(num_passes, acceptable_failures=0, 59 result_selector=__select_last): 60 """A decorator that runs a test case multiple times. 61 62 This decorator can be used to run a test multiple times and aggregate the 63 data into a single test result. By setting `result_selector`, the user can 64 access the returned result of each run, allowing them to average results, 65 return the median, or gather and return standard deviation values. 66 67 This decorator should be used on test cases, and should not be used on 68 static or class methods. The test case must take in an additional argument, 69 `attempt_number`, which returns the current attempt number, starting from 70 1. 71 72 Note that any TestSignal intended to abort or skip the test will take 73 abort or skip immediately. 74 75 Args: 76 num_passes: The number of times the test needs to pass to report the 77 test case as passing. 78 acceptable_failures: The number of failures accepted. If the failures 79 exceeds this number, the test will stop repeating. The maximum 80 number of runs is `num_passes + acceptable_failures`. If the test 81 does fail, result_selector will still be called. 82 result_selector: A lambda that takes in the list of TestSignals and 83 returns the test signal to report the test case as. Note that the 84 list also contains any uncaught exceptions from the test execution. 85 """ 86 def decorator(func): 87 if not func.__name__.startswith('test_'): 88 raise ValueError('Tests must start with "test_".') 89 90 def test_wrapper(self): 91 num_failures = 0 92 num_seen_passes = 0 93 test_signals_received = [] 94 for i in range(num_passes + acceptable_failures): 95 try: 96 func(self, i + 1) 97 except (signals.TestFailure, signals.TestError, 98 AssertionError) as signal: 99 test_signals_received.append(signal) 100 num_failures += 1 101 except signals.TestPass as signal: 102 test_signals_received.append(signal) 103 num_seen_passes += 1 104 except (signals.TestSignal, KeyboardInterrupt): 105 raise 106 except Exception as signal: 107 test_signals_received.append(signal) 108 num_failures += 1 109 else: 110 num_seen_passes += 1 111 test_signals_received.append(signals.TestPass( 112 'Test iteration %s of %s passed without details.' % ( 113 i, func.__name__))) 114 115 if num_failures > acceptable_failures: 116 break 117 elif num_seen_passes == num_passes: 118 break 119 else: 120 self.teardown_test() 121 self.setup_test() 122 123 raise result_selector(test_signals_received, self) 124 125 return test_wrapper 126 127 return decorator 128 129 130def test_tracker_info(uuid, extra_environment_info=None, predicate=None): 131 """Decorator for adding test tracker info to tests results. 132 133 Will add test tracker info inside of Extras/test_tracker_info. 134 135 Example: 136 # This test will be linked to test tracker uuid abcd 137 @test_tracker_info(uuid='abcd') 138 def my_test(self): 139 return False 140 141 Args: 142 uuid: The uuid of the test case in test tracker. 143 extra_environment_info: Extra info about the test tracker environment. 144 predicate: A func that if false when called will ignore this info. 145 """ 146 147 def test_tracker_info_decorator(func): 148 keyed_info = dict(test_tracker_uuid=uuid, 149 test_tracker_environment_info=extra_environment_info) 150 return _TestTrackerInfoDecoratorFunc(func, predicate, keyed_info) 151 152 return test_tracker_info_decorator 153 154 155class _TestInfoDecoratorFunc(object): 156 """Object that acts as a function decorator test info.""" 157 158 def __init__(self, func, predicate, keyed_info): 159 self.func = func 160 self.predicate = predicate 161 self.keyed_info = keyed_info 162 self.__name__ = func.__name__ 163 self.__doc__ = func.__doc__ 164 self.__module__ = func.__module__ 165 166 def __get__(self, instance, owner): 167 """Called by Python to create a binding for an instance closure. 168 169 When called by Python this object will create a special binding for 170 that instance. That binding will know how to interact with this 171 specific decorator. 172 """ 173 return _TestInfoBinding(self, instance) 174 175 def __call__(self, *args, **kwargs): 176 """ 177 When called runs the underlying func and then attaches test info 178 to a signal. 179 """ 180 new_signal = self._get_signal_from_func_call(*args, **kwargs) 181 raise new_signal 182 183 def _get_signal_from_func_call(self, *args, **kwargs): 184 """Calls the underlying func, then attaches test info to the resulting 185 signal and raises the signal. 186 """ 187 cause = None 188 try: 189 result = self.func(*args, **kwargs) 190 191 if result or result is None: 192 new_signal = signals.TestPass('') 193 else: 194 new_signal = signals.TestFailure('') 195 except signals.TestSignal as signal: 196 new_signal = signal 197 except Exception as ex: 198 cause = ex 199 new_signal = signals.TestError(cause) 200 201 if new_signal.extras is None: 202 new_signal.extras = {} 203 if not isinstance(new_signal.extras, dict): 204 raise ValueError('test_info can only append to signal data ' 205 'that has a dict as the extra value.') 206 207 gathered_extras = self._gather_local_info(None, *args, **kwargs) 208 for k, v in gathered_extras.items(): 209 if k not in new_signal.extras: 210 new_signal.extras[k] = v 211 else: 212 if not isinstance(new_signal.extras[k], list): 213 new_signal.extras[k] = [new_signal.extras[k]] 214 215 new_signal.extras[k].insert(0, v) 216 217 raise new_signal from cause 218 219 def gather(self, *args, **kwargs): 220 """ 221 Gathers the info from this decorator without invoking the underlying 222 function. This will also gather all child info if the underlying func 223 has that ability. 224 225 Returns: A dictionary of info. 226 """ 227 if hasattr(self.func, 'gather'): 228 extras = self.func.gather(*args, **kwargs) 229 else: 230 extras = {} 231 232 self._gather_local_info(extras, *args, **kwargs) 233 234 return extras 235 236 def _gather_local_info(self, gather_into, *args, **kwargs): 237 """Gathers info from this decorator and ignores children. 238 239 Args: 240 gather_into: Gathers into a dictionary that already exists. 241 242 Returns: The dictionary with gathered info in it. 243 """ 244 if gather_into is None: 245 extras = {} 246 else: 247 extras = gather_into 248 if not self.predicate or self.predicate(args, kwargs): 249 for k, v in self.keyed_info.items(): 250 if v and k not in extras: 251 extras[k] = v 252 elif v and k in extras: 253 if not isinstance(extras[k], list): 254 extras[k] = [extras[k]] 255 extras[k].insert(0, v) 256 257 return extras 258 259 260class _TestTrackerInfoDecoratorFunc(_TestInfoDecoratorFunc): 261 """ 262 Expands on _TestInfoDecoratorFunc by writing gathered test info to a 263 TestTracker proto file 264 """ 265 266 def __call__(self, *args, **kwargs): 267 """ 268 When called runs the underlying func and then attaches test info 269 to a signal. It then writes the result from the signal to a TestTracker 270 Result proto file. 271 """ 272 try: 273 self._get_signal_from_func_call(*args, **kwargs) 274 except signals.TestSignal as new_signal: 275 if not args or not isinstance(args[0], BaseTestClass): 276 logging.warning('The decorated object must be an instance of' 277 'an ACTS/Mobly test class.') 278 else: 279 self._write_to_testtracker(args[0], new_signal) 280 raise new_signal 281 282 def _write_to_testtracker(self, test_instance, signal): 283 """Write test result from given signal to a TestTracker Result proto 284 file. 285 286 Due to infra contraints on nested structures in userparams, this 287 expects the test_instance to have user_params defined as follows: 288 289 testtracker_properties: A comma-delimited list of 290 'prop_name=<userparam_name>' 291 <userparam_name>: testtracker property value. 292 """ 293 tt_prop_to_param_names = test_instance.user_params.get( 294 'testtracker_properties') 295 296 if not tt_prop_to_param_names: 297 return 298 299 tt_prop_to_param_names = tt_prop_to_param_names.split(',') 300 301 testtracker_properties = {} 302 for entry in tt_prop_to_param_names: 303 prop_name, param_name = entry.split('=') 304 if param_name in test_instance.user_params: 305 testtracker_properties[prop_name] = ( 306 test_instance.user_params[param_name]) 307 308 if (hasattr(test_instance, 'android_devices') and 309 KEY_EFFORT_NAME not in testtracker_properties): 310 testtracker_properties[KEY_EFFORT_NAME] = ( 311 test_instance.android_devices[0].build_info['build_id']) 312 313 try: 314 writer = TestTrackerResultsWriter( 315 test_instance.log_path, testtracker_properties) 316 writer.write_results_from_test_signal( 317 signal, test_instance.begin_time) 318 except TestTrackerError: 319 test_instance.log.exception('TestTracker Error') 320 321 322class _TestInfoBinding(object): 323 """ 324 When Python creates an instance of an object it creates a binding object 325 for each closure that contains what the instance variable should be when 326 called. This object is a similar binding for _TestInfoDecoratorFunc. 327 When Python tries to create a binding of a _TestInfoDecoratorFunc it 328 will return one of these objects to hold the instance for that closure. 329 """ 330 331 def __init__(self, target, instance): 332 """ 333 Args: 334 target: The target for creating a binding to. 335 instance: The instance to bind the target with. 336 """ 337 self.target = target 338 self.instance = instance 339 self.__name__ = target.__name__ 340 341 def __call__(self, *args, **kwargs): 342 """ 343 When this object is called it will call the target with the bound 344 instance. 345 """ 346 return self.target(self.instance, *args, **kwargs) 347 348 def gather(self, *args, **kwargs): 349 """ 350 Will gather the target with the bound instance. 351 """ 352 return self.target.gather(self.instance, *args, **kwargs) 353