| #!/usr/bin/env python3.4 |
| # |
| # Copyright 2020 - 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 itertools |
| import logging |
| import os |
| from acts import asserts |
| from acts import base_test |
| from acts import utils |
| 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.test_decorators import test_tracker_info |
| from acts_contrib.test_utils.wifi import ota_sniffer |
| from acts_contrib.test_utils.wifi import wifi_retail_ap as retail_ap |
| 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 functools import partial |
| from WifiRvrTest import WifiRvrTest |
| |
| AccessPointTuple = collections.namedtuple(('AccessPointTuple'), |
| ['ap_settings']) |
| |
| |
| class WifiTdlsRvrTest(WifiRvrTest): |
| def __init__(self, controllers): |
| base_test.BaseTestClass.__init__(self, controllers) |
| 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 = [ |
| 'tdls_rvr_test_params', 'testbed_params', 'RetailAccessPoints' |
| ] |
| opt_params = ['ap_networks', 'OTASniffer'] |
| self.unpack_userparams(req_params, opt_params) |
| self.access_point = retail_ap.create(self.RetailAccessPoints)[0] |
| self.testclass_params = self.tdls_rvr_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] |
| |
| self.log_path = os.path.join(logging.log_path, 'results') |
| if hasattr(self, |
| 'OTASniffer') and self.testbed_params['sniffer_enable']: |
| self.sniffer = ota_sniffer.create(self.OTASniffer)[0] |
| 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 |
| if self.testclass_params.get('airplane_mode', 1): |
| self.log.info('Turning on airplane mode.') |
| for ad in self.android_devices: |
| asserts.assert_true(utils.force_airplane_mode(ad, True), |
| "Can not turn on airplane mode.") |
| for ad in self.android_devices: |
| wutils.wifi_toggle_state(ad, True) |
| |
| def teardown_class(self): |
| # Turn WiFi OFF |
| for dev in self.android_devices: |
| wutils.wifi_toggle_state(dev, False) |
| self.process_testclass_results() |
| # Teardown AP and release its lockfile |
| self.access_point.teardown() |
| |
| def setup_test(self): |
| for ad in self.android_devices: |
| wputils.start_wifi_logging(ad) |
| |
| def teardown_test(self): |
| self.iperf_server.stop() |
| for ad in self.android_devices: |
| wutils.reset_wifi(ad) |
| wputils.stop_wifi_logging(ad) |
| |
| def on_exception(self, test_name, begin_time): |
| for ad in self.android_devices: |
| ad.take_bug_report(test_name, begin_time) |
| ad.cat_adb_log(test_name, begin_time) |
| wutils.get_ssrdumps(ad) |
| |
| def compute_test_metrics(self, rvr_result): |
| #Set test metrics |
| rvr_result['metrics'] = {} |
| rvr_result['metrics']['peak_tput'] = max( |
| rvr_result['throughput_receive']) |
| if self.publish_testcase_metrics: |
| self.testcase_metric_logger.add_metric( |
| 'peak_tput', rvr_result['metrics']['peak_tput']) |
| |
| test_mode = rvr_result['testcase_params']['mode'] |
| tput_below_limit = [ |
| tput < |
| self.testclass_params['tput_metric_targets'][test_mode]['high'] |
| for tput in rvr_result['throughput_receive'] |
| ] |
| rvr_result['metrics']['high_tput_range'] = -1 |
| for idx in range(len(tput_below_limit)): |
| if all(tput_below_limit[idx:]): |
| if idx == 0: |
| #Throughput was never above limit |
| rvr_result['metrics']['high_tput_range'] = -1 |
| else: |
| rvr_result['metrics']['high_tput_range'] = rvr_result[ |
| 'total_attenuation'][max(idx, 1) - 1] |
| break |
| if self.publish_testcase_metrics: |
| self.testcase_metric_logger.add_metric( |
| 'high_tput_range', rvr_result['metrics']['high_tput_range']) |
| |
| tput_below_limit = [ |
| tput < |
| self.testclass_params['tput_metric_targets'][test_mode]['low'] |
| for tput in rvr_result['throughput_receive'] |
| ] |
| for idx in range(len(tput_below_limit)): |
| if all(tput_below_limit[idx:]): |
| rvr_result['metrics']['low_tput_range'] = rvr_result[ |
| 'total_attenuation'][max(idx, 1) - 1] |
| break |
| else: |
| rvr_result['metrics']['low_tput_range'] = -1 |
| if self.publish_testcase_metrics: |
| self.testcase_metric_logger.add_metric( |
| 'low_tput_range', rvr_result['metrics']['low_tput_range']) |
| |
| def setup_aps(self, testcase_params): |
| self.log.info('Setting AP to channel {} {}'.format( |
| testcase_params['channel'], testcase_params['bandwidth'])) |
| self.access_point.set_channel(testcase_params['interface_id'], |
| testcase_params['channel']) |
| self.access_point.set_bandwidth(testcase_params['interface_id'], |
| testcase_params['bandwidth']) |
| |
| def setup_duts(self, testcase_params): |
| # Check battery level before test |
| for ad in self.android_devices: |
| if not wputils.health_check(ad, 20): |
| asserts.skip('Overheating or Battery low. Skipping test.') |
| ad.go_to_sleep() |
| wutils.reset_wifi(ad) |
| # Turn screen off to preserve battery |
| for ad in self.android_devices: |
| wutils.wifi_connect( |
| ad, |
| self.ap_networks[0][testcase_params['interface_id']], |
| num_of_tries=5, |
| check_connectivity=True) |
| |
| def setup_tdls_connection(self, testcase_params): |
| |
| tdls_config = {} |
| for idx, ad in enumerate(self.android_devices): |
| tdls_config[idx] = { |
| 'ip_address': |
| ad.droid.connectivityGetIPv4Addresses('wlan0')[0], |
| 'mac_address': ad.droid.wifiGetConnectionInfo()['mac_address'], |
| 'tdls_supported': ad.droid.wifiIsTdlsSupported(), |
| 'off_channel_supported': |
| ad.droid.wifiIsOffChannelTdlsSupported() |
| } |
| self.android_devices[0].droid.wifiSetTdlsEnabledWithMacAddress( |
| tdls_config[1]['mac_address'], True) |
| |
| testcase_params['iperf_server_address'] = tdls_config[0]['ip_address'] |
| testcase_params['tdls_config'] = tdls_config |
| testcase_params['channel'] = testcase_params['channel'] |
| testcase_params['mode'] = testcase_params['bandwidth'] |
| testcase_params['test_network'] = self.ap_networks[0][ |
| testcase_params['interface_id']] |
| |
| def setup_tdls_rvr_test(self, testcase_params): |
| # Setup the aps |
| self.setup_aps(testcase_params) |
| # Setup the duts |
| self.setup_duts(testcase_params) |
| # Set attenuator to 0 dB |
| for attenuator in self.attenuators: |
| attenuator.set_atten(0, strict=False) |
| # Setup the aware connection |
| self.setup_tdls_connection(testcase_params) |
| # Set DUT to monitor RSSI and LLStats on |
| self.monitored_dut = self.android_devices[1] |
| |
| 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 |
| """ |
| for ad in self.android_devices: |
| wputils.check_skip_conditions(testcase_params, ad, |
| self.access_point) |
| |
| # Compile RvR 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) |
| ] |
| |
| # Compile iperf arguments |
| if testcase_params['traffic_type'] == 'TCP': |
| testcase_params['iperf_socket_size'] = self.testclass_params.get( |
| 'tcp_socket_size', None) |
| testcase_params['iperf_processes'] = self.testclass_params.get( |
| 'tcp_processes', 1) |
| elif testcase_params['traffic_type'] == 'UDP': |
| testcase_params['iperf_socket_size'] = self.testclass_params.get( |
| 'udp_socket_size', None) |
| testcase_params['iperf_processes'] = self.testclass_params.get( |
| 'udp_processes', 1) |
| testcase_params['iperf_args'] = wputils.get_iperf_arg_string( |
| duration=self.testclass_params['iperf_duration'], |
| reverse_direction=(testcase_params['traffic_direction'] == 'DL'), |
| socket_size=testcase_params['iperf_socket_size'], |
| num_processes=testcase_params['iperf_processes'], |
| traffic_type=testcase_params['traffic_type'], |
| ipv6=False) |
| testcase_params['use_client_output'] = ( |
| testcase_params['traffic_direction'] == 'DL') |
| |
| # Compile AP and infrastructure connection parameters |
| testcase_params['interface_id'] = '2G' if testcase_params[ |
| 'channel'] < 13 else '5G_1' |
| return testcase_params |
| |
| def _test_tdls_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) |
| |
| # Prepare devices and run test |
| self.setup_tdls_rvr_test(testcase_params) |
| rvr_result = self.run_rvr_test(testcase_params) |
| |
| # Post-process results |
| self.testclass_results.append(rvr_result) |
| self.process_test_results(rvr_result) |
| self.pass_fail_check(rvr_result) |
| |
| def generate_test_cases(self, ap_config_list, traffic_type, |
| traffic_directions): |
| """Function that auto-generates test cases for a test class.""" |
| test_cases = [] |
| |
| for ap_config, traffic_direction in itertools.product( |
| ap_config_list, traffic_directions): |
| test_name = 'test_tdls_rvr_{}_{}_ch{}_{}'.format( |
| traffic_type, traffic_direction, ap_config[0], ap_config[1]) |
| test_params = collections.OrderedDict( |
| traffic_type=traffic_type, |
| traffic_direction=traffic_direction, |
| channel=ap_config[0], |
| bandwidth=ap_config[1]) |
| test_class = self.__class__.__name__ |
| if "uuid_list" in self.user_params: |
| test_tracker_uuid = self.user_params["uuid_list"][ |
| test_class][test_name] |
| test_case = test_tracker_info(uuid=test_tracker_uuid)( |
| lambda: self._test_tdls_rvr(test_params)) |
| else: |
| test_case = partial(self._test_tdls_rvr,test_params) |
| setattr(self, test_name, test_case) |
| test_cases.append(test_name) |
| return test_cases |
| |
| |
| class WifiTdlsRvr_FCC_TCP_Test(WifiTdlsRvrTest): |
| def __init__(self, controllers): |
| super().__init__(controllers) |
| ap_config_list = [[6, 'bw20'], [36, 'bw20'], [36, 'bw40'], |
| [36, 'bw80'], [149, 'bw20'], [149, 'bw40'], |
| [149, 'bw80']] |
| self.country_code = 'US' |
| self.tests = self.generate_test_cases(ap_config_list=ap_config_list, |
| traffic_type='TCP', |
| traffic_directions=['DL', 'UL']) |
| |
| |
| class WifiTdlsRvr_FCC_UDP_Test(WifiTdlsRvrTest): |
| def __init__(self, controllers): |
| super().__init__(controllers) |
| ap_config_list = [[6, 'bw20'], [36, 'bw20'], [36, 'bw40'], |
| [36, 'bw80'], [149, 'bw20'], [149, 'bw40'], |
| [149, 'bw80']] |
| self.country_code = 'US' |
| self.tests = self.generate_test_cases(ap_config_list=ap_config_list, |
| traffic_type='UDP', |
| traffic_directions=['DL', 'UL']) |
| |
| |
| class WifiTdlsRvr_ETSI_TCP_Test(WifiTdlsRvrTest): |
| def __init__(self, controllers): |
| super().__init__(controllers) |
| ap_config_list = [[6, 'bw20'], [36, 'bw20'], [36, 'bw40'], |
| [36, 'bw80'], [149, 'bw20'], [149, 'bw40'], |
| [149, 'bw80']] |
| self.country_code = 'GB' |
| self.tests = self.generate_test_cases(ap_config_list=ap_config_list, |
| traffic_type='TCP', |
| traffic_directions=['DL', 'UL']) |
| |
| |
| class WifiTdlsRvr_ETSI_UDP_Test(WifiTdlsRvrTest): |
| def __init__(self, controllers): |
| super().__init__(controllers) |
| ap_config_list = [[6, 'bw20'], [36, 'bw20'], [36, 'bw40'], |
| [36, 'bw80'], [149, 'bw20'], [149, 'bw40'], |
| [149, 'bw80']] |
| self.country_code = 'GB' |
| self.tests = self.generate_test_cases(ap_config_list=ap_config_list, |
| traffic_type='UDP', |
| traffic_directions=['DL', 'UL']) |
