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