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