| #!/usr/bin/env python3.4 |
| # |
| # Copyright 2017 - 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 json |
| import logging |
| import os |
| import statistics |
| from acts import asserts |
| from acts import context |
| from acts import base_test |
| from acts import utils |
| from acts.controllers.utils_lib import ssh |
| from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger |
| from acts_contrib.test_utils.wifi import ota_chamber |
| from acts_contrib.test_utils.wifi import ota_sniffer |
| from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils |
| from acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure |
| 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 functools import partial |
| |
| |
| class WifiPingTest(base_test.BaseTestClass): |
| """Class for ping-based Wifi performance tests. |
| |
| This class implements WiFi ping performance tests such as range and RTT. |
| The class setups up the AP in the desired configurations, configures |
| and connects the phone to the AP, and runs For an example config file to |
| run this test class see example_connectivity_performance_ap_sta.json. |
| """ |
| |
| TEST_TIMEOUT = 10 |
| RSSI_POLL_INTERVAL = 0.2 |
| SHORT_SLEEP = 1 |
| MED_SLEEP = 5 |
| MAX_CONSECUTIVE_ZEROS = 5 |
| DISCONNECTED_PING_RESULT = { |
| 'connected': 0, |
| 'rtt': [], |
| 'time_stamp': [], |
| 'ping_interarrivals': [], |
| 'packet_loss_percentage': 100 |
| } |
| |
| 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): |
| self.dut = self.android_devices[-1] |
| req_params = [ |
| 'ping_test_params', 'testbed_params', 'main_network', |
| 'RetailAccessPoints', 'RemoteServer' |
| ] |
| opt_params = ['OTASniffer'] |
| self.unpack_userparams(req_params, opt_params) |
| self.testclass_params = self.ping_test_params |
| self.num_atten = self.attenuators[0].instrument.num_atten |
| self.ping_server = ssh.connection.SshConnection( |
| ssh.settings.from_config(self.RemoteServer[0]['ssh_config'])) |
| self.access_point = retail_ap.create(self.RetailAccessPoints)[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.info('Access Point Configuration: {}'.format( |
| self.access_point.ap_settings)) |
| self.log_path = os.path.join(logging.log_path, 'results') |
| os.makedirs(self.log_path, exist_ok=True) |
| self.atten_dut_chain_map = {} |
| self.testclass_results = [] |
| |
| # Turn WiFi ON |
| if self.testclass_params.get('airplane_mode', 1): |
| self.log.info('Turning on airplane mode.') |
| asserts.assert_true(utils.force_airplane_mode(self.dut, True), |
| 'Can not turn on airplane mode.') |
| wutils.wifi_toggle_state(self.dut, True) |
| |
| # Configure test retries |
| self.user_params['retry_tests'] = [self.__class__.__name__] |
| |
| def teardown_class(self): |
| for attenuator in self.attenuators: |
| attenuator.set_atten(0, strict=False, retry=True) |
| # Turn WiFi OFF and reset AP |
| self.access_point.teardown() |
| for dev in self.android_devices: |
| wutils.wifi_toggle_state(dev, False) |
| dev.go_to_sleep() |
| self.process_testclass_results() |
| |
| def setup_test(self): |
| self.retry_flag = False |
| |
| def teardown_test(self): |
| self.retry_flag = False |
| |
| def on_retry(self): |
| """Function to control test logic on retried tests. |
| |
| This function is automatically executed on tests that are being |
| retried. In this case the function resets wifi, toggles it off and on |
| and sets a retry_flag to enable further tweaking the test logic on |
| second attempts. |
| """ |
| self.retry_flag = True |
| for dev in self.android_devices: |
| wutils.reset_wifi(dev) |
| wutils.toggle_wifi_off_and_on(dev) |
| |
| def process_testclass_results(self): |
| """Saves all test results to enable comparison.""" |
| testclass_summary = {} |
| for test in self.testclass_results: |
| if 'range' in test['test_name']: |
| testclass_summary[test['test_name']] = test['range'] |
| # Save results |
| results_file_path = os.path.join(self.log_path, |
| 'testclass_summary.json') |
| with open(results_file_path, 'w') as results_file: |
| json.dump(wputils.serialize_dict(testclass_summary), |
| results_file, |
| indent=4) |
| |
| def pass_fail_check_ping_rtt(self, result): |
| """Check the test result and decide if it passed or failed. |
| |
| The function computes RTT statistics and fails any tests in which the |
| tail of the ping latency results exceeds the threshold defined in the |
| configuration file. |
| |
| Args: |
| result: dict containing ping results and other meta data |
| """ |
| ignored_fraction = (self.testclass_params['rtt_ignored_interval'] / |
| self.testclass_params['rtt_ping_duration']) |
| sorted_rtt = [ |
| sorted(x['rtt'][round(ignored_fraction * len(x['rtt'])):]) |
| for x in result['ping_results'] |
| ] |
| disconnected = any([len(x) == 0 for x in sorted_rtt]) |
| if disconnected: |
| asserts.fail('Test failed. DUT disconnected at least once.') |
| |
| rtt_at_test_percentile = [ |
| x[int((1 - self.testclass_params['rtt_test_percentile'] / 100) * |
| len(x))] for x in sorted_rtt |
| ] |
| # Set blackbox metric |
| if self.publish_testcase_metrics: |
| self.testcase_metric_logger.add_metric('ping_rtt', |
| max(rtt_at_test_percentile)) |
| # Evaluate test pass/fail |
| rtt_failed = any([ |
| rtt > self.testclass_params['rtt_threshold'] * 1000 |
| for rtt in rtt_at_test_percentile |
| ]) |
| if rtt_failed: |
| #TODO: figure out how to cleanly exclude RTT tests from retry |
| asserts.explicit_pass( |
| 'Test failed. RTTs at test percentile = {}'.format( |
| rtt_at_test_percentile)) |
| else: |
| asserts.explicit_pass( |
| 'Test Passed. RTTs at test percentile = {}'.format( |
| rtt_at_test_percentile)) |
| |
| def pass_fail_check_ping_range(self, result): |
| """Check the test result and decide if it passed or failed. |
| |
| Checks whether the attenuation at which ping packet losses begin to |
| exceed the threshold matches the range derived from golden |
| rate-vs-range result files. The test fails is ping range is |
| range_gap_threshold worse than RvR range. |
| |
| Args: |
| result: dict containing ping results and meta data |
| """ |
| # Evaluate test pass/fail |
| test_message = ('Attenuation at range is {}dB. ' |
| 'LLStats at Range: {}'.format( |
| result['range'], result['llstats_at_range'])) |
| if result['peak_throughput_pct'] < 95: |
| asserts.fail('(RESULT NOT RELIABLE) {}'.format(test_message)) |
| |
| # If pass, set Blackbox metric |
| if self.publish_testcase_metrics: |
| self.testcase_metric_logger.add_metric('ping_range', |
| result['range']) |
| asserts.explicit_pass(test_message) |
| |
| def pass_fail_check(self, result): |
| if 'range' in result['testcase_params']['test_type']: |
| self.pass_fail_check_ping_range(result) |
| else: |
| self.pass_fail_check_ping_rtt(result) |
| |
| def process_ping_results(self, testcase_params, ping_range_result): |
| """Saves and plots ping results. |
| |
| Args: |
| ping_range_result: dict containing ping results and metadata |
| """ |
| # Compute range |
| ping_loss_over_att = [ |
| x['packet_loss_percentage'] |
| for x in ping_range_result['ping_results'] |
| ] |
| ping_loss_above_threshold = [ |
| x > self.testclass_params['range_ping_loss_threshold'] |
| for x in ping_loss_over_att |
| ] |
| for idx in range(len(ping_loss_above_threshold)): |
| if all(ping_loss_above_threshold[idx:]): |
| range_index = max(idx, 1) - 1 |
| break |
| else: |
| range_index = -1 |
| ping_range_result['atten_at_range'] = testcase_params['atten_range'][ |
| range_index] |
| ping_range_result['peak_throughput_pct'] = 100 - min( |
| ping_loss_over_att) |
| ping_range_result['total_attenuation'] = [ |
| ping_range_result['fixed_attenuation'] + att |
| for att in testcase_params['atten_range'] |
| ] |
| ping_range_result['range'] = (ping_range_result['atten_at_range'] + |
| ping_range_result['fixed_attenuation']) |
| ping_range_result['llstats_at_range'] = ( |
| 'TX MCS = {0} ({1:.1f}%). ' |
| 'RX MCS = {2} ({3:.1f}%)'.format( |
| ping_range_result['llstats'][range_index]['summary'] |
| ['common_tx_mcs'], ping_range_result['llstats'][range_index] |
| ['summary']['common_tx_mcs_freq'] * 100, |
| ping_range_result['llstats'][range_index]['summary'] |
| ['common_rx_mcs'], ping_range_result['llstats'][range_index] |
| ['summary']['common_rx_mcs_freq'] * 100)) |
| |
| # Save results |
| results_file_path = os.path.join( |
| self.log_path, '{}.json'.format(self.current_test_name)) |
| with open(results_file_path, 'w') as results_file: |
| json.dump(wputils.serialize_dict(ping_range_result), |
| results_file, |
| indent=4) |
| |
| # Plot results |
| if 'rtt' in self.current_test_name: |
| figure = BokehFigure(self.current_test_name, |
| x_label='Timestamp (s)', |
| primary_y_label='Round Trip Time (ms)') |
| for idx, result in enumerate(ping_range_result['ping_results']): |
| if len(result['rtt']) > 1: |
| x_data = [ |
| t - result['time_stamp'][0] |
| for t in result['time_stamp'] |
| ] |
| figure.add_line( |
| x_data, result['rtt'], 'RTT @ {}dB'.format( |
| ping_range_result['attenuation'][idx])) |
| |
| output_file_path = os.path.join( |
| self.log_path, '{}.html'.format(self.current_test_name)) |
| figure.generate_figure(output_file_path) |
| |
| def run_ping_test(self, testcase_params): |
| """Main function to test ping. |
| |
| The function sets up the AP in the correct channel and mode |
| configuration and calls get_ping_stats while sweeping attenuation |
| |
| Args: |
| testcase_params: dict containing all test parameters |
| Returns: |
| test_result: dict containing ping results and other meta data |
| """ |
| # Prepare results dict |
| llstats_obj = wputils.LinkLayerStats( |
| self.dut, self.testclass_params.get('llstats_enabled', True)) |
| test_result = collections.OrderedDict() |
| test_result['testcase_params'] = testcase_params.copy() |
| test_result['test_name'] = self.current_test_name |
| test_result['ap_config'] = self.access_point.ap_settings.copy() |
| test_result['attenuation'] = testcase_params['atten_range'] |
| test_result['fixed_attenuation'] = self.testbed_params[ |
| 'fixed_attenuation'][str(testcase_params['channel'])] |
| test_result['rssi_results'] = [] |
| test_result['ping_results'] = [] |
| test_result['llstats'] = [] |
| # Setup sniffer |
| if self.testbed_params['sniffer_enable']: |
| self.sniffer.start_capture( |
| testcase_params['test_network'], |
| chan=testcase_params['channel'], |
| bw=testcase_params['bandwidth'], |
| duration=testcase_params['ping_duration'] * |
| len(testcase_params['atten_range']) + self.TEST_TIMEOUT) |
| # Run ping and sweep attenuation as needed |
| zero_counter = 0 |
| pending_first_ping = 1 |
| for atten in testcase_params['atten_range']: |
| for attenuator in self.attenuators: |
| attenuator.set_atten(atten, strict=False, retry=True) |
| if self.testclass_params.get('monitor_rssi', 1): |
| rssi_future = wputils.get_connected_rssi_nb( |
| self.dut, |
| int(testcase_params['ping_duration'] / 2 / |
| self.RSSI_POLL_INTERVAL), self.RSSI_POLL_INTERVAL, |
| testcase_params['ping_duration'] / 2) |
| # Refresh link layer stats |
| llstats_obj.update_stats() |
| if testcase_params.get('ping_from_dut', False): |
| current_ping_stats = wputils.get_ping_stats( |
| self.dut, |
| wputils.get_server_address(self.ping_server, self.dut_ip, |
| '255.255.255.0'), |
| testcase_params['ping_duration'], |
| testcase_params['ping_interval'], |
| testcase_params['ping_size']) |
| else: |
| current_ping_stats = wputils.get_ping_stats( |
| self.ping_server, self.dut_ip, |
| testcase_params['ping_duration'], |
| testcase_params['ping_interval'], |
| testcase_params['ping_size']) |
| if self.testclass_params.get('monitor_rssi', 1): |
| current_rssi = rssi_future.result() |
| else: |
| current_rssi = collections.OrderedDict([ |
| ('time_stamp', []), ('bssid', []), ('ssid', []), |
| ('frequency', []), |
| ('signal_poll_rssi', wputils.empty_rssi_result()), |
| ('signal_poll_avg_rssi', wputils.empty_rssi_result()), |
| ('chain_0_rssi', wputils.empty_rssi_result()), |
| ('chain_1_rssi', wputils.empty_rssi_result()) |
| ]) |
| test_result['rssi_results'].append(current_rssi) |
| llstats_obj.update_stats() |
| curr_llstats = llstats_obj.llstats_incremental.copy() |
| test_result['llstats'].append(curr_llstats) |
| if current_ping_stats['connected']: |
| llstats_str = 'TX MCS = {0} ({1:.1f}%). RX MCS = {2} ({3:.1f}%)'.format( |
| curr_llstats['summary']['common_tx_mcs'], |
| curr_llstats['summary']['common_tx_mcs_freq'] * 100, |
| curr_llstats['summary']['common_rx_mcs'], |
| curr_llstats['summary']['common_rx_mcs_freq'] * 100) |
| self.log.info( |
| 'Attenuation = {0}dB\tPacket Loss = {1:.1f}%\t' |
| 'Avg RTT = {2:.2f}ms\tRSSI = {3:.1f} [{4:.1f},{5:.1f}]\t{6}\t' |
| .format(atten, |
| current_ping_stats['packet_loss_percentage'], |
| statistics.mean(current_ping_stats['rtt']), |
| current_rssi['signal_poll_rssi']['mean'], |
| current_rssi['chain_0_rssi']['mean'], |
| current_rssi['chain_1_rssi']['mean'], llstats_str)) |
| if current_ping_stats['packet_loss_percentage'] == 100: |
| zero_counter = zero_counter + 1 |
| else: |
| zero_counter = 0 |
| pending_first_ping = 0 |
| else: |
| self.log.info( |
| 'Attenuation = {}dB. Disconnected.'.format(atten)) |
| zero_counter = zero_counter + 1 |
| test_result['ping_results'].append(current_ping_stats.as_dict()) |
| # Test ends when ping loss stable at 0. If test has successfully |
| # started, test ends on MAX_CONSECUTIVE_ZEROS. In case of a restry |
| # extra zeros are allowed to ensure a test properly starts. |
| if self.retry_flag and pending_first_ping: |
| allowable_zeros = self.MAX_CONSECUTIVE_ZEROS**2 |
| else: |
| allowable_zeros = self.MAX_CONSECUTIVE_ZEROS |
| if zero_counter == allowable_zeros: |
| self.log.info('Ping loss stable at 100%. Stopping test now.') |
| for idx in range( |
| len(testcase_params['atten_range']) - |
| len(test_result['ping_results'])): |
| test_result['ping_results'].append( |
| self.DISCONNECTED_PING_RESULT) |
| break |
| # Set attenuator to initial setting |
| for attenuator in self.attenuators: |
| attenuator.set_atten(testcase_params['atten_range'][0], |
| strict=False, |
| retry=True) |
| if self.testbed_params['sniffer_enable']: |
| self.sniffer.stop_capture() |
| return test_result |
| |
| def setup_ap(self, testcase_params): |
| """Sets up the access point in the configuration required by the test. |
| |
| Args: |
| testcase_params: dict containing AP and other test params |
| """ |
| band = self.access_point.band_lookup_by_channel( |
| testcase_params['channel']) |
| if '6G' in band: |
| frequency = wutils.WifiEnums.channel_6G_to_freq[int( |
| testcase_params['channel'].strip('6g'))] |
| else: |
| if testcase_params['channel'] < 13: |
| frequency = wutils.WifiEnums.channel_2G_to_freq[ |
| testcase_params['channel']] |
| else: |
| frequency = wutils.WifiEnums.channel_5G_to_freq[ |
| testcase_params['channel']] |
| if frequency in wutils.WifiEnums.DFS_5G_FREQUENCIES: |
| self.access_point.set_region(self.testbed_params['DFS_region']) |
| else: |
| self.access_point.set_region(self.testbed_params['default_region']) |
| self.access_point.set_channel(band, testcase_params['channel']) |
| self.access_point.set_bandwidth(band, testcase_params['mode']) |
| if 'low' in testcase_params['ap_power']: |
| self.log.info('Setting low AP power.') |
| self.access_point.set_power( |
| band, self.testclass_params['low_ap_tx_power']) |
| self.log.info('Access Point Configuration: {}'.format( |
| self.access_point.ap_settings)) |
| |
| def validate_and_connect(self, testcase_params): |
| if wputils.validate_network(self.dut, |
| testcase_params['test_network']['SSID']): |
| self.log.info('Already connected to desired network') |
| else: |
| current_country = wputils.get_country_code(self.dut) |
| if current_country != self.testclass_params['country_code']: |
| self.log.warning( |
| 'Requested CC: {}, Current CC: {}. Resetting WiFi'.format( |
| self.testclass_params['country_code'], |
| current_country)) |
| wutils.wifi_toggle_state(self.dut, False) |
| wutils.set_wifi_country_code( |
| self.dut, self.testclass_params['country_code']) |
| wutils.wifi_toggle_state(self.dut, True) |
| wutils.reset_wifi(self.dut) |
| wutils.set_wifi_country_code( |
| self.dut, self.testclass_params['country_code']) |
| if self.testbed_params.get('txbf_off', False): |
| wputils.disable_beamforming(self.dut) |
| testcase_params['test_network']['channel'] = testcase_params[ |
| 'channel'] |
| wutils.wifi_connect(self.dut, |
| testcase_params['test_network'], |
| num_of_tries=5, |
| check_connectivity=True) |
| |
| def setup_dut(self, testcase_params): |
| """Sets up the DUT in the configuration required by the test. |
| |
| Args: |
| testcase_params: dict containing AP and other test params |
| """ |
| # Turn screen off to preserve battery |
| if self.testbed_params.get('screen_on', |
| False) or self.testclass_params.get( |
| 'screen_on', False): |
| self.dut.droid.wakeLockAcquireDim() |
| else: |
| self.dut.go_to_sleep() |
| self.validate_and_connect(testcase_params) |
| self.dut_ip = self.dut.droid.connectivityGetIPv4Addresses('wlan0')[0] |
| if testcase_params['channel'] not in self.atten_dut_chain_map.keys(): |
| self.atten_dut_chain_map[testcase_params[ |
| 'channel']] = wputils.get_current_atten_dut_chain_map( |
| self.attenuators, self.dut, self.ping_server) |
| self.log.info('Current Attenuator-DUT Chain Map: {}'.format( |
| self.atten_dut_chain_map[testcase_params['channel']])) |
| for idx, atten in enumerate(self.attenuators): |
| if self.atten_dut_chain_map[testcase_params['channel']][ |
| idx] == testcase_params['attenuated_chain']: |
| atten.offset = atten.instrument.max_atten |
| else: |
| atten.offset = 0 |
| |
| def setup_ping_test(self, testcase_params): |
| """Function that gets devices ready for the test. |
| |
| Args: |
| testcase_params: dict containing test-specific parameters |
| """ |
| # Configure AP |
| self.setup_ap(testcase_params) |
| # Set attenuator to starting attenuation |
| band = wputils.CHANNEL_TO_BAND_MAP[testcase_params['channel']] |
| for attenuator in self.attenuators: |
| attenuator.set_atten( |
| self.testclass_params['range_atten_start'].get(band, 0), |
| strict=False, |
| retry=True) |
| # Reset, configure, and connect DUT |
| self.setup_dut(testcase_params) |
| |
| def get_range_start_atten(self, testcase_params): |
| """Gets the starting attenuation for this ping test. |
| |
| The function gets the starting attenuation by checking whether a test |
| at the same configuration has executed. If so it sets the starting |
| point a configurable number of dBs below the reference test. |
| |
| Args: |
| testcase_params: dict containing all test parameters |
| Returns: |
| start_atten: starting attenuation for current test |
| """ |
| band = wputils.CHANNEL_TO_BAND_MAP[testcase_params['channel']] |
| # If the test is being retried, start from the beginning |
| if self.retry_flag: |
| self.log.info('Retry flag set. Setting attenuation to minimum.') |
| return self.testclass_params['range_atten_start'].get(band, 0) |
| # Get the current and reference test config. The reference test is the |
| # one performed at the current MCS+1 |
| ref_test_params = wputils.extract_sub_dict( |
| testcase_params, testcase_params['reference_params']) |
| # Check if reference test has been run and set attenuation accordingly |
| previous_params = [ |
| wputils.extract_sub_dict(result['testcase_params'], |
| testcase_params['reference_params']) |
| for result in self.testclass_results |
| ] |
| try: |
| ref_index = previous_params[::-1].index(ref_test_params) |
| ref_index = len(previous_params) - 1 - ref_index |
| start_atten = self.testclass_results[ref_index][ |
| 'atten_at_range'] - ( |
| self.testclass_params['adjacent_range_test_gap']) |
| except ValueError: |
| start_atten = self.testclass_params['range_atten_start'].get( |
| band, 0) |
| self.log.info( |
| 'Reference test not found. Starting from {} dB'.format( |
| start_atten)) |
| return start_atten |
| |
| def compile_test_params(self, testcase_params): |
| # Check if test should be skipped. |
| wputils.check_skip_conditions(testcase_params, self.dut, |
| self.access_point, |
| getattr(self, 'ota_chamber', None)) |
| |
| band = self.access_point.band_lookup_by_channel( |
| testcase_params['channel']) |
| testcase_params['test_network'] = self.main_network[band] |
| testcase_params['band'] = band |
| if testcase_params['chain_mask'] in ['0', '1']: |
| testcase_params['attenuated_chain'] = 'DUT-Chain-{}'.format( |
| 1 if testcase_params['chain_mask'] == '0' else 0) |
| else: |
| # Set attenuated chain to -1. Do not set to None as this will be |
| # compared to RF chain map which may include None |
| testcase_params['attenuated_chain'] = -1 |
| if testcase_params['test_type'] == 'test_ping_range': |
| testcase_params.update( |
| ping_interval=self.testclass_params['range_ping_interval'], |
| ping_duration=self.testclass_params['range_ping_duration'], |
| ping_size=self.testclass_params['ping_size'], |
| ) |
| elif testcase_params['test_type'] == 'test_fast_ping_rtt': |
| testcase_params.update( |
| ping_interval=self.testclass_params['rtt_ping_interval'] |
| ['fast'], |
| ping_duration=self.testclass_params['rtt_ping_duration'], |
| ping_size=self.testclass_params['ping_size'], |
| ) |
| elif testcase_params['test_type'] == 'test_slow_ping_rtt': |
| testcase_params.update( |
| ping_interval=self.testclass_params['rtt_ping_interval'] |
| ['slow'], |
| ping_duration=self.testclass_params['rtt_ping_duration'], |
| ping_size=self.testclass_params['ping_size']) |
| |
| if testcase_params['test_type'] == 'test_ping_range': |
| start_atten = self.get_range_start_atten(testcase_params) |
| num_atten_steps = int( |
| (self.testclass_params['range_atten_stop'] - start_atten) / |
| self.testclass_params['range_atten_step']) |
| testcase_params['atten_range'] = [ |
| start_atten + x * self.testclass_params['range_atten_step'] |
| for x in range(0, num_atten_steps) |
| ] |
| else: |
| testcase_params['atten_range'] = self.testclass_params[ |
| 'rtt_test_attenuation'] |
| return testcase_params |
| |
| def _test_ping(self, testcase_params): |
| """ Function that gets called for each range test case |
| |
| The function gets called in each range test case. It customizes the |
| range test based on the test name of the test that called it |
| |
| Args: |
| testcase_params: dict containing preliminary set of parameters |
| """ |
| # Compile test parameters from config and test name |
| testcase_params = self.compile_test_params(testcase_params) |
| # Run ping test |
| self.setup_ping_test(testcase_params) |
| ping_result = self.run_ping_test(testcase_params) |
| # Postprocess results |
| self.process_ping_results(testcase_params, ping_result) |
| self.testclass_results.append(ping_result) |
| self.pass_fail_check(ping_result) |
| |
| def generate_test_cases(self, ap_power, channels, modes, chain_mask, |
| test_types, **kwargs): |
| """Function that auto-generates test cases for a test class.""" |
| test_cases = [] |
| allowed_configs = { |
| 20: [ |
| 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 64, 100, |
| 116, 132, 140, 149, 153, 157, 161, '6g37', '6g117', '6g213' |
| ], |
| 40: [36, 44, 100, 149, 157, '6g37', '6g117', '6g213'], |
| 80: [36, 100, 149, '6g37', '6g117', '6g213'], |
| 160: [36, '6g37', '6g117', '6g213'] |
| } |
| |
| for channel, mode, chain, test_type in itertools.product( |
| channels, modes, chain_mask, test_types): |
| bandwidth = int(''.join([x for x in mode if x.isdigit()])) |
| if channel not in allowed_configs[bandwidth]: |
| continue |
| testcase_name = '{}_ch{}_{}_ch{}'.format(test_type, channel, mode, |
| chain) |
| testcase_params = collections.OrderedDict(test_type=test_type, |
| ap_power=ap_power, |
| channel=channel, |
| mode=mode, |
| bandwidth=bandwidth, |
| chain_mask=chain, |
| **kwargs) |
| setattr(self, testcase_name, |
| partial(self._test_ping, testcase_params)) |
| test_cases.append(testcase_name) |
| return test_cases |
| |
| |
| class WifiPing_TwoChain_Test(WifiPingTest): |
| |
| def __init__(self, controllers): |
| super().__init__(controllers) |
| self.tests = self.generate_test_cases( |
| ap_power='standard', |
| channels=[ |
| 1, 6, 11, 36, 40, 44, 48, 149, 153, 157, 161, '6g37', '6g117', |
| '6g213' |
| ], |
| modes=['bw20', 'bw80', 'bw160'], |
| test_types=[ |
| 'test_ping_range', 'test_fast_ping_rtt', 'test_slow_ping_rtt' |
| ], |
| chain_mask=['2x2'], |
| reference_params=['band', 'chain_mask']) |
| |
| |
| class WifiPing_PerChainRange_Test(WifiPingTest): |
| |
| def __init__(self, controllers): |
| super().__init__(controllers) |
| self.tests = self.generate_test_cases( |
| ap_power='standard', |
| chain_mask=['0', '1', '2x2'], |
| channels=[ |
| 1, 6, 11, 36, 40, 44, 48, 149, 153, 157, 161, '6g37', '6g117', |
| '6g213' |
| ], |
| modes=['bw20', 'bw80', 'bw160'], |
| test_types=['test_ping_range'], |
| reference_params=['band', 'chain_mask']) |
| |
| |
| class WifiPing_LowPowerAP_Test(WifiPingTest): |
| |
| def __init__(self, controllers): |
| super().__init__(controllers) |
| self.tests = self.generate_test_cases( |
| ap_power='low_power', |
| chain_mask=['0', '1', '2x2'], |
| channels=[1, 6, 11, 36, 40, 44, 48, 149, 153, 157, 161], |
| modes=['bw20', 'bw80'], |
| test_types=['test_ping_range'], |
| reference_params=['band', 'chain_mask']) |
| |
| |
| # Over-the air version of ping tests |
| class WifiOtaPingTest(WifiPingTest): |
| """Class to test over-the-air ping |
| |
| This class tests WiFi ping performance in an OTA chamber. It enables |
| setting turntable orientation and other chamber parameters to study |
| performance in varying channel conditions |
| """ |
| |
| 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 = False |
| |
| def setup_class(self): |
| WifiPingTest.setup_class(self) |
| self.ota_chamber = ota_chamber.create( |
| self.user_params['OTAChamber'])[0] |
| |
| def teardown_class(self): |
| WifiPingTest.teardown_class(self) |
| self.process_testclass_results() |
| self.ota_chamber.reset_chamber() |
| |
| def process_testclass_results(self): |
| """Saves all test results to enable comparison.""" |
| WifiPingTest.process_testclass_results(self) |
| |
| range_vs_angle = collections.OrderedDict() |
| for test in self.testclass_results: |
| curr_params = test['testcase_params'] |
| curr_config = wputils.extract_sub_dict( |
| curr_params, ['channel', 'mode', 'chain_mask']) |
| curr_config_id = tuple(curr_config.items()) |
| if curr_config_id in range_vs_angle: |
| if curr_params['position'] not in range_vs_angle[ |
| curr_config_id]['position']: |
| range_vs_angle[curr_config_id]['position'].append( |
| curr_params['position']) |
| range_vs_angle[curr_config_id]['range'].append( |
| test['range']) |
| range_vs_angle[curr_config_id]['llstats_at_range'].append( |
| test['llstats_at_range']) |
| else: |
| range_vs_angle[curr_config_id]['range'][-1] = test['range'] |
| range_vs_angle[curr_config_id]['llstats_at_range'][ |
| -1] = test['llstats_at_range'] |
| else: |
| range_vs_angle[curr_config_id] = { |
| 'position': [curr_params['position']], |
| 'range': [test['range']], |
| 'llstats_at_range': [test['llstats_at_range']] |
| } |
| chamber_mode = self.testclass_results[0]['testcase_params'][ |
| 'chamber_mode'] |
| if chamber_mode == 'orientation': |
| x_label = 'Angle (deg)' |
| elif chamber_mode == 'stepped stirrers': |
| x_label = 'Position Index' |
| figure = BokehFigure( |
| title='Range vs. Position', |
| x_label=x_label, |
| primary_y_label='Range (dB)', |
| ) |
| for curr_config_id, curr_config_data in range_vs_angle.items(): |
| curr_config = collections.OrderedDict(curr_config_id) |
| figure.add_line(x_data=curr_config_data['position'], |
| y_data=curr_config_data['range'], |
| hover_text=curr_config_data['llstats_at_range'], |
| legend='{}'.format(curr_config_id)) |
| average_range = sum(curr_config_data['range']) / len( |
| curr_config_data['range']) |
| self.log.info('Average range for {} is: {}dB'.format( |
| curr_config_id, average_range)) |
| metric_name = 'ota_summary_ch{}_{}_ch{}.avg_range'.format( |
| curr_config['channel'], curr_config['mode'], |
| curr_config['chain_mask']) |
| self.testclass_metric_logger.add_metric(metric_name, average_range) |
| current_context = context.get_current_context().get_full_output_path() |
| plot_file_path = os.path.join(current_context, 'results.html') |
| figure.generate_figure(plot_file_path) |
| |
| # Save results |
| results_file_path = os.path.join(current_context, |
| 'testclass_summary.json') |
| with open(results_file_path, 'w') as results_file: |
| json.dump(wputils.serialize_dict(range_vs_angle), |
| results_file, |
| indent=4) |
| |
| def setup_dut(self, testcase_params): |
| """Sets up the DUT in the configuration required by the test. |
| |
| Args: |
| testcase_params: dict containing AP and other test params |
| """ |
| wputils.set_chain_mask(self.dut, testcase_params['chain_mask']) |
| # Turn screen off to preserve battery |
| if self.testbed_params.get('screen_on', |
| False) or self.testclass_params.get( |
| 'screen_on', False): |
| self.dut.droid.wakeLockAcquireDim() |
| else: |
| self.dut.go_to_sleep() |
| self.validate_and_connect(testcase_params) |
| self.dut_ip = self.dut.droid.connectivityGetIPv4Addresses('wlan0')[0] |
| |
| def setup_ping_test(self, testcase_params): |
| # Setup turntable |
| if testcase_params['chamber_mode'] == 'orientation': |
| self.ota_chamber.set_orientation(testcase_params['position']) |
| elif testcase_params['chamber_mode'] == 'stepped stirrers': |
| self.ota_chamber.step_stirrers(testcase_params['total_positions']) |
| # Continue setting up ping test |
| WifiPingTest.setup_ping_test(self, testcase_params) |
| |
| def generate_test_cases(self, ap_power, channels, modes, chain_masks, |
| chamber_mode, positions, **kwargs): |
| test_cases = [] |
| allowed_configs = { |
| 20: [ |
| 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 64, 100, |
| 116, 132, 140, 149, 153, 157, 161, '6g37', '6g117', '6g213' |
| ], |
| 40: [36, 44, 100, 149, 157, '6g37', '6g117', '6g213'], |
| 80: [36, 100, 149, '6g37', '6g117', '6g213'], |
| 160: [36, '6g37', '6g117', '6g213'] |
| } |
| for channel, mode, chain_mask, position in itertools.product( |
| channels, modes, chain_masks, positions): |
| bandwidth = int(''.join([x for x in mode if x.isdigit()])) |
| if channel not in allowed_configs[bandwidth]: |
| continue |
| testcase_name = 'test_ping_range_ch{}_{}_ch{}_pos{}'.format( |
| channel, mode, chain_mask, position) |
| testcase_params = collections.OrderedDict( |
| test_type='test_ping_range', |
| ap_power=ap_power, |
| channel=channel, |
| mode=mode, |
| bandwidth=bandwidth, |
| chain_mask=chain_mask, |
| chamber_mode=chamber_mode, |
| total_positions=len(positions), |
| position=position, |
| **kwargs) |
| setattr(self, testcase_name, |
| partial(self._test_ping, testcase_params)) |
| test_cases.append(testcase_name) |
| return test_cases |
| |
| |
| class WifiOtaPing_TenDegree_Test(WifiOtaPingTest): |
| |
| def __init__(self, controllers): |
| WifiOtaPingTest.__init__(self, controllers) |
| self.tests = self.generate_test_cases( |
| ap_power='standard', |
| channels=[6, 36, 149, '6g37', '6g117', '6g213'], |
| modes=['bw20', 'bw80', 'bw160'], |
| chain_masks=['2x2'], |
| chamber_mode='orientation', |
| positions=list(range(0, 360, 10)), |
| reference_params=['channel', 'mode', 'chain_mask']) |
| |
| |
| class WifiOtaPing_SteppedStirrers_Test(WifiOtaPingTest): |
| |
| def __init__(self, controllers): |
| WifiOtaPingTest.__init__(self, controllers) |
| self.tests = self.generate_test_cases( |
| ap_power='standard', |
| channels=[6, 36, 149], |
| modes=['bw20'], |
| chain_masks=['2x2'], |
| chamber_mode='stepped stirrers', |
| positions=list(range(100)), |
| reference_params=['channel', 'mode', 'chain_mask']) |
| |
| |
| class WifiOtaPing_LowPowerAP_TenDegree_Test(WifiOtaPingTest): |
| |
| def __init__(self, controllers): |
| WifiOtaPingTest.__init__(self, controllers) |
| self.tests = self.generate_test_cases( |
| ap_power='low_power', |
| channels=[6, 36, 149], |
| modes=['bw20'], |
| chain_masks=['2x2'], |
| chamber_mode='orientation', |
| positions=list(range(0, 360, 10)), |
| reference_params=['channel', 'mode', 'chain_mask']) |
| |
| |
| class WifiOtaPing_LowPowerAP_SteppedStirrers_Test(WifiOtaPingTest): |
| |
| def __init__(self, controllers): |
| WifiOtaPingTest.__init__(self, controllers) |
| self.tests = self.generate_test_cases( |
| ap_power='low_power', |
| channels=[6, 36, 149], |
| modes=['bw20'], |
| chain_masks=['2x2'], |
| chamber_mode='stepped stirrers', |
| positions=list(range(100)), |
| reference_params=['channel', 'mode', 'chain_mask']) |
| |
| |
| class WifiOtaPing_LowPowerAP_PerChain_TenDegree_Test(WifiOtaPingTest): |
| |
| def __init__(self, controllers): |
| WifiOtaPingTest.__init__(self, controllers) |
| self.tests = self.generate_test_cases( |
| ap_power='low_power', |
| channels=[6, 36, 149], |
| modes=['bw20'], |
| chain_masks=[0, 1, '2x2'], |
| chamber_mode='orientation', |
| positions=list(range(0, 360, 10)), |
| reference_params=['channel', 'mode', 'chain_mask']) |
| |
| |
| class WifiOtaPing_PerChain_TenDegree_Test(WifiOtaPingTest): |
| |
| def __init__(self, controllers): |
| WifiOtaPingTest.__init__(self, controllers) |
| self.tests = self.generate_test_cases( |
| ap_power='standard', |
| channels=[6, 36, 149, '6g37', '6g117', '6g213'], |
| modes=['bw20'], |
| chain_masks=[0, 1, '2x2'], |
| chamber_mode='orientation', |
| positions=list(range(0, 360, 10)), |
| reference_params=['channel', 'mode', 'chain_mask']) |
