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 collections 18import itertools 19import json 20import logging 21import numpy 22import os 23import time 24from acts import asserts 25from acts import base_test 26from acts import utils 27from acts.controllers import iperf_server as ipf 28from acts.controllers.utils_lib import ssh 29from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 30from acts_contrib.test_utils.wifi import ota_chamber 31from acts_contrib.test_utils.wifi import ota_sniffer 32from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 33from acts_contrib.test_utils.wifi import wifi_retail_ap as retail_ap 34from acts_contrib.test_utils.wifi import wifi_test_utils as wutils 35from functools import partial 36 37 38class WifiRvrTest(base_test.BaseTestClass): 39 """Class to test WiFi rate versus range. 40 41 This class implements WiFi rate versus range tests on single AP single STA 42 links. The class setups up the AP in the desired configurations, configures 43 and connects the phone to the AP, and runs iperf throughput test while 44 sweeping attenuation. For an example config file to run this test class see 45 example_connectivity_performance_ap_sta.json. 46 """ 47 48 TEST_TIMEOUT = 6 49 MAX_CONSECUTIVE_ZEROS = 3 50 51 def __init__(self, controllers): 52 base_test.BaseTestClass.__init__(self, controllers) 53 self.testcase_metric_logger = ( 54 BlackboxMappedMetricLogger.for_test_case()) 55 self.testclass_metric_logger = ( 56 BlackboxMappedMetricLogger.for_test_class()) 57 self.publish_testcase_metrics = True 58 59 def setup_class(self): 60 """Initializes common test hardware and parameters. 61 62 This function initializes hardwares and compiles parameters that are 63 common to all tests in this class. 64 """ 65 req_params = [ 66 'RetailAccessPoints', 'rvr_test_params', 'testbed_params', 67 'RemoteServer', 'main_network' 68 ] 69 opt_params = ['golden_files_list', 'OTASniffer'] 70 self.unpack_userparams(req_params, opt_params) 71 self.testclass_params = self.rvr_test_params 72 self.num_atten = self.attenuators[0].instrument.num_atten 73 self.iperf_server = self.iperf_servers[0] 74 self.remote_server = ssh.connection.SshConnection( 75 ssh.settings.from_config(self.RemoteServer[0]['ssh_config'])) 76 self.iperf_client = self.iperf_clients[0] 77 self.access_point = retail_ap.create(self.RetailAccessPoints)[0] 78 if hasattr(self, 79 'OTASniffer') and self.testbed_params['sniffer_enable']: 80 self.sniffer = ota_sniffer.create(self.OTASniffer)[0] 81 self.log.info('Access Point Configuration: {}'.format( 82 self.access_point.ap_settings)) 83 self.log_path = os.path.join(logging.log_path, 'results') 84 os.makedirs(self.log_path, exist_ok=True) 85 if not hasattr(self, 'golden_files_list'): 86 if 'golden_results_path' in self.testbed_params: 87 self.golden_files_list = [ 88 os.path.join(self.testbed_params['golden_results_path'], 89 file) for file in 90 os.listdir(self.testbed_params['golden_results_path']) 91 ] 92 else: 93 self.log.warning('No golden files found.') 94 self.golden_files_list = [] 95 self.testclass_results = [] 96 97 # Turn WiFi ON 98 if self.testclass_params.get('airplane_mode', 1): 99 for dev in self.android_devices: 100 self.log.info('Turning on airplane mode.') 101 asserts.assert_true(utils.force_airplane_mode(dev, True), 102 'Can not turn on airplane mode.') 103 wutils.wifi_toggle_state(dev, True) 104 105 def teardown_test(self): 106 self.iperf_server.stop() 107 108 def teardown_class(self): 109 # Turn WiFi OFF 110 for dev in self.android_devices: 111 wutils.wifi_toggle_state(dev, False) 112 self.process_testclass_results() 113 114 def process_testclass_results(self): 115 """Saves plot with all test results to enable comparison.""" 116 # Plot and save all results 117 plots = collections.OrderedDict() 118 for result in self.testclass_results: 119 plot_id = (result['testcase_params']['channel'], 120 result['testcase_params']['mode']) 121 if plot_id not in plots: 122 plots[plot_id] = wputils.BokehFigure( 123 title='Channel {} {} ({})'.format( 124 result['testcase_params']['channel'], 125 result['testcase_params']['mode'], 126 result['testcase_params']['traffic_type']), 127 x_label='Attenuation (dB)', 128 primary_y_label='Throughput (Mbps)') 129 plots[plot_id].add_line(result['total_attenuation'], 130 result['throughput_receive'], 131 result['test_name'], 132 marker='circle') 133 figure_list = [] 134 for plot_id, plot in plots.items(): 135 plot.generate_figure() 136 figure_list.append(plot) 137 output_file_path = os.path.join(self.log_path, 'results.html') 138 wputils.BokehFigure.save_figures(figure_list, output_file_path) 139 140 def pass_fail_check(self, rvr_result): 141 """Check the test result and decide if it passed or failed. 142 143 Checks the RvR test result and compares to a throughput limites for 144 the same configuration. The pass/fail tolerances are provided in the 145 config file. 146 147 Args: 148 rvr_result: dict containing attenuation, throughput and other data 149 """ 150 try: 151 throughput_limits = self.compute_throughput_limits(rvr_result) 152 except: 153 asserts.explicit_pass( 154 'Test passed by default. Golden file not found') 155 156 failure_count = 0 157 for idx, current_throughput in enumerate( 158 rvr_result['throughput_receive']): 159 if (current_throughput < throughput_limits['lower_limit'][idx] 160 or current_throughput > 161 throughput_limits['upper_limit'][idx]): 162 failure_count = failure_count + 1 163 164 # Set test metrics 165 rvr_result['metrics']['failure_count'] = failure_count 166 if self.publish_testcase_metrics: 167 self.testcase_metric_logger.add_metric('failure_count', 168 failure_count) 169 170 # Assert pass or fail 171 if failure_count >= self.testclass_params['failure_count_tolerance']: 172 asserts.fail('Test failed. Found {} points outside limits.'.format( 173 failure_count)) 174 asserts.explicit_pass( 175 'Test passed. Found {} points outside throughput limits.'.format( 176 failure_count)) 177 178 def compute_throughput_limits(self, rvr_result): 179 """Compute throughput limits for current test. 180 181 Checks the RvR test result and compares to a throughput limites for 182 the same configuration. The pass/fail tolerances are provided in the 183 config file. 184 185 Args: 186 rvr_result: dict containing attenuation, throughput and other meta 187 data 188 Returns: 189 throughput_limits: dict containing attenuation and throughput limit data 190 """ 191 test_name = self.current_test_name 192 golden_path = next(file_name for file_name in self.golden_files_list 193 if test_name in file_name) 194 with open(golden_path, 'r') as golden_file: 195 golden_results = json.load(golden_file) 196 golden_attenuation = [ 197 att + golden_results['fixed_attenuation'] 198 for att in golden_results['attenuation'] 199 ] 200 attenuation = [] 201 lower_limit = [] 202 upper_limit = [] 203 for idx, current_throughput in enumerate( 204 rvr_result['throughput_receive']): 205 current_att = rvr_result['attenuation'][idx] + rvr_result[ 206 'fixed_attenuation'] 207 att_distances = [ 208 abs(current_att - golden_att) 209 for golden_att in golden_attenuation 210 ] 211 sorted_distances = sorted(enumerate(att_distances), 212 key=lambda x: x[1]) 213 closest_indeces = [dist[0] for dist in sorted_distances[0:3]] 214 closest_throughputs = [ 215 golden_results['throughput_receive'][index] 216 for index in closest_indeces 217 ] 218 closest_throughputs.sort() 219 220 attenuation.append(current_att) 221 lower_limit.append( 222 max( 223 closest_throughputs[0] - max( 224 self.testclass_params['abs_tolerance'], 225 closest_throughputs[0] * 226 self.testclass_params['pct_tolerance'] / 100), 0)) 227 upper_limit.append(closest_throughputs[-1] + max( 228 self.testclass_params['abs_tolerance'], closest_throughputs[-1] 229 * self.testclass_params['pct_tolerance'] / 100)) 230 throughput_limits = { 231 'attenuation': attenuation, 232 'lower_limit': lower_limit, 233 'upper_limit': upper_limit 234 } 235 return throughput_limits 236 237 def plot_rvr_result(self, rvr_result): 238 """Saves plots and JSON formatted results. 239 240 Args: 241 rvr_result: dict containing attenuation, throughput and other meta 242 data 243 """ 244 # Save output as text file 245 test_name = self.current_test_name 246 results_file_path = os.path.join( 247 self.log_path, '{}.json'.format(self.current_test_name)) 248 with open(results_file_path, 'w') as results_file: 249 json.dump(rvr_result, results_file, indent=4) 250 # Plot and save 251 figure = wputils.BokehFigure(title=test_name, 252 x_label='Attenuation (dB)', 253 primary_y_label='Throughput (Mbps)') 254 try: 255 golden_path = next(file_name 256 for file_name in self.golden_files_list 257 if test_name in file_name) 258 with open(golden_path, 'r') as golden_file: 259 golden_results = json.load(golden_file) 260 golden_attenuation = [ 261 att + golden_results['fixed_attenuation'] 262 for att in golden_results['attenuation'] 263 ] 264 throughput_limits = self.compute_throughput_limits(rvr_result) 265 shaded_region = { 266 'x_vector': throughput_limits['attenuation'], 267 'lower_limit': throughput_limits['lower_limit'], 268 'upper_limit': throughput_limits['upper_limit'] 269 } 270 figure.add_line(golden_attenuation, 271 golden_results['throughput_receive'], 272 'Golden Results', 273 color='green', 274 marker='circle', 275 shaded_region=shaded_region) 276 except: 277 self.log.warning('ValueError: Golden file not found') 278 279 # Generate graph annotatios 280 hover_text = [ 281 'TX MCS = {0} ({1:.1f}%). RX MCS = {2} ({3:.1f}%)'.format( 282 curr_llstats['summary']['common_tx_mcs'], 283 curr_llstats['summary']['common_tx_mcs_freq'] * 100, 284 curr_llstats['summary']['common_rx_mcs'], 285 curr_llstats['summary']['common_rx_mcs_freq'] * 100) 286 for curr_llstats in rvr_result['llstats'] 287 ] 288 figure.add_line(rvr_result['total_attenuation'], 289 rvr_result['throughput_receive'], 290 'Test Results', 291 hover_text=hover_text, 292 color='red', 293 marker='circle') 294 295 output_file_path = os.path.join(self.log_path, 296 '{}.html'.format(test_name)) 297 figure.generate_figure(output_file_path) 298 299 def compute_test_metrics(self, rvr_result): 300 #Set test metrics 301 rvr_result['metrics'] = {} 302 rvr_result['metrics']['peak_tput'] = max( 303 rvr_result['throughput_receive']) 304 if self.publish_testcase_metrics: 305 self.testcase_metric_logger.add_metric( 306 'peak_tput', rvr_result['metrics']['peak_tput']) 307 308 test_mode = rvr_result['ap_settings'][rvr_result['testcase_params'] 309 ['band']]['bandwidth'] 310 tput_below_limit = [ 311 tput < 312 self.testclass_params['tput_metric_targets'][test_mode]['high'] 313 for tput in rvr_result['throughput_receive'] 314 ] 315 rvr_result['metrics']['high_tput_range'] = -1 316 for idx in range(len(tput_below_limit)): 317 if all(tput_below_limit[idx:]): 318 if idx == 0: 319 #Throughput was never above limit 320 rvr_result['metrics']['high_tput_range'] = -1 321 else: 322 rvr_result['metrics']['high_tput_range'] = rvr_result[ 323 'total_attenuation'][max(idx, 1) - 1] 324 break 325 if self.publish_testcase_metrics: 326 self.testcase_metric_logger.add_metric( 327 'high_tput_range', rvr_result['metrics']['high_tput_range']) 328 329 tput_below_limit = [ 330 tput < 331 self.testclass_params['tput_metric_targets'][test_mode]['low'] 332 for tput in rvr_result['throughput_receive'] 333 ] 334 for idx in range(len(tput_below_limit)): 335 if all(tput_below_limit[idx:]): 336 rvr_result['metrics']['low_tput_range'] = rvr_result[ 337 'total_attenuation'][max(idx, 1) - 1] 338 break 339 else: 340 rvr_result['metrics']['low_tput_range'] = -1 341 if self.publish_testcase_metrics: 342 self.testcase_metric_logger.add_metric( 343 'low_tput_range', rvr_result['metrics']['low_tput_range']) 344 345 def process_test_results(self, rvr_result): 346 self.plot_rvr_result(rvr_result) 347 self.compute_test_metrics(rvr_result) 348 349 def run_rvr_test(self, testcase_params): 350 """Test function to run RvR. 351 352 The function runs an RvR test in the current device/AP configuration. 353 Function is called from another wrapper function that sets up the 354 testbed for the RvR test 355 356 Args: 357 testcase_params: dict containing test-specific parameters 358 Returns: 359 rvr_result: dict containing rvr_results and meta data 360 """ 361 self.log.info('Start running RvR') 362 # Refresh link layer stats before test 363 llstats_obj = wputils.LinkLayerStats( 364 self.monitored_dut, 365 self.testclass_params.get('monitor_llstats', 1)) 366 zero_counter = 0 367 throughput = [] 368 llstats = [] 369 rssi = [] 370 for atten in testcase_params['atten_range']: 371 for dev in self.android_devices: 372 if not wputils.health_check(dev, 5, 50): 373 asserts.skip('DUT health check failed. Skipping test.') 374 # Set Attenuation 375 for attenuator in self.attenuators: 376 attenuator.set_atten(atten, strict=False) 377 # Refresh link layer stats 378 llstats_obj.update_stats() 379 # Setup sniffer 380 if self.testbed_params['sniffer_enable']: 381 self.sniffer.start_capture( 382 network=testcase_params['test_network'], 383 chan=int(testcase_params['channel']), 384 bw=testcase_params['bandwidth'], 385 duration=self.testclass_params['iperf_duration'] / 5) 386 # Start iperf session 387 if self.testclass_params.get('monitor_rssi', 1): 388 rssi_future = wputils.get_connected_rssi_nb( 389 self.monitored_dut, 390 self.testclass_params['iperf_duration'] - 1, 391 1, 392 1, 393 interface=self.monitored_interface) 394 self.iperf_server.start(tag=str(atten)) 395 client_output_path = self.iperf_client.start( 396 testcase_params['iperf_server_address'], 397 testcase_params['iperf_args'], str(atten), 398 self.testclass_params['iperf_duration'] + self.TEST_TIMEOUT) 399 server_output_path = self.iperf_server.stop() 400 if self.testclass_params.get('monitor_rssi', 1): 401 rssi_result = rssi_future.result() 402 current_rssi = { 403 'signal_poll_rssi': 404 rssi_result['signal_poll_rssi']['mean'], 405 'chain_0_rssi': rssi_result['chain_0_rssi']['mean'], 406 'chain_1_rssi': rssi_result['chain_1_rssi']['mean'] 407 } 408 else: 409 current_rssi = { 410 'signal_poll_rssi': float('nan'), 411 'chain_0_rssi': float('nan'), 412 'chain_1_rssi': float('nan') 413 } 414 rssi.append(current_rssi) 415 # Stop sniffer 416 if self.testbed_params['sniffer_enable']: 417 self.sniffer.stop_capture(tag=str(atten)) 418 # Parse and log result 419 if testcase_params['use_client_output']: 420 iperf_file = client_output_path 421 else: 422 iperf_file = server_output_path 423 try: 424 iperf_result = ipf.IPerfResult(iperf_file) 425 curr_throughput = numpy.mean(iperf_result.instantaneous_rates[ 426 self.testclass_params['iperf_ignored_interval']:-1] 427 ) * 8 * (1.024**2) 428 except: 429 self.log.warning( 430 'ValueError: Cannot get iperf result. Setting to 0') 431 curr_throughput = 0 432 throughput.append(curr_throughput) 433 llstats_obj.update_stats() 434 curr_llstats = llstats_obj.llstats_incremental.copy() 435 llstats.append(curr_llstats) 436 self.log.info( 437 ('Throughput at {0:.2f} dB is {1:.2f} Mbps. ' 438 'RSSI = {2:.2f} [{3:.2f}, {4:.2f}].').format( 439 atten, curr_throughput, current_rssi['signal_poll_rssi'], 440 current_rssi['chain_0_rssi'], 441 current_rssi['chain_1_rssi'])) 442 if curr_throughput == 0 and ( 443 current_rssi['signal_poll_rssi'] < -80 444 or numpy.isnan(current_rssi['signal_poll_rssi'])): 445 zero_counter = zero_counter + 1 446 else: 447 zero_counter = 0 448 if zero_counter == self.MAX_CONSECUTIVE_ZEROS: 449 self.log.info( 450 'Throughput stable at 0 Mbps. Stopping test now.') 451 throughput.extend( 452 [0] * 453 (len(testcase_params['atten_range']) - len(throughput))) 454 break 455 for attenuator in self.attenuators: 456 attenuator.set_atten(0, strict=False) 457 # Compile test result and meta data 458 rvr_result = collections.OrderedDict() 459 rvr_result['test_name'] = self.current_test_name 460 rvr_result['testcase_params'] = testcase_params.copy() 461 rvr_result['ap_settings'] = self.access_point.ap_settings.copy() 462 rvr_result['fixed_attenuation'] = self.testbed_params[ 463 'fixed_attenuation'][str(testcase_params['channel'])] 464 rvr_result['attenuation'] = list(testcase_params['atten_range']) 465 rvr_result['total_attenuation'] = [ 466 att + rvr_result['fixed_attenuation'] 467 for att in rvr_result['attenuation'] 468 ] 469 rvr_result['rssi'] = rssi 470 rvr_result['throughput_receive'] = throughput 471 rvr_result['llstats'] = llstats 472 return rvr_result 473 474 def setup_ap(self, testcase_params): 475 """Sets up the access point in the configuration required by the test. 476 477 Args: 478 testcase_params: dict containing AP and other test params 479 """ 480 if '2G' in testcase_params['band']: 481 frequency = wutils.WifiEnums.channel_2G_to_freq[ 482 testcase_params['channel']] 483 else: 484 frequency = wutils.WifiEnums.channel_5G_to_freq[ 485 testcase_params['channel']] 486 if frequency in wutils.WifiEnums.DFS_5G_FREQUENCIES: 487 self.access_point.set_region(self.testbed_params['DFS_region']) 488 else: 489 self.access_point.set_region(self.testbed_params['default_region']) 490 self.access_point.set_channel(testcase_params['band'], 491 testcase_params['channel']) 492 self.access_point.set_bandwidth(testcase_params['band'], 493 testcase_params['mode']) 494 self.log.info('Access Point Configuration: {}'.format( 495 self.access_point.ap_settings)) 496 497 def setup_dut(self, testcase_params): 498 """Sets up the DUT in the configuration required by the test. 499 500 Args: 501 testcase_params: dict containing AP and other test params 502 """ 503 self.sta_dut = self.android_devices[0] 504 # Check battery level before test 505 if not wputils.health_check( 506 self.sta_dut, 507 20) and testcase_params['traffic_direction'] == 'UL': 508 asserts.skip('Overheating or Battery level low. Skipping test.') 509 # Turn screen off to preserve battery 510 self.sta_dut.go_to_sleep() 511 if wputils.validate_network(self.sta_dut, 512 testcase_params['test_network']['SSID']): 513 self.log.info('Already connected to desired network') 514 else: 515 wutils.reset_wifi(self.sta_dut) 516 wutils.set_wifi_country_code(self.sta_dut, 517 self.testclass_params['country_code']) 518 if self.testbed_params['sniffer_enable']: 519 self.sniffer.start_capture( 520 network={'SSID': testcase_params['test_network']['SSID']}, 521 chan=testcase_params['channel'], 522 bw=testcase_params['bandwidth'], 523 duration=180) 524 try: 525 wutils.wifi_connect(self.sta_dut, 526 testcase_params['test_network'], 527 num_of_tries=5, 528 check_connectivity=True) 529 finally: 530 if self.testbed_params['sniffer_enable']: 531 self.sniffer.stop_capture(tag='connection_setup') 532 533 def setup_rvr_test(self, testcase_params): 534 """Function that gets devices ready for the test. 535 536 Args: 537 testcase_params: dict containing test-specific parameters 538 """ 539 # Configure AP 540 self.setup_ap(testcase_params) 541 # Set attenuator to 0 dB 542 for attenuator in self.attenuators: 543 attenuator.set_atten(0, strict=False) 544 # Reset, configure, and connect DUT 545 self.setup_dut(testcase_params) 546 # Wait before running the first wifi test 547 first_test_delay = self.testclass_params.get('first_test_delay', 600) 548 if first_test_delay > 0 and len(self.testclass_results) == 0: 549 self.log.info('Waiting before the first RvR test.') 550 time.sleep(first_test_delay) 551 self.setup_dut(testcase_params) 552 # Get iperf_server address 553 sta_dut_ip = self.sta_dut.droid.connectivityGetIPv4Addresses( 554 'wlan0')[0] 555 if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 556 testcase_params['iperf_server_address'] = sta_dut_ip 557 else: 558 if self.testbed_params.get('lan_traffic_only', True): 559 testcase_params[ 560 'iperf_server_address'] = wputils.get_server_address( 561 self.remote_server, sta_dut_ip, '255.255.255.0') 562 else: 563 testcase_params[ 564 'iperf_server_address'] = wputils.get_server_address( 565 self.remote_server, sta_dut_ip, 'public') 566 # Set DUT to monitor RSSI and LLStats on 567 self.monitored_dut = self.sta_dut 568 self.monitored_interface = None 569 570 def compile_test_params(self, testcase_params): 571 """Function that completes all test params based on the test name. 572 573 Args: 574 testcase_params: dict containing test-specific parameters 575 """ 576 num_atten_steps = int((self.testclass_params['atten_stop'] - 577 self.testclass_params['atten_start']) / 578 self.testclass_params['atten_step']) 579 testcase_params['atten_range'] = [ 580 self.testclass_params['atten_start'] + 581 x * self.testclass_params['atten_step'] 582 for x in range(0, num_atten_steps) 583 ] 584 band = self.access_point.band_lookup_by_channel( 585 testcase_params['channel']) 586 testcase_params['band'] = band 587 testcase_params['test_network'] = self.main_network[band] 588 if testcase_params['traffic_type'] == 'TCP': 589 testcase_params['iperf_socket_size'] = self.testclass_params.get( 590 'tcp_socket_size', None) 591 testcase_params['iperf_processes'] = self.testclass_params.get( 592 'tcp_processes', 1) 593 elif testcase_params['traffic_type'] == 'UDP': 594 testcase_params['iperf_socket_size'] = self.testclass_params.get( 595 'udp_socket_size', None) 596 testcase_params['iperf_processes'] = self.testclass_params.get( 597 'udp_processes', 1) 598 if (testcase_params['traffic_direction'] == 'DL' 599 and not isinstance(self.iperf_server, ipf.IPerfServerOverAdb) 600 ) or (testcase_params['traffic_direction'] == 'UL' 601 and isinstance(self.iperf_server, ipf.IPerfServerOverAdb)): 602 testcase_params['iperf_args'] = wputils.get_iperf_arg_string( 603 duration=self.testclass_params['iperf_duration'], 604 reverse_direction=1, 605 traffic_type=testcase_params['traffic_type'], 606 socket_size=testcase_params['iperf_socket_size'], 607 num_processes=testcase_params['iperf_processes'], 608 udp_throughput=self.testclass_params['UDP_rates'][ 609 testcase_params['mode']]) 610 testcase_params['use_client_output'] = True 611 else: 612 testcase_params['iperf_args'] = wputils.get_iperf_arg_string( 613 duration=self.testclass_params['iperf_duration'], 614 reverse_direction=0, 615 traffic_type=testcase_params['traffic_type'], 616 socket_size=testcase_params['iperf_socket_size'], 617 num_processes=testcase_params['iperf_processes'], 618 udp_throughput=self.testclass_params['UDP_rates'][ 619 testcase_params['mode']]) 620 testcase_params['use_client_output'] = False 621 return testcase_params 622 623 def _test_rvr(self, testcase_params): 624 """ Function that gets called for each test case 625 626 Args: 627 testcase_params: dict containing test-specific parameters 628 """ 629 # Compile test parameters from config and test name 630 testcase_params = self.compile_test_params(testcase_params) 631 632 # Prepare devices and run test 633 self.setup_rvr_test(testcase_params) 634 rvr_result = self.run_rvr_test(testcase_params) 635 636 # Post-process results 637 self.testclass_results.append(rvr_result) 638 self.process_test_results(rvr_result) 639 self.pass_fail_check(rvr_result) 640 641 def generate_test_cases(self, channels, modes, traffic_types, 642 traffic_directions): 643 """Function that auto-generates test cases for a test class.""" 644 test_cases = [] 645 allowed_configs = { 646 20: [ 647 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 64, 100, 648 116, 132, 140, 149, 153, 157, 161 649 ], 650 40: [36, 44, 100, 149, 157], 651 80: [36, 100, 149], 652 160: [36] 653 } 654 655 for channel, mode, traffic_type, traffic_direction in itertools.product( 656 channels, modes, traffic_types, traffic_directions): 657 bandwidth = int(''.join([x for x in mode if x.isdigit()])) 658 if channel not in allowed_configs[bandwidth]: 659 continue 660 test_name = 'test_rvr_{}_{}_ch{}_{}'.format( 661 traffic_type, traffic_direction, channel, mode) 662 test_params = collections.OrderedDict( 663 channel=channel, 664 mode=mode, 665 bandwidth=bandwidth, 666 traffic_type=traffic_type, 667 traffic_direction=traffic_direction) 668 setattr(self, test_name, partial(self._test_rvr, test_params)) 669 test_cases.append(test_name) 670 return test_cases 671 672 673class WifiRvr_TCP_Test(WifiRvrTest): 674 def __init__(self, controllers): 675 super().__init__(controllers) 676 self.tests = self.generate_test_cases( 677 channels=[1, 6, 11, 36, 40, 44, 48, 149, 153, 157, 161], 678 modes=['bw20', 'bw40', 'bw80', 'bw160'], 679 traffic_types=['TCP'], 680 traffic_directions=['DL', 'UL']) 681 682 683class WifiRvr_VHT_TCP_Test(WifiRvrTest): 684 def __init__(self, controllers): 685 super().__init__(controllers) 686 self.tests = self.generate_test_cases( 687 channels=[1, 6, 11, 36, 40, 44, 48, 149, 153, 157, 161], 688 modes=['VHT20', 'VHT40', 'VHT80'], 689 traffic_types=['TCP'], 690 traffic_directions=['DL', 'UL']) 691 692 693class WifiRvr_HE_TCP_Test(WifiRvrTest): 694 def __init__(self, controllers): 695 super().__init__(controllers) 696 self.tests = self.generate_test_cases( 697 channels=[1, 6, 11, 36, 40, 44, 48, 149, 153, 157, 161], 698 modes=['HE20', 'HE40', 'HE80', 'HE160'], 699 traffic_types=['TCP'], 700 traffic_directions=['DL', 'UL']) 701 702 703class WifiRvr_SampleUDP_Test(WifiRvrTest): 704 def __init__(self, controllers): 705 super().__init__(controllers) 706 self.tests = self.generate_test_cases( 707 channels=[6, 36, 149], 708 modes=['bw20', 'bw40', 'bw80', 'bw160'], 709 traffic_types=['UDP'], 710 traffic_directions=['DL', 'UL']) 711 712 713class WifiRvr_VHT_SampleUDP_Test(WifiRvrTest): 714 def __init__(self, controllers): 715 super().__init__(controllers) 716 self.tests = self.generate_test_cases( 717 channels=[6, 36, 149], 718 modes=['VHT20', 'VHT40', 'VHT80', 'VHT160'], 719 traffic_types=['UDP'], 720 traffic_directions=['DL', 'UL']) 721 722 723class WifiRvr_HE_SampleUDP_Test(WifiRvrTest): 724 def __init__(self, controllers): 725 super().__init__(controllers) 726 self.tests = self.generate_test_cases( 727 channels=[6, 36, 149], 728 modes=['HE20', 'HE40', 'HE80', 'HE160'], 729 traffic_types=['UDP'], 730 traffic_directions=['DL', 'UL']) 731 732 733class WifiRvr_SampleDFS_Test(WifiRvrTest): 734 def __init__(self, controllers): 735 super().__init__(controllers) 736 self.tests = self.generate_test_cases( 737 channels=[64, 100, 116, 132, 140], 738 modes=['bw20', 'bw40', 'bw80'], 739 traffic_types=['TCP'], 740 traffic_directions=['DL', 'UL']) 741 742 743# Over-the air version of RVR tests 744class WifiOtaRvrTest(WifiRvrTest): 745 """Class to test over-the-air RvR 746 747 This class implements measures WiFi RvR tests in an OTA chamber. It enables 748 setting turntable orientation and other chamber parameters to study 749 performance in varying channel conditions 750 """ 751 def __init__(self, controllers): 752 base_test.BaseTestClass.__init__(self, controllers) 753 self.testcase_metric_logger = ( 754 BlackboxMappedMetricLogger.for_test_case()) 755 self.testclass_metric_logger = ( 756 BlackboxMappedMetricLogger.for_test_class()) 757 self.publish_testcase_metrics = False 758 759 def setup_class(self): 760 WifiRvrTest.setup_class(self) 761 self.ota_chamber = ota_chamber.create( 762 self.user_params['OTAChamber'])[0] 763 764 def teardown_class(self): 765 WifiRvrTest.teardown_class(self) 766 self.ota_chamber.reset_chamber() 767 768 def extract_test_id(self, testcase_params, id_fields): 769 test_id = collections.OrderedDict( 770 (param, testcase_params[param]) for param in id_fields) 771 return test_id 772 773 def process_testclass_results(self): 774 """Saves plot with all test results to enable comparison.""" 775 # Plot individual test id results raw data and compile metrics 776 plots = collections.OrderedDict() 777 compiled_data = collections.OrderedDict() 778 for result in self.testclass_results: 779 test_id = tuple( 780 self.extract_test_id( 781 result['testcase_params'], 782 ['channel', 'mode', 'traffic_type', 'traffic_direction' 783 ]).items()) 784 if test_id not in plots: 785 # Initialize test id data when not present 786 compiled_data[test_id] = {'throughput': [], 'metrics': {}} 787 compiled_data[test_id]['metrics'] = { 788 key: [] 789 for key in result['metrics'].keys() 790 } 791 plots[test_id] = wputils.BokehFigure( 792 title='Channel {} {} ({} {})'.format( 793 result['testcase_params']['channel'], 794 result['testcase_params']['mode'], 795 result['testcase_params']['traffic_type'], 796 result['testcase_params']['traffic_direction']), 797 x_label='Attenuation (dB)', 798 primary_y_label='Throughput (Mbps)') 799 # Compile test id data and metrics 800 compiled_data[test_id]['throughput'].append( 801 result['throughput_receive']) 802 compiled_data[test_id]['total_attenuation'] = result[ 803 'total_attenuation'] 804 for metric_key, metric_value in result['metrics'].items(): 805 compiled_data[test_id]['metrics'][metric_key].append( 806 metric_value) 807 # Add test id to plots 808 plots[test_id].add_line(result['total_attenuation'], 809 result['throughput_receive'], 810 result['test_name'], 811 width=1, 812 style='dashed', 813 marker='circle') 814 815 # Compute average RvRs and compount metrics over orientations 816 for test_id, test_data in compiled_data.items(): 817 test_id_dict = dict(test_id) 818 metric_tag = '{}_{}_ch{}_{}'.format( 819 test_id_dict['traffic_type'], 820 test_id_dict['traffic_direction'], test_id_dict['channel'], 821 test_id_dict['mode']) 822 high_tput_hit_freq = numpy.mean( 823 numpy.not_equal(test_data['metrics']['high_tput_range'], -1)) 824 self.testclass_metric_logger.add_metric( 825 '{}.high_tput_hit_freq'.format(metric_tag), high_tput_hit_freq) 826 for metric_key, metric_value in test_data['metrics'].items(): 827 metric_key = '{}.avg_{}'.format(metric_tag, metric_key) 828 metric_value = numpy.mean(metric_value) 829 self.testclass_metric_logger.add_metric( 830 metric_key, metric_value) 831 test_data['avg_rvr'] = numpy.mean(test_data['throughput'], 0) 832 test_data['median_rvr'] = numpy.median(test_data['throughput'], 0) 833 plots[test_id].add_line(test_data['total_attenuation'], 834 test_data['avg_rvr'], 835 legend='Average Throughput', 836 marker='circle') 837 plots[test_id].add_line(test_data['total_attenuation'], 838 test_data['median_rvr'], 839 legend='Median Throughput', 840 marker='square') 841 842 figure_list = [] 843 for test_id, plot in plots.items(): 844 plot.generate_figure() 845 figure_list.append(plot) 846 output_file_path = os.path.join(self.log_path, 'results.html') 847 wputils.BokehFigure.save_figures(figure_list, output_file_path) 848 849 def setup_rvr_test(self, testcase_params): 850 # Set turntable orientation 851 self.ota_chamber.set_orientation(testcase_params['orientation']) 852 # Continue test setup 853 WifiRvrTest.setup_rvr_test(self, testcase_params) 854 855 def generate_test_cases(self, channels, modes, angles, traffic_types, 856 directions): 857 test_cases = [] 858 allowed_configs = { 859 20: [ 860 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 64, 100, 861 116, 132, 140, 149, 153, 157, 161 862 ], 863 40: [36, 44, 100, 149, 157], 864 80: [36, 100, 149], 865 160: [36] 866 } 867 for channel, mode, angle, traffic_type, direction in itertools.product( 868 channels, modes, angles, traffic_types, directions): 869 bandwidth = int(''.join([x for x in mode if x.isdigit()])) 870 if channel not in allowed_configs[bandwidth]: 871 continue 872 testcase_name = 'test_rvr_{}_{}_ch{}_{}_{}deg'.format( 873 traffic_type, direction, channel, mode, angle) 874 test_params = collections.OrderedDict(channel=channel, 875 mode=mode, 876 bandwidth=bandwidth, 877 traffic_type=traffic_type, 878 traffic_direction=direction, 879 orientation=angle) 880 setattr(self, testcase_name, partial(self._test_rvr, test_params)) 881 test_cases.append(testcase_name) 882 return test_cases 883 884 885class WifiOtaRvr_StandardOrientation_Test(WifiOtaRvrTest): 886 def __init__(self, controllers): 887 WifiOtaRvrTest.__init__(self, controllers) 888 self.tests = self.generate_test_cases( 889 [1, 6, 11, 36, 40, 44, 48, 149, 153, 157, 161], 890 ['bw20', 'bw40', 'bw80'], list(range(0, 360, 45)), ['TCP'], ['DL']) 891 892 893class WifiOtaRvr_SampleChannel_Test(WifiOtaRvrTest): 894 def __init__(self, controllers): 895 WifiOtaRvrTest.__init__(self, controllers) 896 self.tests = self.generate_test_cases([6], ['bw20'], 897 list(range(0, 360, 45)), ['TCP'], 898 ['DL']) 899 self.tests.extend( 900 self.generate_test_cases([36, 149], ['bw80'], 901 list(range(0, 360, 45)), ['TCP'], ['DL'])) 902 903 904class WifiOtaRvr_SingleOrientation_Test(WifiOtaRvrTest): 905 def __init__(self, controllers): 906 WifiOtaRvrTest.__init__(self, controllers) 907 self.tests = self.generate_test_cases( 908 [6, 36, 40, 44, 48, 149, 153, 157, 161], ['bw20', 'bw40', 'bw80'], 909 [0], ['TCP'], ['DL', 'UL']) 910