1#!/usr/bin/env python3.4 2# 3# Copyright 2018 - 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 logging 19import os 20from acts import asserts 21from acts import base_test 22from acts.controllers import iperf_server as ipf 23from acts.controllers import iperf_client as ipc 24from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 25from acts_contrib.test_utils.wifi import ota_sniffer 26from acts_contrib.test_utils.wifi import wifi_test_utils as wutils 27from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 28from acts_contrib.test_utils.wifi import wifi_retail_ap as retail_ap 29from WifiRvrTest import WifiRvrTest 30 31AccessPointTuple = collections.namedtuple(('AccessPointTuple'), 32 ['ap_settings']) 33 34 35class WifiSoftApRvrTest(WifiRvrTest): 36 def __init__(self, controllers): 37 base_test.BaseTestClass.__init__(self, controllers) 38 self.tests = ('test_rvr_TCP_DL_2GHz', 'test_rvr_TCP_UL_2GHz', 39 'test_rvr_TCP_DL_5GHz', 'test_rvr_TCP_UL_5GHz', 40 'test_rvr_TCP_DL_2GHz_backhaul_2GHz', 41 'test_rvr_TCP_UL_2GHz_backhaul_2GHz', 42 'test_rvr_TCP_DL_5GHz_backhaul_2GHz', 43 'test_rvr_TCP_UL_5GHz_backhaul_2GHz', 44 'test_rvr_TCP_DL_2GHz_backhaul_5GHz', 45 'test_rvr_TCP_UL_2GHz_backhaul_5GHz', 46 'test_rvr_TCP_DL_5GHz_backhaul_5GHz', 47 'test_rvr_TCP_UL_5GHz_backhaul_5GHz') 48 self.testcase_metric_logger = ( 49 BlackboxMappedMetricLogger.for_test_case()) 50 self.testclass_metric_logger = ( 51 BlackboxMappedMetricLogger.for_test_class()) 52 self.publish_testcase_metrics = True 53 54 def setup_class(self): 55 """Initializes common test hardware and parameters. 56 57 This function initializes hardwares and compiles parameters that are 58 common to all tests in this class. 59 """ 60 req_params = [ 61 'sap_test_params', 'testbed_params', 'RetailAccessPoints', 62 'ap_networks' 63 ] 64 opt_params = ['golden_files_list', 'OTASniffer'] 65 self.unpack_userparams(req_params, opt_params) 66 self.access_points = retail_ap.create(self.RetailAccessPoints) 67 self.testclass_params = self.sap_test_params 68 self.num_atten = self.attenuators[0].instrument.num_atten 69 self.iperf_server = ipf.create([{ 70 'AndroidDevice': 71 self.android_devices[0].serial, 72 'port': 73 '5201' 74 }])[0] 75 self.iperf_client = ipc.create([{ 76 'AndroidDevice': 77 self.android_devices[1].serial, 78 'port': 79 '5201' 80 }])[0] 81 if hasattr(self, 82 'OTASniffer') and self.testbed_params['sniffer_enable']: 83 try: 84 self.sniffer = ota_sniffer.create(self.OTASniffer)[0] 85 except: 86 self.log.warning('Could not start sniffer. Disabling sniffs.') 87 self.testbed_params['sniffer_enable'] = 0 88 89 self.log_path = os.path.join(logging.log_path, 'results') 90 os.makedirs(self.log_path, exist_ok=True) 91 if not hasattr(self, 'golden_files_list'): 92 if 'golden_results_path' in self.testbed_params: 93 self.golden_files_list = [ 94 os.path.join(self.testbed_params['golden_results_path'], 95 file) for file in 96 os.listdir(self.testbed_params['golden_results_path']) 97 ] 98 else: 99 self.log.warning('No golden files found.') 100 self.golden_files_list = [] 101 self.testclass_results = [] 102 103 # Turn WiFi ON 104 for dev in self.android_devices: 105 wutils.wifi_toggle_state(dev, True) 106 107 def teardown_class(self): 108 # Turn WiFi OFF 109 wutils.stop_wifi_tethering(self.android_devices[0]) 110 for dev in self.android_devices: 111 wutils.wifi_toggle_state(dev, False) 112 dev.go_to_sleep() 113 self.process_testclass_results() 114 # Teardown AP and release it's lockfile 115 for ap in self.access_points: 116 ap.teardown() 117 118 def teardown_test(self): 119 self.iperf_server.stop() 120 wutils.stop_wifi_tethering(self.android_devices[0]) 121 122 def get_sap_connection_info(self): 123 info = {} 124 info['client_ip_address'] = self.android_devices[ 125 1].droid.connectivityGetIPv4Addresses('wlan0')[0] 126 ifconfig_out = self.android_devices[0].adb.shell('ifconfig') 127 soft_ap_interface = 'wlan1' if 'wlan1' in ifconfig_out else 'wlan2' 128 info['ap_ip_address'] = self.android_devices[ 129 0].droid.connectivityGetIPv4Addresses(soft_ap_interface)[0] 130 131 connection_rssi = wputils.get_connected_rssi(self.android_devices[1], 132 interface='wlan0') 133 info['frequency'] = connection_rssi['frequency'][0] 134 info['channel'] = wutils.WifiEnums.freq_to_channel[int( 135 info['frequency'])] 136 info['mode'] = 'bw20' if info['channel'] < 13 else 'bw80' 137 return info 138 139 def setup_aps(self, testcase_params): 140 for network in testcase_params['ap_networks']: 141 self.log.info('Setting AP {} {} interface on channel {}'.format( 142 network['ap_id'], network['interface_id'], network['channel'])) 143 self.access_points[network['ap_id']].set_channel( 144 network['interface_id'], network['channel']) 145 146 def setup_duts(self, testcase_params): 147 """Function that gets devices ready for the test. 148 149 Args: 150 testcase_params: dict containing test-specific parameters 151 """ 152 self.ap_dut = self.android_devices[0] 153 self.sta_dut = self.android_devices[1] 154 for dev in self.android_devices: 155 if not wputils.health_check(dev, 20): 156 asserts.skip('DUT health check failed. Skipping test.') 157 # Reset WiFi on all devices 158 for dev in self.android_devices: 159 dev.go_to_sleep() 160 wutils.reset_wifi(dev) 161 wutils.set_wifi_country_code(dev, wutils.WifiEnums.CountryCode.US) 162 163 for network in testcase_params['ap_networks']: 164 for connected_dut in network['connected_dut']: 165 self.log.info("Connecting DUT {} to {}".format( 166 connected_dut, self.ap_networks[network['ap_id']][ 167 network['interface_id']])) 168 wutils.wifi_connect(self.android_devices[connected_dut], 169 self.ap_networks[network['ap_id']][ 170 network['interface_id']], 171 num_of_tries=5, 172 check_connectivity=True) 173 174 def setup_sap_connection(self, testcase_params): 175 # Setup Soft AP 176 sap_config = wutils.create_softap_config() 177 self.log.info('SoftAP Config: {}'.format(sap_config)) 178 wutils.start_wifi_tethering(self.android_devices[0], 179 sap_config[wutils.WifiEnums.SSID_KEY], 180 sap_config[wutils.WifiEnums.PWD_KEY], 181 testcase_params['sap_band_enum']) 182 # Connect DUT to Network 183 testcase_params['test_network'] = { 184 'SSID': sap_config[wutils.WifiEnums.SSID_KEY], 185 'password': sap_config[wutils.WifiEnums.PWD_KEY] 186 } 187 wutils.wifi_connect(self.sta_dut, 188 testcase_params['test_network'], 189 num_of_tries=5, 190 check_connectivity=False) 191 # Compile meta data 192 sap_info = self.get_sap_connection_info() 193 print("SAP Info: {}".format(sap_info)) 194 testcase_params['channel'] = sap_info['channel'] 195 if testcase_params['channel'] < 13: 196 testcase_params['band'] = '2GHz' 197 else: 198 testcase_params['band'] = '5GHz' 199 testcase_params['mode'] = sap_info['mode'] 200 self.access_point = AccessPointTuple({ 201 testcase_params['band']: { 202 'SSID': sap_config[wutils.WifiEnums.SSID_KEY], 203 'password': sap_config[wutils.WifiEnums.PWD_KEY], 204 'channel': sap_info['channel'], 205 'mode': sap_info['mode'], 206 'bandwidth': sap_info['mode'], 207 } 208 }) 209 testcase_params['iperf_server_address'] = sap_info['ap_ip_address'] 210 211 def setup_sap_rvr_test(self, testcase_params): 212 """Function that gets devices ready for the test. 213 214 Args: 215 testcase_params: dict containing test-specific parameters 216 """ 217 # Configure DUTs 218 self.setup_aps(testcase_params) 219 # Set attenuator to 0 dB 220 for attenuator in self.attenuators: 221 attenuator.set_atten(0, strict=False) 222 # Configure DUTs 223 self.setup_duts(testcase_params) 224 # Setup sap connection 225 self.setup_sap_connection(testcase_params) 226 # Set DUT to monitor RSSI and LLStats on 227 self.monitored_dut = self.sta_dut 228 self.monitored_interface = 'wlan0' 229 230 def compile_test_params(self, testcase_params): 231 """Function that completes all test params based on the test name. 232 233 Args: 234 testcase_params: dict containing test-specific parameters 235 """ 236 num_atten_steps = int((self.testclass_params['atten_stop'] - 237 self.testclass_params['atten_start']) / 238 self.testclass_params['atten_step']) 239 testcase_params['atten_range'] = [ 240 self.testclass_params['atten_start'] + 241 x * self.testclass_params['atten_step'] 242 for x in range(0, num_atten_steps) 243 ] 244 245 if testcase_params['traffic_direction'] == 'DL': 246 testcase_params['iperf_args'] = wputils.get_iperf_arg_string( 247 duration=self.testclass_params['iperf_duration'], 248 reverse_direction=1, 249 traffic_type=testcase_params['traffic_type']) 250 testcase_params['use_client_output'] = True 251 else: 252 testcase_params['iperf_args'] = wputils.get_iperf_arg_string( 253 duration=self.testclass_params['iperf_duration'], 254 reverse_direction=0, 255 traffic_type=testcase_params['traffic_type']) 256 testcase_params['use_client_output'] = False 257 258 # Compile AP and infrastructure connection parameters 259 ap_networks = [] 260 if testcase_params['dut_connected'][0]: 261 band = testcase_params['dut_connected'][0].split('_')[0] 262 ap_networks.append({ 263 'ap_id': 264 0, 265 'interface_id': 266 band if band == '2G' else band + '_1', 267 'band': 268 band, 269 'channel': 270 1 if band == '2G' else 36, 271 'connected_dut': [0] 272 }) 273 274 if testcase_params['dut_connected'][1]: 275 if testcase_params['dut_connected'][0] == testcase_params[ 276 'dut_connected'][1]: 277 # if connected to same network, add it to the above 278 ap_networks[0]['connected_dut'].append(1) 279 else: 280 band = testcase_params['dut_connected'][1].split('_')[0] 281 if not testcase_params['dut_connected'][0]: 282 # if it's the only dut connected, assign it to ap 0 283 ap_id = 0 284 else: 285 ap_id = 1 286 ap_networks.append({ 287 'ap_id': 288 ap_id, 289 'interface_id': 290 band if band == '2G' else band + '_1', 291 'band': 292 band, 293 'channel': 294 11 if band == '2G' else 149, 295 'connected_dut': [1] 296 }) 297 testcase_params['ap_networks'] = ap_networks 298 299 return testcase_params 300 301 def _test_sap_rvr(self, testcase_params): 302 """ Function that gets called for each test case 303 304 Args: 305 testcase_params: dict containing test-specific parameters 306 """ 307 # Compile test parameters from config and test name 308 testcase_params = self.compile_test_params(testcase_params) 309 310 self.setup_sap_rvr_test(testcase_params) 311 result = self.run_rvr_test(testcase_params) 312 self.testclass_results.append(result) 313 self.process_test_results(result) 314 self.pass_fail_check(result) 315 316 #Test cases 317 def test_rvr_TCP_DL_2GHz(self): 318 testcase_params = collections.OrderedDict( 319 sap_band='2GHz', 320 sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G, 321 traffic_type='TCP', 322 traffic_direction='DL', 323 dut_connected=[False, False]) 324 self._test_sap_rvr(testcase_params) 325 326 def test_rvr_TCP_UL_2GHz(self): 327 testcase_params = collections.OrderedDict( 328 sap_band='2GHz', 329 sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G, 330 traffic_type='TCP', 331 traffic_direction='UL', 332 dut_connected=[False, False]) 333 self._test_sap_rvr(testcase_params) 334 335 def test_rvr_TCP_DL_5GHz(self): 336 testcase_params = collections.OrderedDict( 337 sap_band='5GHz', 338 sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G, 339 traffic_type='TCP', 340 traffic_direction='DL', 341 dut_connected=[False, False]) 342 self._test_sap_rvr(testcase_params) 343 344 def test_rvr_TCP_UL_5GHz(self): 345 testcase_params = collections.OrderedDict( 346 sap_band='5GHz', 347 sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G, 348 traffic_type='TCP', 349 traffic_direction='UL', 350 dut_connected=[False, False]) 351 self._test_sap_rvr(testcase_params) 352 353 def test_rvr_TCP_DL_2GHz_backhaul_2GHz(self): 354 testcase_params = collections.OrderedDict( 355 sap_band='2GHz', 356 sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G, 357 traffic_type='TCP', 358 traffic_direction='DL', 359 dut_connected=['2G_1', False]) 360 self._test_sap_rvr(testcase_params) 361 362 def test_rvr_TCP_UL_2GHz_backhaul_2GHz(self): 363 testcase_params = collections.OrderedDict( 364 sap_band='2GHz', 365 sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G, 366 traffic_type='TCP', 367 traffic_direction='UL', 368 dut_connected=['2G_1', False]) 369 self._test_sap_rvr(testcase_params) 370 371 def test_rvr_TCP_DL_5GHz_backhaul_2GHz(self): 372 testcase_params = collections.OrderedDict( 373 sap_band='5GHz', 374 sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G, 375 traffic_type='TCP', 376 traffic_direction='DL', 377 dut_connected=['2G_1', False]) 378 self._test_sap_rvr(testcase_params) 379 380 def test_rvr_TCP_UL_5GHz_backhaul_2GHz(self): 381 testcase_params = collections.OrderedDict( 382 sap_band='5GHz', 383 sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G, 384 traffic_type='TCP', 385 traffic_direction='UL', 386 dut_connected=['2G_1', False]) 387 self._test_sap_rvr(testcase_params) 388 389 def test_rvr_TCP_DL_2GHz_backhaul_5GHz(self): 390 testcase_params = collections.OrderedDict( 391 sap_band='2GHz', 392 sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G, 393 traffic_type='TCP', 394 traffic_direction='DL', 395 dut_connected=['5G_1', False]) 396 self._test_sap_rvr(testcase_params) 397 398 def test_rvr_TCP_UL_2GHz_backhaul_5GHz(self): 399 testcase_params = collections.OrderedDict( 400 sap_band='2GHz', 401 sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G, 402 traffic_type='TCP', 403 traffic_direction='UL', 404 dut_connected=['5G_1', False]) 405 self._test_sap_rvr(testcase_params) 406 407 def test_rvr_TCP_DL_5GHz_backhaul_5GHz(self): 408 testcase_params = collections.OrderedDict( 409 sap_band='5GHz', 410 sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G, 411 traffic_type='TCP', 412 traffic_direction='DL', 413 dut_connected=['5G_1', False]) 414 self._test_sap_rvr(testcase_params) 415 416 def test_rvr_TCP_UL_5GHz_backhaul_5GHz(self): 417 testcase_params = collections.OrderedDict( 418 sap_band='5GHz', 419 sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G, 420 traffic_type='TCP', 421 traffic_direction='UL', 422 dut_connected=['5G_1', False]) 423 self._test_sap_rvr(testcase_params) 424