1#!/usr/bin/env python3.4 2# 3# Copyright 2017 - 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 json 18import logging 19import math 20import os 21import time 22from acts import asserts 23from acts import base_test 24from acts import utils 25from acts.controllers import iperf_server as ipf 26from acts.metrics.loggers.blackbox import BlackboxMetricLogger 27from acts.test_utils.wifi import wifi_power_test_utils as wputils 28from acts.test_utils.wifi import wifi_retail_ap as retail_ap 29from acts.test_utils.wifi import wifi_test_utils as wutils 30 31TEST_TIMEOUT = 10 32SHORT_SLEEP = 1 33MED_SLEEP = 6 34 35 36class WifiThroughputStabilityTest(base_test.BaseTestClass): 37 """Class to test WiFi throughput stability. 38 39 This class tests throughput stability and identifies cases where throughput 40 fluctuates over time. The class setups up the AP, configures and connects 41 the phone, and runs iperf throughput test at several attenuations For an 42 example config file to run this test class see 43 example_connectivity_performance_ap_sta.json. 44 """ 45 46 def __init__(self, controllers): 47 base_test.BaseTestClass.__init__(self, controllers) 48 # Define metrics to be uploaded to BlackBox 49 self.min_throughput_metric = BlackboxMetricLogger.for_test_case( 50 metric_name='min_throughput') 51 self.avg_throughput_metric = BlackboxMetricLogger.for_test_case( 52 metric_name='avg_throughput') 53 self.std_dev_percent_metric = BlackboxMetricLogger.for_test_case( 54 metric_name='std_dev_percent') 55 56 # Generate test cases 57 modes = [(6, "VHT20"), (36, "VHT20"), (36, "VHT40"), (36, "VHT80"), 58 (149, "VHT20"), (149, "VHT40"), (149, "VHT80")] 59 traffic_types = [("TCP", "DL"), ("TCP", "UL"), ("UDP", "DL"), ("UDP", 60 "UL")] 61 signal_levels = ["high", "low"] 62 self.generate_test_cases(modes, traffic_types, signal_levels) 63 64 def setup_class(self): 65 self.dut = self.android_devices[0] 66 req_params = [ 67 "throughput_stability_test_params", "testbed_params", 68 "main_network", "RetailAccessPoints" 69 ] 70 opt_params = ["golden_files_list"] 71 self.unpack_userparams(req_params, opt_params) 72 self.test_params = self.throughput_stability_test_params 73 self.num_atten = self.attenuators[0].instrument.num_atten 74 self.iperf_server = self.iperf_servers[0] 75 self.iperf_client = self.iperf_clients[0] 76 self.access_points = retail_ap.create(self.RetailAccessPoints) 77 self.access_point = self.access_points[0] 78 self.log_path = os.path.join(logging.log_path, "test_results") 79 utils.create_dir(self.log_path) 80 self.log.info("Access Point Configuration: {}".format( 81 self.access_point.ap_settings)) 82 if not hasattr(self, "golden_files_list"): 83 self.golden_files_list = [ 84 os.path.join(self.testbed_params["golden_results_path"], 85 file) for file in os.listdir( 86 self.testbed_params["golden_results_path"]) 87 ] 88 89 def teardown_test(self): 90 self.iperf_server.stop() 91 92 def pass_fail_check(self, test_result_dict): 93 """Check the test result and decide if it passed or failed. 94 95 Checks the throughput stability test's PASS/FAIL criteria based on 96 minimum instantaneous throughput, and standard deviation. 97 98 Args: 99 test_result_dict: dict containing attenuation, throughput and other 100 meta data 101 """ 102 #TODO(@oelayach): Check throughput vs RvR golden file 103 avg_throughput = test_result_dict["iperf_results"]["avg_throughput"] 104 min_throughput = test_result_dict["iperf_results"]["min_throughput"] 105 std_dev_percent = ( 106 test_result_dict["iperf_results"]["std_deviation"] / 107 test_result_dict["iperf_results"]["avg_throughput"]) * 100 108 # Set blackbox metrics 109 self.avg_throughput_metric.metric_value = avg_throughput 110 self.min_throughput_metric.metric_value = min_throughput 111 self.std_dev_percent_metric.metric_value = std_dev_percent 112 # Evaluate pass/fail 113 min_throughput_check = ( 114 (min_throughput / avg_throughput) * 115 100) > self.test_params["min_throughput_threshold"] 116 std_deviation_check = std_dev_percent < self.test_params["std_deviation_threshold"] 117 118 if min_throughput_check and std_deviation_check: 119 asserts.explicit_pass( 120 "Test Passed. Throughput at {0:.2f}dB attenuation is stable. " 121 "Mean throughput is {1:.2f} Mbps with a standard deviation of " 122 "{2:.2f}% and dips down to {3:.2f} Mbps.".format( 123 self.atten_level, avg_throughput, std_dev_percent, 124 min_throughput)) 125 asserts.fail( 126 "Test Failed. Throughput at {0:.2f}dB attenuation is unstable. " 127 "Mean throughput is {1:.2f} Mbps with a standard deviation of " 128 "{2:.2f}% and dips down to {3:.2f} Mbps.".format( 129 self.atten_level, avg_throughput, std_dev_percent, 130 min_throughput)) 131 132 def post_process_results(self, test_result): 133 """Extracts results and saves plots and JSON formatted results. 134 135 Args: 136 test_result: dict containing attenuation, iPerfResult object and 137 other meta data 138 Returns: 139 test_result_dict: dict containing post-processed results including 140 avg throughput, other metrics, and other meta data 141 """ 142 # Save output as text file 143 test_name = self.current_test_name 144 results_file_path = "{}/{}.txt".format(self.log_path, 145 self.current_test_name) 146 test_result_dict = {} 147 test_result_dict["ap_settings"] = test_result["ap_settings"].copy() 148 test_result_dict["attenuation"] = self.atten_level 149 if test_result["iperf_result"].instantaneous_rates: 150 instantaneous_rates_Mbps = [ 151 rate * 8 * (1.024**2) 152 for rate in test_result["iperf_result"].instantaneous_rates[ 153 self.test_params["iperf_ignored_interval"]:-1] 154 ] 155 else: 156 instantaneous_rates_Mbps = float("nan") 157 test_result_dict["iperf_results"] = { 158 "instantaneous_rates": 159 instantaneous_rates_Mbps, 160 "avg_throughput": 161 math.fsum(instantaneous_rates_Mbps) / 162 len(instantaneous_rates_Mbps), 163 "std_deviation": 164 test_result["iperf_result"].get_std_deviation( 165 self.test_params["iperf_ignored_interval"]) * 8, 166 "min_throughput": 167 min(instantaneous_rates_Mbps) 168 } 169 with open(results_file_path, 'w') as results_file: 170 json.dump(test_result_dict, results_file) 171 # Plot and save 172 legends = self.current_test_name 173 x_label = 'Time (s)' 174 y_label = 'Throughput (Mbps)' 175 time_data = list(range(0, len(instantaneous_rates_Mbps))) 176 data_sets = [[time_data], [instantaneous_rates_Mbps]] 177 fig_property = { 178 "title": test_name, 179 "x_label": x_label, 180 "y_label": y_label, 181 "linewidth": 3, 182 "markersize": 10 183 } 184 output_file_path = "{}/{}.html".format(self.log_path, test_name) 185 wputils.bokeh_plot( 186 data_sets, 187 legends, 188 fig_property, 189 shaded_region=None, 190 output_file_path=output_file_path) 191 return test_result_dict 192 193 def throughput_stability_test_func(self, channel, mode): 194 """Main function to test throughput stability. 195 196 The function sets up the AP in the correct channel and mode 197 configuration and runs an iperf test to measure throughput. 198 199 Args: 200 channel: Specifies AP's channel 201 mode: Specifies AP's bandwidth/mode (11g, VHT20, VHT40, VHT80) 202 Returns: 203 test_result: dict containing test result and meta data 204 """ 205 #Initialize RvR test parameters 206 test_result = {} 207 # Configure AP 208 band = self.access_point.band_lookup_by_channel(channel) 209 if "2G" in band: 210 frequency = wutils.WifiEnums.channel_2G_to_freq[channel] 211 else: 212 frequency = wutils.WifiEnums.channel_5G_to_freq[channel] 213 if frequency in wutils.WifiEnums.DFS_5G_FREQUENCIES: 214 self.access_point.set_region(self.testbed_params["DFS_region"]) 215 else: 216 self.access_point.set_region(self.testbed_params["default_region"]) 217 self.access_point.set_channel(band, channel) 218 self.access_point.set_bandwidth(band, mode) 219 self.log.info("Access Point Configuration: {}".format( 220 self.access_point.ap_settings)) 221 # Set attenuator to test level 222 self.log.info("Setting attenuation to {} dB".format(self.atten_level)) 223 for attenuator in self.attenuators: 224 attenuator.set_atten(self.atten_level) 225 # Connect DUT to Network 226 wutils.wifi_toggle_state(self.dut, True) 227 wutils.reset_wifi(self.dut) 228 self.main_network[band]["channel"] = channel 229 self.dut.droid.wifiSetCountryCode(self.test_params["country_code"]) 230 wutils.wifi_connect( 231 self.dut, 232 self.main_network[band], 233 num_of_tries=5, 234 check_connectivity=False) 235 time.sleep(MED_SLEEP) 236 # Get iperf_server address 237 if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 238 iperf_server_address = self.dut.droid.connectivityGetIPv4Addresses( 239 'wlan0')[0] 240 else: 241 iperf_server_address = self.testbed_params["iperf_server_address"] 242 # Run test and log result 243 # Start iperf session 244 self.log.info("Starting iperf test.") 245 self.iperf_server.start(tag=str(self.atten_level)) 246 client_output_path = self.iperf_client.start( 247 iperf_server_address, self.iperf_args, str(self.atten_level), 248 self.test_params["iperf_duration"] + TEST_TIMEOUT) 249 server_output_path = self.iperf_server.stop() 250 # Set attenuator to 0 dB 251 for attenuator in self.attenuators: 252 attenuator.set_atten(0) 253 # Parse and log result 254 if self.use_client_output: 255 iperf_file = client_output_path 256 else: 257 iperf_file = server_output_path 258 try: 259 iperf_result = ipf.IPerfResult(iperf_file) 260 except: 261 asserts.fail("Cannot get iperf result.") 262 test_result["ap_settings"] = self.access_point.ap_settings.copy() 263 test_result["attenuation"] = self.atten_level 264 test_result["iperf_result"] = iperf_result 265 return test_result 266 267 def get_target_atten_tput(self): 268 """Function gets attenuation used for test 269 270 The function fetches the attenuation at which the test should be 271 performed, and the expected target average throughput. 272 273 Returns: 274 test_target: dict containing target test attenuation and expected 275 throughput 276 """ 277 # Fetch the golden RvR results 278 test_name = self.current_test_name 279 rvr_golden_file_name = "test_rvr_" + "_".join(test_name.split("_")[4:]) 280 golden_path = [ 281 file_name for file_name in self.golden_files_list 282 if rvr_golden_file_name in file_name 283 ] 284 with open(golden_path[0], 'r') as golden_file: 285 golden_results = json.load(golden_file) 286 test_target = {} 287 rssi_high_low = test_name.split("_")[3] 288 if rssi_high_low == "low": 289 # Get last test point where throughput is above self.test_params["low_rssi_backoff_from_range"] 290 throughput_below_target = [ 291 x < self.test_params["low_rssi_backoff_from_range"] 292 for x in golden_results["throughput_receive"] 293 ] 294 atten_idx = throughput_below_target.index(1) - 1 295 test_target["target_attenuation"] = golden_results["attenuation"][ 296 atten_idx] 297 test_target["target_throughput"] = golden_results[ 298 "throughput_receive"][atten_idx] 299 if rssi_high_low == "high": 300 # Test at lowest attenuation point 301 test_target["target_attenuation"] = golden_results["attenuation"][ 302 0] 303 test_target["target_throughput"] = golden_results[ 304 "throughput_receive"][0] 305 return test_target 306 307 def _test_throughput_stability(self): 308 """ Function that gets called for each test case 309 310 The function gets called in each test case. The function customizes 311 the test based on the test name of the test that called it 312 """ 313 test_params = self.current_test_name.split("_") 314 channel = int(test_params[6][2:]) 315 mode = test_params[7] 316 test_target = self.get_target_atten_tput() 317 self.atten_level = test_target["target_attenuation"] 318 self.iperf_args = '-i 1 -t {} -J'.format( 319 self.test_params["iperf_duration"]) 320 if test_params[4] == "UDP": 321 self.iperf_args = self.iperf_args + " -u -b {}".format( 322 self.test_params["UDP_rates"][mode]) 323 if (test_params[5] == "DL" 324 and not isinstance(self.iperf_server, ipf.IPerfServerOverAdb) 325 ) or (test_params[5] == "UL" 326 and isinstance(self.iperf_server, ipf.IPerfServerOverAdb)): 327 self.iperf_args = self.iperf_args + ' -R' 328 self.use_client_output = True 329 else: 330 self.use_client_output = False 331 test_result = self.throughput_stability_test_func(channel, mode) 332 test_result_postprocessed = self.post_process_results(test_result) 333 self.pass_fail_check(test_result_postprocessed) 334 335 def generate_test_cases(self, modes, traffic_types, signal_levels): 336 """Function that auto-generates test cases for a test class.""" 337 testcase_wrapper = self._test_throughput_stability 338 for mode in modes: 339 for traffic_type in traffic_types: 340 for signal_level in signal_levels: 341 testcase_name = "test_tput_stability_{}_{}_{}_ch{}_{}".format( 342 signal_level, traffic_type[0], traffic_type[1], 343 mode[0], mode[1]) 344 setattr(self, testcase_name, testcase_wrapper) 345 self.tests.append(testcase_name) 346