1#!/usr/bin/env python3.4 2# 3# Copyright 2020 - 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 collections 18import itertools 19import logging 20import os 21from acts import asserts 22from acts import base_test 23from acts import utils 24from acts.controllers import iperf_server as ipf 25from acts.controllers import iperf_client as ipc 26from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 27from acts.test_decorators import test_tracker_info 28from acts_contrib.test_utils.wifi import ota_sniffer 29from acts_contrib.test_utils.wifi import wifi_retail_ap as retail_ap 30from acts_contrib.test_utils.wifi import wifi_test_utils as wutils 31from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 32from functools import partial 33from WifiRvrTest import WifiRvrTest 34 35AccessPointTuple = collections.namedtuple(('AccessPointTuple'), 36 ['ap_settings']) 37 38 39class WifiTdlsRvrTest(WifiRvrTest): 40 def __init__(self, controllers): 41 base_test.BaseTestClass.__init__(self, controllers) 42 self.testcase_metric_logger = ( 43 BlackboxMappedMetricLogger.for_test_case()) 44 self.testclass_metric_logger = ( 45 BlackboxMappedMetricLogger.for_test_class()) 46 self.publish_testcase_metrics = True 47 48 def setup_class(self): 49 """Initializes common test hardware and parameters. 50 51 This function initializes hardwares and compiles parameters that are 52 common to all tests in this class. 53 """ 54 req_params = [ 55 'tdls_rvr_test_params', 'testbed_params', 'RetailAccessPoints' 56 ] 57 opt_params = ['ap_networks', 'OTASniffer'] 58 self.unpack_userparams(req_params, opt_params) 59 self.access_point = retail_ap.create(self.RetailAccessPoints)[0] 60 self.testclass_params = self.tdls_rvr_test_params 61 self.num_atten = self.attenuators[0].instrument.num_atten 62 self.iperf_server = ipf.create([{ 63 'AndroidDevice': 64 self.android_devices[0].serial, 65 'port': 66 '5201' 67 }])[0] 68 self.iperf_client = ipc.create([{ 69 'AndroidDevice': 70 self.android_devices[1].serial, 71 'port': 72 '5201' 73 }])[0] 74 75 self.log_path = os.path.join(logging.log_path, 'results') 76 if hasattr(self, 77 'OTASniffer') and self.testbed_params['sniffer_enable']: 78 self.sniffer = ota_sniffer.create(self.OTASniffer)[0] 79 os.makedirs(self.log_path, exist_ok=True) 80 if not hasattr(self, 'golden_files_list'): 81 if 'golden_results_path' in self.testbed_params: 82 self.golden_files_list = [ 83 os.path.join(self.testbed_params['golden_results_path'], 84 file) for file in 85 os.listdir(self.testbed_params['golden_results_path']) 86 ] 87 else: 88 self.log.warning('No golden files found.') 89 self.golden_files_list = [] 90 91 self.testclass_results = [] 92 93 # Turn WiFi ON 94 if self.testclass_params.get('airplane_mode', 1): 95 self.log.info('Turning on airplane mode.') 96 for ad in self.android_devices: 97 asserts.assert_true(utils.force_airplane_mode(ad, True), 98 "Can not turn on airplane mode.") 99 for ad in self.android_devices: 100 wutils.wifi_toggle_state(ad, True) 101 102 def teardown_class(self): 103 # Turn WiFi OFF 104 for dev in self.android_devices: 105 wutils.wifi_toggle_state(dev, False) 106 self.process_testclass_results() 107 # Teardown AP and release its lockfile 108 self.access_point.teardown() 109 110 def setup_test(self): 111 for ad in self.android_devices: 112 wputils.start_wifi_logging(ad) 113 114 def teardown_test(self): 115 self.iperf_server.stop() 116 for ad in self.android_devices: 117 wutils.reset_wifi(ad) 118 wputils.stop_wifi_logging(ad) 119 120 def on_exception(self, test_name, begin_time): 121 for ad in self.android_devices: 122 ad.take_bug_report(test_name, begin_time) 123 ad.cat_adb_log(test_name, begin_time) 124 wutils.get_ssrdumps(ad) 125 126 def compute_test_metrics(self, rvr_result): 127 #Set test metrics 128 rvr_result['metrics'] = {} 129 rvr_result['metrics']['peak_tput'] = max( 130 rvr_result['throughput_receive']) 131 if self.publish_testcase_metrics: 132 self.testcase_metric_logger.add_metric( 133 'peak_tput', rvr_result['metrics']['peak_tput']) 134 135 test_mode = rvr_result['testcase_params']['mode'] 136 tput_below_limit = [ 137 tput < 138 self.testclass_params['tput_metric_targets'][test_mode]['high'] 139 for tput in rvr_result['throughput_receive'] 140 ] 141 rvr_result['metrics']['high_tput_range'] = -1 142 for idx in range(len(tput_below_limit)): 143 if all(tput_below_limit[idx:]): 144 if idx == 0: 145 #Throughput was never above limit 146 rvr_result['metrics']['high_tput_range'] = -1 147 else: 148 rvr_result['metrics']['high_tput_range'] = rvr_result[ 149 'total_attenuation'][max(idx, 1) - 1] 150 break 151 if self.publish_testcase_metrics: 152 self.testcase_metric_logger.add_metric( 153 'high_tput_range', rvr_result['metrics']['high_tput_range']) 154 155 tput_below_limit = [ 156 tput < 157 self.testclass_params['tput_metric_targets'][test_mode]['low'] 158 for tput in rvr_result['throughput_receive'] 159 ] 160 for idx in range(len(tput_below_limit)): 161 if all(tput_below_limit[idx:]): 162 rvr_result['metrics']['low_tput_range'] = rvr_result[ 163 'total_attenuation'][max(idx, 1) - 1] 164 break 165 else: 166 rvr_result['metrics']['low_tput_range'] = -1 167 if self.publish_testcase_metrics: 168 self.testcase_metric_logger.add_metric( 169 'low_tput_range', rvr_result['metrics']['low_tput_range']) 170 171 def setup_aps(self, testcase_params): 172 self.log.info('Setting AP to channel {} {}'.format( 173 testcase_params['channel'], testcase_params['bandwidth'])) 174 self.access_point.set_channel(testcase_params['interface_id'], 175 testcase_params['channel']) 176 self.access_point.set_bandwidth(testcase_params['interface_id'], 177 testcase_params['bandwidth']) 178 179 def setup_duts(self, testcase_params): 180 # Check battery level before test 181 for ad in self.android_devices: 182 if not wputils.health_check(ad, 20): 183 asserts.skip('Overheating or Battery low. Skipping test.') 184 ad.go_to_sleep() 185 wutils.reset_wifi(ad) 186 # Turn screen off to preserve battery 187 for ad in self.android_devices: 188 wutils.wifi_connect( 189 ad, 190 self.ap_networks[0][testcase_params['interface_id']], 191 num_of_tries=5, 192 check_connectivity=True) 193 194 def setup_tdls_connection(self, testcase_params): 195 196 tdls_config = {} 197 for idx, ad in enumerate(self.android_devices): 198 tdls_config[idx] = { 199 'ip_address': 200 ad.droid.connectivityGetIPv4Addresses('wlan0')[0], 201 'mac_address': ad.droid.wifiGetConnectionInfo()['mac_address'], 202 'tdls_supported': ad.droid.wifiIsTdlsSupported(), 203 'off_channel_supported': 204 ad.droid.wifiIsOffChannelTdlsSupported() 205 } 206 self.android_devices[0].droid.wifiSetTdlsEnabledWithMacAddress( 207 tdls_config[1]['mac_address'], True) 208 209 testcase_params['iperf_server_address'] = tdls_config[0]['ip_address'] 210 testcase_params['tdls_config'] = tdls_config 211 testcase_params['channel'] = testcase_params['channel'] 212 testcase_params['mode'] = testcase_params['bandwidth'] 213 testcase_params['test_network'] = self.ap_networks[0][ 214 testcase_params['interface_id']] 215 216 def setup_tdls_rvr_test(self, testcase_params): 217 # Setup the aps 218 self.setup_aps(testcase_params) 219 # Setup the duts 220 self.setup_duts(testcase_params) 221 # Set attenuator to 0 dB 222 for attenuator in self.attenuators: 223 attenuator.set_atten(0, strict=False) 224 # Setup the aware connection 225 self.setup_tdls_connection(testcase_params) 226 # Set DUT to monitor RSSI and LLStats on 227 self.monitored_dut = self.android_devices[1] 228 229 def compile_test_params(self, testcase_params): 230 """Function that completes all test params based on the test name. 231 232 Args: 233 testcase_params: dict containing test-specific parameters 234 """ 235 for ad in self.android_devices: 236 wputils.check_skip_conditions(testcase_params, ad, 237 self.access_point) 238 239 # Compile RvR parameters 240 num_atten_steps = int((self.testclass_params['atten_stop'] - 241 self.testclass_params['atten_start']) / 242 self.testclass_params['atten_step']) 243 testcase_params['atten_range'] = [ 244 self.testclass_params['atten_start'] + 245 x * self.testclass_params['atten_step'] 246 for x in range(0, num_atten_steps) 247 ] 248 249 # Compile iperf arguments 250 if testcase_params['traffic_type'] == 'TCP': 251 testcase_params['iperf_socket_size'] = self.testclass_params.get( 252 'tcp_socket_size', None) 253 testcase_params['iperf_processes'] = self.testclass_params.get( 254 'tcp_processes', 1) 255 elif testcase_params['traffic_type'] == 'UDP': 256 testcase_params['iperf_socket_size'] = self.testclass_params.get( 257 'udp_socket_size', None) 258 testcase_params['iperf_processes'] = self.testclass_params.get( 259 'udp_processes', 1) 260 testcase_params['iperf_args'] = wputils.get_iperf_arg_string( 261 duration=self.testclass_params['iperf_duration'], 262 reverse_direction=(testcase_params['traffic_direction'] == 'DL'), 263 socket_size=testcase_params['iperf_socket_size'], 264 num_processes=testcase_params['iperf_processes'], 265 traffic_type=testcase_params['traffic_type'], 266 ipv6=False) 267 testcase_params['use_client_output'] = ( 268 testcase_params['traffic_direction'] == 'DL') 269 270 # Compile AP and infrastructure connection parameters 271 testcase_params['interface_id'] = '2G' if testcase_params[ 272 'channel'] < 13 else '5G_1' 273 return testcase_params 274 275 def _test_tdls_rvr(self, testcase_params): 276 """ Function that gets called for each test case 277 278 Args: 279 testcase_params: dict containing test-specific parameters 280 """ 281 # Compile test parameters from config and test name 282 testcase_params = self.compile_test_params(testcase_params) 283 284 # Prepare devices and run test 285 self.setup_tdls_rvr_test(testcase_params) 286 rvr_result = self.run_rvr_test(testcase_params) 287 288 # Post-process results 289 self.testclass_results.append(rvr_result) 290 self.process_test_results(rvr_result) 291 self.pass_fail_check(rvr_result) 292 293 def generate_test_cases(self, ap_config_list, traffic_type, 294 traffic_directions): 295 """Function that auto-generates test cases for a test class.""" 296 test_cases = [] 297 298 for ap_config, traffic_direction in itertools.product( 299 ap_config_list, traffic_directions): 300 test_name = 'test_tdls_rvr_{}_{}_ch{}_{}'.format( 301 traffic_type, traffic_direction, ap_config[0], ap_config[1]) 302 test_params = collections.OrderedDict( 303 traffic_type=traffic_type, 304 traffic_direction=traffic_direction, 305 channel=ap_config[0], 306 bandwidth=ap_config[1]) 307 test_class = self.__class__.__name__ 308 if "uuid_list" in self.user_params: 309 test_tracker_uuid = self.user_params["uuid_list"][ 310 test_class][test_name] 311 test_case = test_tracker_info(uuid=test_tracker_uuid)( 312 lambda: self._test_tdls_rvr(test_params)) 313 else: 314 test_case = partial(self._test_tdls_rvr,test_params) 315 setattr(self, test_name, test_case) 316 test_cases.append(test_name) 317 return test_cases 318 319 320class WifiTdlsRvr_FCC_TCP_Test(WifiTdlsRvrTest): 321 def __init__(self, controllers): 322 super().__init__(controllers) 323 ap_config_list = [[6, 'bw20'], [36, 'bw20'], [36, 'bw40'], 324 [36, 'bw80'], [149, 'bw20'], [149, 'bw40'], 325 [149, 'bw80']] 326 self.country_code = 'US' 327 self.tests = self.generate_test_cases(ap_config_list=ap_config_list, 328 traffic_type='TCP', 329 traffic_directions=['DL', 'UL']) 330 331 332class WifiTdlsRvr_FCC_UDP_Test(WifiTdlsRvrTest): 333 def __init__(self, controllers): 334 super().__init__(controllers) 335 ap_config_list = [[6, 'bw20'], [36, 'bw20'], [36, 'bw40'], 336 [36, 'bw80'], [149, 'bw20'], [149, 'bw40'], 337 [149, 'bw80']] 338 self.country_code = 'US' 339 self.tests = self.generate_test_cases(ap_config_list=ap_config_list, 340 traffic_type='UDP', 341 traffic_directions=['DL', 'UL']) 342 343 344class WifiTdlsRvr_ETSI_TCP_Test(WifiTdlsRvrTest): 345 def __init__(self, controllers): 346 super().__init__(controllers) 347 ap_config_list = [[6, 'bw20'], [36, 'bw20'], [36, 'bw40'], 348 [36, 'bw80'], [149, 'bw20'], [149, 'bw40'], 349 [149, 'bw80']] 350 self.country_code = 'GB' 351 self.tests = self.generate_test_cases(ap_config_list=ap_config_list, 352 traffic_type='TCP', 353 traffic_directions=['DL', 'UL']) 354 355 356class WifiTdlsRvr_ETSI_UDP_Test(WifiTdlsRvrTest): 357 def __init__(self, controllers): 358 super().__init__(controllers) 359 ap_config_list = [[6, 'bw20'], [36, 'bw20'], [36, 'bw40'], 360 [36, 'bw80'], [149, 'bw20'], [149, 'bw40'], 361 [149, 'bw80']] 362 self.country_code = 'GB' 363 self.tests = self.generate_test_cases(ap_config_list=ap_config_list, 364 traffic_type='UDP', 365 traffic_directions=['DL', 'UL']) 366