| #!/usr/bin/env python3.4 |
| # |
| # Copyright 2018 - 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 |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an 'AS IS' BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import collections |
| import logging |
| import os |
| from acts import asserts |
| from acts import base_test |
| from acts.controllers import iperf_server as ipf |
| from acts.controllers import iperf_client as ipc |
| from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger |
| from acts_contrib.test_utils.wifi import ota_sniffer |
| from acts_contrib.test_utils.wifi import wifi_test_utils as wutils |
| from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils |
| from acts_contrib.test_utils.wifi import wifi_retail_ap as retail_ap |
| from WifiRvrTest import WifiRvrTest |
| |
| AccessPointTuple = collections.namedtuple(('AccessPointTuple'), |
| ['ap_settings']) |
| |
| |
| class WifiSoftApRvrTest(WifiRvrTest): |
| def __init__(self, controllers): |
| base_test.BaseTestClass.__init__(self, controllers) |
| self.tests = ('test_rvr_TCP_DL_2GHz', 'test_rvr_TCP_UL_2GHz', |
| 'test_rvr_TCP_DL_5GHz', 'test_rvr_TCP_UL_5GHz', |
| 'test_rvr_TCP_DL_2GHz_backhaul_2GHz', |
| 'test_rvr_TCP_UL_2GHz_backhaul_2GHz', |
| 'test_rvr_TCP_DL_5GHz_backhaul_2GHz', |
| 'test_rvr_TCP_UL_5GHz_backhaul_2GHz', |
| 'test_rvr_TCP_DL_2GHz_backhaul_5GHz', |
| 'test_rvr_TCP_UL_2GHz_backhaul_5GHz', |
| 'test_rvr_TCP_DL_5GHz_backhaul_5GHz', |
| 'test_rvr_TCP_UL_5GHz_backhaul_5GHz') |
| self.testcase_metric_logger = ( |
| BlackboxMappedMetricLogger.for_test_case()) |
| self.testclass_metric_logger = ( |
| BlackboxMappedMetricLogger.for_test_class()) |
| self.publish_testcase_metrics = True |
| |
| def setup_class(self): |
| """Initializes common test hardware and parameters. |
| |
| This function initializes hardwares and compiles parameters that are |
| common to all tests in this class. |
| """ |
| req_params = [ |
| 'sap_test_params', 'testbed_params', 'RetailAccessPoints', |
| 'ap_networks' |
| ] |
| opt_params = ['golden_files_list', 'OTASniffer'] |
| self.unpack_userparams(req_params, opt_params) |
| self.access_points = retail_ap.create(self.RetailAccessPoints) |
| self.testclass_params = self.sap_test_params |
| self.num_atten = self.attenuators[0].instrument.num_atten |
| self.iperf_server = ipf.create([{ |
| 'AndroidDevice': |
| self.android_devices[0].serial, |
| 'port': |
| '5201' |
| }])[0] |
| self.iperf_client = ipc.create([{ |
| 'AndroidDevice': |
| self.android_devices[1].serial, |
| 'port': |
| '5201' |
| }])[0] |
| if hasattr(self, |
| 'OTASniffer') and self.testbed_params['sniffer_enable']: |
| try: |
| self.sniffer = ota_sniffer.create(self.OTASniffer)[0] |
| except: |
| self.log.warning('Could not start sniffer. Disabling sniffs.') |
| self.testbed_params['sniffer_enable'] = 0 |
| |
| self.log_path = os.path.join(logging.log_path, 'results') |
| os.makedirs(self.log_path, exist_ok=True) |
| if not hasattr(self, 'golden_files_list'): |
| if 'golden_results_path' in self.testbed_params: |
| self.golden_files_list = [ |
| os.path.join(self.testbed_params['golden_results_path'], |
| file) for file in |
| os.listdir(self.testbed_params['golden_results_path']) |
| ] |
| else: |
| self.log.warning('No golden files found.') |
| self.golden_files_list = [] |
| self.testclass_results = [] |
| |
| # Turn WiFi ON |
| for dev in self.android_devices: |
| wutils.wifi_toggle_state(dev, True) |
| |
| def teardown_class(self): |
| # Turn WiFi OFF |
| wutils.stop_wifi_tethering(self.android_devices[0]) |
| for dev in self.android_devices: |
| wutils.wifi_toggle_state(dev, False) |
| dev.go_to_sleep() |
| self.process_testclass_results() |
| # Teardown AP and release it's lockfile |
| for ap in self.access_points: |
| ap.teardown() |
| |
| def teardown_test(self): |
| self.iperf_server.stop() |
| wutils.stop_wifi_tethering(self.android_devices[0]) |
| |
| def get_sap_connection_info(self): |
| info = {} |
| info['client_ip_address'] = self.android_devices[ |
| 1].droid.connectivityGetIPv4Addresses('wlan0')[0] |
| ifconfig_out = self.android_devices[0].adb.shell('ifconfig') |
| soft_ap_interface = 'wlan1' if 'wlan1' in ifconfig_out else 'wlan2' |
| info['ap_ip_address'] = self.android_devices[ |
| 0].droid.connectivityGetIPv4Addresses(soft_ap_interface)[0] |
| |
| connection_rssi = wputils.get_connected_rssi(self.android_devices[1], |
| interface='wlan0') |
| info['frequency'] = connection_rssi['frequency'][0] |
| info['channel'] = wutils.WifiEnums.freq_to_channel[int( |
| info['frequency'])] |
| info['mode'] = 'bw20' if info['channel'] < 13 else 'bw80' |
| return info |
| |
| def setup_aps(self, testcase_params): |
| for network in testcase_params['ap_networks']: |
| self.log.info('Setting AP {} {} interface on channel {}'.format( |
| network['ap_id'], network['interface_id'], network['channel'])) |
| self.access_points[network['ap_id']].set_channel( |
| network['interface_id'], network['channel']) |
| |
| def setup_duts(self, testcase_params): |
| """Function that gets devices ready for the test. |
| |
| Args: |
| testcase_params: dict containing test-specific parameters |
| """ |
| self.ap_dut = self.android_devices[0] |
| self.sta_dut = self.android_devices[1] |
| for dev in self.android_devices: |
| if not wputils.health_check(dev, 20): |
| asserts.skip('DUT health check failed. Skipping test.') |
| # Reset WiFi on all devices |
| for dev in self.android_devices: |
| dev.go_to_sleep() |
| wutils.reset_wifi(dev) |
| wutils.set_wifi_country_code(dev, wutils.WifiEnums.CountryCode.US) |
| |
| for network in testcase_params['ap_networks']: |
| for connected_dut in network['connected_dut']: |
| self.log.info("Connecting DUT {} to {}".format( |
| connected_dut, self.ap_networks[network['ap_id']][ |
| network['interface_id']])) |
| wutils.wifi_connect(self.android_devices[connected_dut], |
| self.ap_networks[network['ap_id']][ |
| network['interface_id']], |
| num_of_tries=5, |
| check_connectivity=True) |
| |
| def setup_sap_connection(self, testcase_params): |
| # Setup Soft AP |
| sap_config = wutils.create_softap_config() |
| self.log.info('SoftAP Config: {}'.format(sap_config)) |
| wutils.start_wifi_tethering(self.android_devices[0], |
| sap_config[wutils.WifiEnums.SSID_KEY], |
| sap_config[wutils.WifiEnums.PWD_KEY], |
| testcase_params['sap_band_enum']) |
| # Connect DUT to Network |
| testcase_params['test_network'] = { |
| 'SSID': sap_config[wutils.WifiEnums.SSID_KEY], |
| 'password': sap_config[wutils.WifiEnums.PWD_KEY] |
| } |
| wutils.wifi_connect(self.sta_dut, |
| testcase_params['test_network'], |
| num_of_tries=5, |
| check_connectivity=False) |
| # Compile meta data |
| sap_info = self.get_sap_connection_info() |
| print("SAP Info: {}".format(sap_info)) |
| testcase_params['channel'] = sap_info['channel'] |
| if testcase_params['channel'] < 13: |
| testcase_params['band'] = '2GHz' |
| else: |
| testcase_params['band'] = '5GHz' |
| testcase_params['mode'] = sap_info['mode'] |
| self.access_point = AccessPointTuple({ |
| testcase_params['band']: { |
| 'SSID': sap_config[wutils.WifiEnums.SSID_KEY], |
| 'password': sap_config[wutils.WifiEnums.PWD_KEY], |
| 'channel': sap_info['channel'], |
| 'mode': sap_info['mode'], |
| 'bandwidth': sap_info['mode'], |
| } |
| }) |
| testcase_params['iperf_server_address'] = sap_info['ap_ip_address'] |
| |
| def setup_sap_rvr_test(self, testcase_params): |
| """Function that gets devices ready for the test. |
| |
| Args: |
| testcase_params: dict containing test-specific parameters |
| """ |
| # Configure DUTs |
| self.setup_aps(testcase_params) |
| # Set attenuator to 0 dB |
| for attenuator in self.attenuators: |
| attenuator.set_atten(0, strict=False) |
| # Configure DUTs |
| self.setup_duts(testcase_params) |
| # Setup sap connection |
| self.setup_sap_connection(testcase_params) |
| # Set DUT to monitor RSSI and LLStats on |
| self.monitored_dut = self.sta_dut |
| self.monitored_interface = 'wlan0' |
| |
| def compile_test_params(self, testcase_params): |
| """Function that completes all test params based on the test name. |
| |
| Args: |
| testcase_params: dict containing test-specific parameters |
| """ |
| num_atten_steps = int((self.testclass_params['atten_stop'] - |
| self.testclass_params['atten_start']) / |
| self.testclass_params['atten_step']) |
| testcase_params['atten_range'] = [ |
| self.testclass_params['atten_start'] + |
| x * self.testclass_params['atten_step'] |
| for x in range(0, num_atten_steps) |
| ] |
| |
| if testcase_params['traffic_direction'] == 'DL': |
| testcase_params['iperf_args'] = wputils.get_iperf_arg_string( |
| duration=self.testclass_params['iperf_duration'], |
| reverse_direction=1, |
| traffic_type=testcase_params['traffic_type']) |
| testcase_params['use_client_output'] = True |
| else: |
| testcase_params['iperf_args'] = wputils.get_iperf_arg_string( |
| duration=self.testclass_params['iperf_duration'], |
| reverse_direction=0, |
| traffic_type=testcase_params['traffic_type']) |
| testcase_params['use_client_output'] = False |
| |
| # Compile AP and infrastructure connection parameters |
| ap_networks = [] |
| if testcase_params['dut_connected'][0]: |
| band = testcase_params['dut_connected'][0].split('_')[0] |
| ap_networks.append({ |
| 'ap_id': |
| 0, |
| 'interface_id': |
| band if band == '2G' else band + '_1', |
| 'band': |
| band, |
| 'channel': |
| 1 if band == '2G' else 36, |
| 'connected_dut': [0] |
| }) |
| |
| if testcase_params['dut_connected'][1]: |
| if testcase_params['dut_connected'][0] == testcase_params[ |
| 'dut_connected'][1]: |
| # if connected to same network, add it to the above |
| ap_networks[0]['connected_dut'].append(1) |
| else: |
| band = testcase_params['dut_connected'][1].split('_')[0] |
| if not testcase_params['dut_connected'][0]: |
| # if it's the only dut connected, assign it to ap 0 |
| ap_id = 0 |
| else: |
| ap_id = 1 |
| ap_networks.append({ |
| 'ap_id': |
| ap_id, |
| 'interface_id': |
| band if band == '2G' else band + '_1', |
| 'band': |
| band, |
| 'channel': |
| 11 if band == '2G' else 149, |
| 'connected_dut': [1] |
| }) |
| testcase_params['ap_networks'] = ap_networks |
| |
| return testcase_params |
| |
| def _test_sap_rvr(self, testcase_params): |
| """ Function that gets called for each test case |
| |
| Args: |
| testcase_params: dict containing test-specific parameters |
| """ |
| # Compile test parameters from config and test name |
| testcase_params = self.compile_test_params(testcase_params) |
| |
| self.setup_sap_rvr_test(testcase_params) |
| result = self.run_rvr_test(testcase_params) |
| self.testclass_results.append(result) |
| self.process_test_results(result) |
| self.pass_fail_check(result) |
| |
| #Test cases |
| def test_rvr_TCP_DL_2GHz(self): |
| testcase_params = collections.OrderedDict( |
| sap_band='2GHz', |
| sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G, |
| traffic_type='TCP', |
| traffic_direction='DL', |
| dut_connected=[False, False]) |
| self._test_sap_rvr(testcase_params) |
| |
| def test_rvr_TCP_UL_2GHz(self): |
| testcase_params = collections.OrderedDict( |
| sap_band='2GHz', |
| sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G, |
| traffic_type='TCP', |
| traffic_direction='UL', |
| dut_connected=[False, False]) |
| self._test_sap_rvr(testcase_params) |
| |
| def test_rvr_TCP_DL_5GHz(self): |
| testcase_params = collections.OrderedDict( |
| sap_band='5GHz', |
| sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G, |
| traffic_type='TCP', |
| traffic_direction='DL', |
| dut_connected=[False, False]) |
| self._test_sap_rvr(testcase_params) |
| |
| def test_rvr_TCP_UL_5GHz(self): |
| testcase_params = collections.OrderedDict( |
| sap_band='5GHz', |
| sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G, |
| traffic_type='TCP', |
| traffic_direction='UL', |
| dut_connected=[False, False]) |
| self._test_sap_rvr(testcase_params) |
| |
| def test_rvr_TCP_DL_2GHz_backhaul_2GHz(self): |
| testcase_params = collections.OrderedDict( |
| sap_band='2GHz', |
| sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G, |
| traffic_type='TCP', |
| traffic_direction='DL', |
| dut_connected=['2G_1', False]) |
| self._test_sap_rvr(testcase_params) |
| |
| def test_rvr_TCP_UL_2GHz_backhaul_2GHz(self): |
| testcase_params = collections.OrderedDict( |
| sap_band='2GHz', |
| sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G, |
| traffic_type='TCP', |
| traffic_direction='UL', |
| dut_connected=['2G_1', False]) |
| self._test_sap_rvr(testcase_params) |
| |
| def test_rvr_TCP_DL_5GHz_backhaul_2GHz(self): |
| testcase_params = collections.OrderedDict( |
| sap_band='5GHz', |
| sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G, |
| traffic_type='TCP', |
| traffic_direction='DL', |
| dut_connected=['2G_1', False]) |
| self._test_sap_rvr(testcase_params) |
| |
| def test_rvr_TCP_UL_5GHz_backhaul_2GHz(self): |
| testcase_params = collections.OrderedDict( |
| sap_band='5GHz', |
| sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G, |
| traffic_type='TCP', |
| traffic_direction='UL', |
| dut_connected=['2G_1', False]) |
| self._test_sap_rvr(testcase_params) |
| |
| def test_rvr_TCP_DL_2GHz_backhaul_5GHz(self): |
| testcase_params = collections.OrderedDict( |
| sap_band='2GHz', |
| sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G, |
| traffic_type='TCP', |
| traffic_direction='DL', |
| dut_connected=['5G_1', False]) |
| self._test_sap_rvr(testcase_params) |
| |
| def test_rvr_TCP_UL_2GHz_backhaul_5GHz(self): |
| testcase_params = collections.OrderedDict( |
| sap_band='2GHz', |
| sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_2G, |
| traffic_type='TCP', |
| traffic_direction='UL', |
| dut_connected=['5G_1', False]) |
| self._test_sap_rvr(testcase_params) |
| |
| def test_rvr_TCP_DL_5GHz_backhaul_5GHz(self): |
| testcase_params = collections.OrderedDict( |
| sap_band='5GHz', |
| sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G, |
| traffic_type='TCP', |
| traffic_direction='DL', |
| dut_connected=['5G_1', False]) |
| self._test_sap_rvr(testcase_params) |
| |
| def test_rvr_TCP_UL_5GHz_backhaul_5GHz(self): |
| testcase_params = collections.OrderedDict( |
| sap_band='5GHz', |
| sap_band_enum=wutils.WifiEnums.WIFI_CONFIG_APBAND_5G, |
| traffic_type='TCP', |
| traffic_direction='UL', |
| dut_connected=['5G_1', False]) |
| self._test_sap_rvr(testcase_params) |
