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 context 27from acts import utils 28from acts.controllers import iperf_server as ipf 29from acts.controllers.utils_lib import ssh 30from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 31from acts_contrib.test_utils.wifi import ota_chamber 32from acts_contrib.test_utils.wifi import ota_sniffer 33from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 34from acts_contrib.test_utils.wifi.wifi_performance_test_utils.bokeh_figure import BokehFigure 35from acts_contrib.test_utils.wifi import wifi_retail_ap as retail_ap 36from acts_contrib.test_utils.wifi import wifi_test_utils as wutils 37from functools import partial 38 39TEST_TIMEOUT = 10 40SHORT_SLEEP = 1 41MED_SLEEP = 6 42 43 44class WifiThroughputStabilityTest(base_test.BaseTestClass): 45 """Class to test WiFi throughput stability. 46 47 This class tests throughput stability and identifies cases where throughput 48 fluctuates over time. The class setups up the AP, configures and connects 49 the phone, and runs iperf throughput test at several attenuations For an 50 example config file to run this test class see 51 example_connectivity_performance_ap_sta.json. 52 """ 53 54 def __init__(self, controllers): 55 base_test.BaseTestClass.__init__(self, controllers) 56 # Define metrics to be uploaded to BlackBox 57 self.testcase_metric_logger = ( 58 BlackboxMappedMetricLogger.for_test_case()) 59 self.testclass_metric_logger = ( 60 BlackboxMappedMetricLogger.for_test_class()) 61 self.publish_testcase_metrics = True 62 # Generate test cases 63 self.tests = self.generate_test_cases( 64 [6, 36, 149, '6g37'], ['bw20', 'bw40', 'bw80', 'bw160'], 65 ['TCP', 'UDP'], ['DL', 'UL'], ['high', 'low']) 66 67 def generate_test_cases(self, channels, modes, traffic_types, 68 traffic_directions, signal_levels): 69 """Function that auto-generates test cases for a test class.""" 70 allowed_configs = { 71 20: [ 72 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 64, 100, 73 116, 132, 140, 149, 153, 157, 161, '6g37', '6g117', '6g213' 74 ], 75 40: [36, 44, 100, 149, 157, '6g37', '6g117', '6g213'], 76 80: [36, 100, 149, '6g37', '6g117', '6g213'], 77 160: [36, '6g37', '6g117', '6g213'] 78 } 79 80 test_cases = [] 81 for channel, mode, signal_level, traffic_type, traffic_direction in itertools.product( 82 channels, 83 modes, 84 signal_levels, 85 traffic_types, 86 traffic_directions, 87 ): 88 bandwidth = int(''.join([x for x in mode if x.isdigit()])) 89 if channel not in allowed_configs[bandwidth]: 90 continue 91 testcase_params = collections.OrderedDict( 92 channel=channel, 93 mode=mode, 94 bandwidth=bandwidth, 95 traffic_type=traffic_type, 96 traffic_direction=traffic_direction, 97 signal_level=signal_level) 98 testcase_name = ('test_tput_stability' 99 '_{}_{}_{}_ch{}_{}'.format( 100 signal_level, traffic_type, traffic_direction, 101 channel, mode)) 102 setattr(self, testcase_name, 103 partial(self._test_throughput_stability, testcase_params)) 104 test_cases.append(testcase_name) 105 return test_cases 106 107 def setup_class(self): 108 self.dut = self.android_devices[0] 109 req_params = [ 110 'throughput_stability_test_params', 'testbed_params', 111 'main_network', 'RetailAccessPoints', 'RemoteServer' 112 ] 113 opt_params = ['OTASniffer'] 114 self.unpack_userparams(req_params, opt_params) 115 self.testclass_params = self.throughput_stability_test_params 116 self.num_atten = self.attenuators[0].instrument.num_atten 117 self.remote_server = ssh.connection.SshConnection( 118 ssh.settings.from_config(self.RemoteServer[0]['ssh_config'])) 119 self.iperf_server = self.iperf_servers[0] 120 self.iperf_client = self.iperf_clients[0] 121 self.access_point = retail_ap.create(self.RetailAccessPoints)[0] 122 if hasattr(self, 123 'OTASniffer') and self.testbed_params['sniffer_enable']: 124 try: 125 self.sniffer = ota_sniffer.create(self.OTASniffer)[0] 126 except: 127 self.log.warning('Could not start sniffer. Disabling sniffs.') 128 self.testbed_params['sniffer_enable'] = 0 129 self.sniffer_subsampling = 1 130 self.log_path = os.path.join(logging.log_path, 'test_results') 131 os.makedirs(self.log_path, exist_ok=True) 132 self.log.info('Access Point Configuration: {}'.format( 133 self.access_point.ap_settings)) 134 self.ref_attenuations = {} 135 self.testclass_results = [] 136 137 # Turn WiFi ON 138 if self.testclass_params.get('airplane_mode', 1): 139 self.log.info('Turning on airplane mode.') 140 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 141 'Can not turn on airplane mode.') 142 wutils.wifi_toggle_state(self.dut, True) 143 144 def teardown_test(self): 145 self.iperf_server.stop() 146 147 def teardown_class(self): 148 self.access_point.teardown() 149 # Turn WiFi OFF 150 for dev in self.android_devices: 151 wutils.wifi_toggle_state(dev, False) 152 dev.go_to_sleep() 153 154 def pass_fail_check(self, test_result): 155 """Check the test result and decide if it passed or failed. 156 157 Checks the throughput stability test's PASS/FAIL criteria based on 158 minimum instantaneous throughput, and standard deviation. 159 160 Args: 161 test_result_dict: dict containing attenuation, throughput and other 162 meta data 163 """ 164 # Evaluate pass/fail 165 min_throughput_check = ( 166 (test_result['iperf_summary']['min_throughput'] / 167 test_result['iperf_summary']['avg_throughput']) * 168 100) > self.testclass_params['min_throughput_threshold'] 169 std_deviation_check = test_result['iperf_summary'][ 170 'std_dev_percent'] < self.testclass_params[ 171 'std_deviation_threshold'] 172 173 if min_throughput_check and std_deviation_check: 174 asserts.explicit_pass('Test Passed.') 175 asserts.fail('Test Failed.') 176 177 def post_process_results(self, test_result): 178 """Extracts results and saves plots and JSON formatted results. 179 180 Args: 181 test_result: dict containing attenuation, iPerfResult object and 182 other meta data 183 Returns: 184 test_result_dict: dict containing post-processed results including 185 avg throughput, other metrics, and other meta data 186 """ 187 # Save output as text file 188 results_file_path = os.path.join( 189 self.log_path, '{}.txt'.format(self.current_test_name)) 190 with open(results_file_path, 'w') as results_file: 191 json.dump(wputils.serialize_dict(test_result), results_file) 192 # Plot and save 193 # Set blackbox metrics 194 if self.publish_testcase_metrics: 195 self.testcase_metric_logger.add_metric( 196 'avg_throughput', 197 test_result['iperf_summary']['avg_throughput']) 198 self.testcase_metric_logger.add_metric( 199 'min_throughput', 200 test_result['iperf_summary']['min_throughput']) 201 self.testcase_metric_logger.add_metric( 202 'std_dev_percent', 203 test_result['iperf_summary']['std_dev_percent']) 204 figure = BokehFigure(self.current_test_name, 205 x_label='Time (s)', 206 primary_y_label='Throughput (Mbps)') 207 time_data = list( 208 range( 209 0, 210 len(test_result['iperf_summary']['instantaneous_rates']))) 211 figure.add_line( 212 time_data, 213 test_result['iperf_summary']['instantaneous_rates'], 214 legend=self.current_test_name, 215 marker='circle') 216 output_file_path = os.path.join( 217 self.log_path, '{}.html'.format(self.current_test_name)) 218 figure.generate_figure(output_file_path) 219 return test_result 220 221 def setup_ap(self, testcase_params): 222 """Sets up the access point in the configuration required by the test. 223 224 Args: 225 testcase_params: dict containing AP and other test params 226 """ 227 band = self.access_point.band_lookup_by_channel( 228 testcase_params['channel']) 229 if '6G' in band: 230 frequency = wutils.WifiEnums.channel_6G_to_freq[int( 231 testcase_params['channel'].strip('6g'))] 232 else: 233 if testcase_params['channel'] < 13: 234 frequency = wutils.WifiEnums.channel_2G_to_freq[ 235 testcase_params['channel']] 236 else: 237 frequency = wutils.WifiEnums.channel_5G_to_freq[ 238 testcase_params['channel']] 239 if frequency in wutils.WifiEnums.DFS_5G_FREQUENCIES: 240 self.access_point.set_region(self.testbed_params['DFS_region']) 241 else: 242 self.access_point.set_region(self.testbed_params['default_region']) 243 self.access_point.set_channel(band, testcase_params['channel']) 244 self.access_point.set_bandwidth(band, testcase_params['mode']) 245 self.log.info('Access Point Configuration: {}'.format( 246 self.access_point.ap_settings)) 247 248 def setup_dut(self, testcase_params): 249 """Sets up the DUT in the configuration required by the test. 250 251 Args: 252 testcase_params: dict containing AP and other test params 253 """ 254 # Turn screen off to preserve battery 255 if self.testbed_params.get('screen_on', 256 False) or self.testclass_params.get( 257 'screen_on', False): 258 self.dut.droid.wakeLockAcquireDim() 259 else: 260 self.dut.go_to_sleep() 261 band = self.access_point.band_lookup_by_channel( 262 testcase_params['channel']) 263 if wputils.validate_network(self.dut, 264 testcase_params['test_network']['SSID']): 265 self.log.info('Already connected to desired network') 266 else: 267 wutils.wifi_toggle_state(self.dut, True) 268 wutils.reset_wifi(self.dut) 269 if self.testbed_params.get('txbf_off', False): 270 wputils.disable_beamforming(self.dut) 271 wutils.set_wifi_country_code(self.dut, 272 self.testclass_params['country_code']) 273 self.main_network[band]['channel'] = testcase_params['channel'] 274 wutils.wifi_connect(self.dut, 275 testcase_params['test_network'], 276 num_of_tries=5, 277 check_connectivity=True) 278 self.dut_ip = self.dut.droid.connectivityGetIPv4Addresses('wlan0')[0] 279 280 def setup_throughput_stability_test(self, testcase_params): 281 """Function that gets devices ready for the test. 282 283 Args: 284 testcase_params: dict containing test-specific parameters 285 """ 286 # Configure AP 287 self.setup_ap(testcase_params) 288 # Set attenuator to 0 dB 289 for attenuator in self.attenuators: 290 attenuator.set_atten(0, strict=False, retry=True) 291 # Reset, configure, and connect DUT 292 self.setup_dut(testcase_params) 293 # Wait before running the first wifi test 294 first_test_delay = self.testclass_params.get('first_test_delay', 600) 295 if first_test_delay > 0 and len(self.testclass_results) == 0: 296 self.log.info('Waiting before the first test.') 297 time.sleep(first_test_delay) 298 self.setup_dut(testcase_params) 299 # Get and set attenuation levels for test 300 testcase_params['atten_level'] = self.get_target_atten(testcase_params) 301 self.log.info('Setting attenuation to {} dB'.format( 302 testcase_params['atten_level'])) 303 for attenuator in self.attenuators: 304 attenuator.set_atten(testcase_params['atten_level']) 305 # Configure iperf 306 if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 307 testcase_params['iperf_server_address'] = self.dut_ip 308 else: 309 testcase_params[ 310 'iperf_server_address'] = wputils.get_server_address( 311 self.remote_server, self.dut_ip, '255.255.255.0') 312 313 def run_throughput_stability_test(self, testcase_params): 314 """Main function to test throughput stability. 315 316 The function sets up the AP in the correct channel and mode 317 configuration and runs an iperf test to measure throughput. 318 319 Args: 320 testcase_params: dict containing test specific parameters 321 Returns: 322 test_result: dict containing test result and meta data 323 """ 324 # Run test and log result 325 # Start iperf session 326 self.log.info('Starting iperf test.') 327 test_result = collections.OrderedDict() 328 llstats_obj = wputils.LinkLayerStats(self.dut) 329 llstats_obj.update_stats() 330 if self.testbed_params['sniffer_enable'] and len( 331 self.testclass_results) % self.sniffer_subsampling == 0: 332 self.sniffer.start_capture( 333 network=testcase_params['test_network'], 334 chan=testcase_params['channel'], 335 bw=testcase_params['bandwidth'], 336 duration=self.testclass_params['iperf_duration'] / 5) 337 self.iperf_server.start(tag=str(testcase_params['atten_level'])) 338 current_rssi = wputils.get_connected_rssi_nb( 339 dut=self.dut, 340 num_measurements=self.testclass_params['iperf_duration'] - 1, 341 polling_frequency=1, 342 first_measurement_delay=1, 343 disconnect_warning=1, 344 ignore_samples=1) 345 client_output_path = self.iperf_client.start( 346 testcase_params['iperf_server_address'], 347 testcase_params['iperf_args'], str(testcase_params['atten_level']), 348 self.testclass_params['iperf_duration'] + TEST_TIMEOUT) 349 current_rssi = current_rssi.result() 350 server_output_path = self.iperf_server.stop() 351 # Stop sniffer 352 if self.testbed_params['sniffer_enable'] and len( 353 self.testclass_results) % self.sniffer_subsampling == 0: 354 self.sniffer.stop_capture() 355 # Set attenuator to 0 dB 356 for attenuator in self.attenuators: 357 attenuator.set_atten(0) 358 # Parse and log result 359 if testcase_params['use_client_output']: 360 iperf_file = client_output_path 361 else: 362 iperf_file = server_output_path 363 try: 364 iperf_result = ipf.IPerfResult(iperf_file) 365 except: 366 iperf_result = ipf.IPerfResult('{}') #empty iperf result 367 self.log.warning('Cannot get iperf result.') 368 if iperf_result.instantaneous_rates: 369 instantaneous_rates_Mbps = [ 370 rate * 8 * (1.024**2) 371 for rate in iperf_result.instantaneous_rates[ 372 self.testclass_params['iperf_ignored_interval']:-1] 373 ] 374 tput_standard_deviation = iperf_result.get_std_deviation( 375 self.testclass_params['iperf_ignored_interval']) * 8 376 else: 377 instantaneous_rates_Mbps = [float('nan')] 378 tput_standard_deviation = float('nan') 379 test_result['iperf_summary'] = { 380 'instantaneous_rates': 381 instantaneous_rates_Mbps, 382 'avg_throughput': 383 numpy.mean(instantaneous_rates_Mbps), 384 'std_deviation': 385 tput_standard_deviation, 386 'min_throughput': 387 min(instantaneous_rates_Mbps), 388 'std_dev_percent': 389 (tput_standard_deviation / numpy.mean(instantaneous_rates_Mbps)) * 390 100 391 } 392 llstats_obj.update_stats() 393 curr_llstats = llstats_obj.llstats_incremental.copy() 394 test_result['testcase_params'] = testcase_params.copy() 395 test_result['ap_settings'] = self.access_point.ap_settings.copy() 396 test_result['attenuation'] = testcase_params['atten_level'] 397 test_result['iperf_result'] = iperf_result 398 test_result['rssi_result'] = current_rssi 399 test_result['llstats'] = curr_llstats 400 401 llstats = ( 402 'TX MCS = {0} ({1:.1f}%). ' 403 'RX MCS = {2} ({3:.1f}%)'.format( 404 test_result['llstats']['summary']['common_tx_mcs'], 405 test_result['llstats']['summary']['common_tx_mcs_freq'] * 100, 406 test_result['llstats']['summary']['common_rx_mcs'], 407 test_result['llstats']['summary']['common_rx_mcs_freq'] * 100)) 408 409 test_message = ( 410 'Atten: {0:.2f}dB, RSSI: {1:.2f}dB. ' 411 'Throughput (Mean: {2:.2f}, Std. Dev:{3:.2f}%, Min: {4:.2f} Mbps).' 412 'LLStats : {5}'.format( 413 test_result['attenuation'], 414 test_result['rssi_result']['signal_poll_rssi']['mean'], 415 test_result['iperf_summary']['avg_throughput'], 416 test_result['iperf_summary']['std_dev_percent'], 417 test_result['iperf_summary']['min_throughput'], llstats)) 418 419 self.log.info(test_message) 420 421 self.testclass_results.append(test_result) 422 return test_result 423 424 def get_target_atten(self, testcase_params): 425 """Function gets attenuation used for test 426 427 The function fetches the attenuation at which the test should be 428 performed. 429 430 Args: 431 testcase_params: dict containing test specific parameters 432 Returns: 433 test_atten: target attenuation for test 434 """ 435 # Get attenuation from reference test if it has been run 436 ref_test_fields = ['channel', 'mode', 'signal_level'] 437 test_id = wputils.extract_sub_dict(testcase_params, ref_test_fields) 438 test_id = tuple(test_id.items()) 439 if test_id in self.ref_attenuations: 440 return self.ref_attenuations[test_id] 441 442 # Get attenuation for target RSSI 443 if testcase_params['signal_level'] == 'low': 444 target_rssi = self.testclass_params['low_throughput_rssi_target'] 445 else: 446 target_rssi = self.testclass_params['high_throughput_rssi_target'] 447 target_atten = wputils.get_atten_for_target_rssi( 448 target_rssi, self.attenuators, self.dut, self.remote_server) 449 450 self.ref_attenuations[test_id] = target_atten 451 return self.ref_attenuations[test_id] 452 453 def compile_test_params(self, testcase_params): 454 """Function that completes setting the test case parameters.""" 455 # Check if test should be skipped based on parameters. 456 wputils.check_skip_conditions(testcase_params, self.dut, 457 self.access_point, 458 getattr(self, 'ota_chamber', None)) 459 460 band = self.access_point.band_lookup_by_channel( 461 testcase_params['channel']) 462 testcase_params['test_network'] = self.main_network[band] 463 464 if testcase_params['traffic_type'] == 'TCP': 465 testcase_params['iperf_socket_size'] = self.testclass_params.get( 466 'tcp_socket_size', None) 467 testcase_params['iperf_processes'] = self.testclass_params.get( 468 'tcp_processes', 1) 469 elif testcase_params['traffic_type'] == 'UDP': 470 testcase_params['iperf_socket_size'] = self.testclass_params.get( 471 'udp_socket_size', None) 472 testcase_params['iperf_processes'] = self.testclass_params.get( 473 'udp_processes', 1) 474 if (testcase_params['traffic_direction'] == 'DL' 475 and not isinstance(self.iperf_server, ipf.IPerfServerOverAdb) 476 ) or (testcase_params['traffic_direction'] == 'UL' 477 and isinstance(self.iperf_server, ipf.IPerfServerOverAdb)): 478 testcase_params['iperf_args'] = wputils.get_iperf_arg_string( 479 duration=self.testclass_params['iperf_duration'], 480 reverse_direction=1, 481 traffic_type=testcase_params['traffic_type'], 482 socket_size=testcase_params['iperf_socket_size'], 483 num_processes=testcase_params['iperf_processes'], 484 udp_throughput=self.testclass_params['UDP_rates'][ 485 testcase_params['mode']]) 486 testcase_params['use_client_output'] = True 487 else: 488 testcase_params['iperf_args'] = wputils.get_iperf_arg_string( 489 duration=self.testclass_params['iperf_duration'], 490 reverse_direction=0, 491 traffic_type=testcase_params['traffic_type'], 492 socket_size=testcase_params['iperf_socket_size'], 493 num_processes=testcase_params['iperf_processes'], 494 udp_throughput=self.testclass_params['UDP_rates'][ 495 testcase_params['mode']]) 496 testcase_params['use_client_output'] = False 497 498 return testcase_params 499 500 def _test_throughput_stability(self, testcase_params): 501 """ Function that gets called for each test case 502 503 The function gets called in each test case. The function customizes 504 the test based on the test name of the test that called it 505 506 Args: 507 testcase_params: dict containing test specific parameters 508 """ 509 testcase_params = self.compile_test_params(testcase_params) 510 self.setup_throughput_stability_test(testcase_params) 511 test_result = self.run_throughput_stability_test(testcase_params) 512 test_result_postprocessed = self.post_process_results(test_result) 513 self.pass_fail_check(test_result_postprocessed) 514 515 516# Over-the air version of ping tests 517class WifiOtaThroughputStabilityTest(WifiThroughputStabilityTest): 518 """Class to test over-the-air ping 519 520 This class tests WiFi ping performance in an OTA chamber. It enables 521 setting turntable orientation and other chamber parameters to study 522 performance in varying channel conditions 523 """ 524 525 def __init__(self, controllers): 526 base_test.BaseTestClass.__init__(self, controllers) 527 # Define metrics to be uploaded to BlackBox 528 self.testcase_metric_logger = ( 529 BlackboxMappedMetricLogger.for_test_case()) 530 self.testclass_metric_logger = ( 531 BlackboxMappedMetricLogger.for_test_class()) 532 self.publish_testcase_metrics = False 533 534 def setup_class(self): 535 WifiThroughputStabilityTest.setup_class(self) 536 self.ota_chamber = ota_chamber.create( 537 self.user_params['OTAChamber'])[0] 538 539 def teardown_class(self): 540 WifiThroughputStabilityTest.teardown_class(self) 541 self.ota_chamber.reset_chamber() 542 self.process_testclass_results() 543 544 def extract_test_id(self, testcase_params, id_fields): 545 test_id = collections.OrderedDict( 546 (param, testcase_params[param]) for param in id_fields) 547 return test_id 548 549 def process_testclass_results(self): 550 """Saves all test results to enable comparison.""" 551 testclass_data = collections.OrderedDict() 552 for test in self.testclass_results: 553 current_params = test['testcase_params'] 554 channel_data = testclass_data.setdefault(current_params['channel'], 555 collections.OrderedDict()) 556 test_id = tuple( 557 self.extract_test_id(current_params, [ 558 'mode', 'traffic_type', 'traffic_direction', 'signal_level' 559 ]).items()) 560 test_data = channel_data.setdefault( 561 test_id, collections.OrderedDict(position=[], throughput=[])) 562 test_data['position'].append(current_params['position']) 563 test_data['throughput'].append( 564 test['iperf_summary']['avg_throughput']) 565 566 chamber_mode = self.testclass_results[0]['testcase_params'][ 567 'chamber_mode'] 568 if chamber_mode == 'orientation': 569 x_label = 'Angle (deg)' 570 elif chamber_mode == 'stepped stirrers': 571 x_label = 'Position Index' 572 573 # Publish test class metrics 574 for channel, channel_data in testclass_data.items(): 575 for test_id, test_data in channel_data.items(): 576 test_id_dict = dict(test_id) 577 metric_tag = 'ota_summary_{}_{}_{}_ch{}_{}'.format( 578 test_id_dict['signal_level'], test_id_dict['traffic_type'], 579 test_id_dict['traffic_direction'], channel, 580 test_id_dict['mode']) 581 metric_name = metric_tag + '.avg_throughput' 582 metric_value = numpy.nanmean(test_data['throughput']) 583 self.testclass_metric_logger.add_metric( 584 metric_name, metric_value) 585 metric_name = metric_tag + '.min_throughput' 586 metric_value = min(test_data['throughput']) 587 self.testclass_metric_logger.add_metric( 588 metric_name, metric_value) 589 590 # Plot test class results 591 plots = [] 592 for channel, channel_data in testclass_data.items(): 593 current_plot = BokehFigure( 594 title='Channel {} - Rate vs. Position'.format(channel), 595 x_label=x_label, 596 primary_y_label='Rate (Mbps)', 597 ) 598 for test_id, test_data in channel_data.items(): 599 test_id_dict = dict(test_id) 600 legend = '{}, {} {}, {} RSSI'.format( 601 test_id_dict['mode'], test_id_dict['traffic_type'], 602 test_id_dict['traffic_direction'], 603 test_id_dict['signal_level']) 604 current_plot.add_line(test_data['position'], 605 test_data['throughput'], legend) 606 current_plot.generate_figure() 607 plots.append(current_plot) 608 current_context = context.get_current_context().get_full_output_path() 609 plot_file_path = os.path.join(current_context, 'results.html') 610 BokehFigure.save_figures(plots, plot_file_path) 611 612 def setup_throughput_stability_test(self, testcase_params): 613 WifiThroughputStabilityTest.setup_throughput_stability_test( 614 self, testcase_params) 615 # Setup turntable 616 if testcase_params['chamber_mode'] == 'orientation': 617 self.ota_chamber.set_orientation(testcase_params['position']) 618 elif testcase_params['chamber_mode'] == 'stepped stirrers': 619 self.ota_chamber.step_stirrers(testcase_params['total_positions']) 620 621 def get_target_atten(self, testcase_params): 622 band = wputils.CHANNEL_TO_BAND_MAP[testcase_params['channel']] 623 if testcase_params['signal_level'] == 'high': 624 test_atten = self.testclass_params['ota_atten_levels'][band][0] 625 elif testcase_params['signal_level'] == 'low': 626 test_atten = self.testclass_params['ota_atten_levels'][band][1] 627 return test_atten 628 629 def _test_throughput_stability_over_orientation(self, testcase_params): 630 """ Function that gets called for each test case 631 632 The function gets called in each test case. The function customizes 633 the test based on the test name of the test that called it 634 635 Args: 636 testcase_params: dict containing test specific parameters 637 """ 638 testcase_params = self.compile_test_params(testcase_params) 639 for position in testcase_params['positions']: 640 testcase_params['position'] = position 641 self.setup_throughput_stability_test(testcase_params) 642 test_result = self.run_throughput_stability_test(testcase_params) 643 self.post_process_results(test_result) 644 645 def generate_test_cases(self, channels, modes, traffic_types, 646 traffic_directions, signal_levels, chamber_mode, 647 positions): 648 allowed_configs = { 649 20: [ 650 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 36, 40, 44, 48, 64, 100, 651 116, 132, 140, 149, 153, 157, 161, '6g37', '6g117', '6g213' 652 ], 653 40: [36, 44, 100, 149, 157, '6g37', '6g117', '6g213'], 654 80: [36, 100, 149, '6g37', '6g117', '6g213'], 655 160: [36, '6g37', '6g117', '6g213'] 656 } 657 658 test_cases = [] 659 for channel, mode, signal_level, traffic_type, traffic_direction in itertools.product( 660 channels, modes, signal_levels, traffic_types, 661 traffic_directions): 662 bandwidth = int(''.join([x for x in mode if x.isdigit()])) 663 if channel not in allowed_configs[bandwidth]: 664 continue 665 testcase_params = collections.OrderedDict( 666 channel=channel, 667 mode=mode, 668 bandwidth=bandwidth, 669 traffic_type=traffic_type, 670 traffic_direction=traffic_direction, 671 signal_level=signal_level, 672 chamber_mode=chamber_mode, 673 total_positions=len(positions), 674 positions=positions) 675 testcase_name = ('test_tput_stability' 676 '_{}_{}_{}_ch{}_{}'.format( 677 signal_level, traffic_type, traffic_direction, 678 channel, mode)) 679 setattr( 680 self, testcase_name, 681 partial(self._test_throughput_stability_over_orientation, 682 testcase_params)) 683 test_cases.append(testcase_name) 684 return test_cases 685 686 687class WifiOtaThroughputStability_TenDegree_Test(WifiOtaThroughputStabilityTest 688 ): 689 690 def __init__(self, controllers): 691 WifiOtaThroughputStabilityTest.__init__(self, controllers) 692 self.tests = self.generate_test_cases([6, 36, 149, '6g37'], 693 ['bw20', 'bw80', 'bw160'], 694 ['TCP'], ['DL', 'UL'], 695 ['high', 'low'], 'orientation', 696 list(range(0, 360, 10))) 697 698 def setup_class(self): 699 WifiOtaThroughputStabilityTest.setup_class(self) 700 self.sniffer_subsampling = 6 701 702 703class WifiOtaThroughputStability_45Degree_Test(WifiOtaThroughputStabilityTest): 704 705 def __init__(self, controllers): 706 WifiOtaThroughputStabilityTest.__init__(self, controllers) 707 self.tests = self.generate_test_cases([6, 36, 149, '6g37'], 708 ['bw20', 'bw80', 'bw160'], 709 ['TCP'], ['DL', 'UL'], 710 ['high', 'low'], 'orientation', 711 list(range(0, 360, 45))) 712 713 714class WifiOtaThroughputStability_SteppedStirrers_Test( 715 WifiOtaThroughputStabilityTest): 716 717 def __init__(self, controllers): 718 WifiOtaThroughputStabilityTest.__init__(self, controllers) 719 self.tests = self.generate_test_cases([6, 36, 149, '6g37'], 720 ['bw20', 'bw80', 'bw160'], 721 ['TCP'], ['DL', 'UL'], 722 ['high', 'low'], 723 'stepped stirrers', 724 list(range(100))) 725 726 def setup_class(self): 727 WifiOtaThroughputStabilityTest.setup_class(self) 728 self.sniffer_subsampling = 10 729