1#!/usr/bin/env python3.4 2# 3# Copyright 2019 - 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 json 19import math 20import os 21import time 22from acts import asserts 23from acts import base_test 24from acts import context 25from acts import utils 26from acts.controllers import iperf_server as ipf 27from acts.controllers.utils_lib import ssh 28from acts.metrics.loggers.blackbox import BlackboxMappedMetricLogger 29from acts.test_utils.wifi import wifi_performance_test_utils as wputils 30from acts.test_utils.wifi import wifi_retail_ap as retail_ap 31from acts.test_utils.wifi import wifi_test_utils as wutils 32 33SHORT_SLEEP = 1 34MED_SLEEP = 5 35TRAFFIC_GAP_THRESH = 0.5 36IPERF_INTERVAL = 0.25 37 38 39class WifiRoamingPerformanceTest(base_test.BaseTestClass): 40 """Class for ping-based Wifi performance tests. 41 42 This class implements WiFi ping performance tests such as range and RTT. 43 The class setups up the AP in the desired configurations, configures 44 and connects the phone to the AP, and runs For an example config file to 45 run this test class see example_connectivity_performance_ap_sta.json. 46 """ 47 def __init__(self, controllers): 48 base_test.BaseTestClass.__init__(self, controllers) 49 self.testcase_metric_logger = ( 50 BlackboxMappedMetricLogger.for_test_case()) 51 self.testclass_metric_logger = ( 52 BlackboxMappedMetricLogger.for_test_class()) 53 self.publish_testcase_metrics = True 54 55 def setup_class(self): 56 """Initializes common test hardware and parameters. 57 58 This function initializes hardwares and compiles parameters that are 59 common to all tests in this class. 60 """ 61 self.dut = self.android_devices[-1] 62 req_params = [ 63 'RetailAccessPoints', 'roaming_test_params', 'testbed_params' 64 ] 65 opt_params = ['main_network', 'RemoteServer'] 66 self.unpack_userparams(req_params, opt_params) 67 self.testclass_params = self.roaming_test_params 68 self.num_atten = self.attenuators[0].instrument.num_atten 69 self.remote_server = ssh.connection.SshConnection( 70 ssh.settings.from_config(self.RemoteServer[0]['ssh_config'])) 71 self.remote_server.setup_master_ssh() 72 self.iperf_server = self.iperf_servers[0] 73 self.iperf_client = self.iperf_clients[0] 74 self.access_point = retail_ap.create(self.RetailAccessPoints)[0] 75 self.log.info('Access Point Configuration: {}'.format( 76 self.access_point.ap_settings)) 77 78 # Get RF connection map 79 self.log.info("Getting RF connection map.") 80 wutils.wifi_toggle_state(self.dut, True) 81 self.rf_map_by_network, self.rf_map_by_atten = ( 82 wputils.get_full_rf_connection_map(self.attenuators, self.dut, 83 self.remote_server, 84 self.main_network)) 85 self.log.info("RF Map (by Network): {}".format(self.rf_map_by_network)) 86 self.log.info("RF Map (by Atten): {}".format(self.rf_map_by_atten)) 87 88 #Turn WiFi ON 89 if self.testclass_params.get('airplane_mode', 1): 90 self.log.info('Turning on airplane mode.') 91 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 92 "Can not turn on airplane mode.") 93 wutils.wifi_toggle_state(self.dut, True) 94 95 def pass_fail_traffic_continuity(self, result): 96 """Pass fail check for traffic continuity 97 98 Currently, the function only reports test results and implicitly passes 99 the test. A pass fail criterion is current being researched. 100 101 Args: 102 result: dict containing test results 103 """ 104 self.log.info('Detected {} roam transitions:'.format( 105 len(result['roam_transitions']))) 106 for event in result['roam_transitions']: 107 self.log.info('Roam: {} -> {})'.format(event[0], event[1])) 108 self.log.info('Roam transition statistics: {}'.format( 109 result['roam_counts'])) 110 111 formatted_traffic_gaps = [ 112 round(gap, 2) for gap in result['traffic_disruption'] 113 ] 114 self.log.info('Detected {} traffic gaps of duration: {}'.format( 115 len(result['traffic_disruption']), formatted_traffic_gaps)) 116 117 if result['total_roams'] > 0: 118 disruption_percentage = (len(result['traffic_disruption']) / 119 result['total_roams']) * 100 120 max_disruption = max(result['traffic_disruption']) 121 else: 122 disruption_percentage = 0 123 max_disruption = 0 124 self.testcase_metric_logger.add_metric('disruption_percentage', 125 disruption_percentage) 126 self.testcase_metric_logger.add_metric('max_disruption', 127 max_disruption) 128 129 if disruption_percentage == 0: 130 asserts.explicit_pass('Test passed. No traffic disruptions found.') 131 elif max_disruption > self.testclass_params[ 132 'traffic_disruption_threshold']: 133 asserts.fail('Test failed. Disruption Percentage = {}%. ' 134 'Max traffic disruption: {}s.'.format( 135 disruption_percentage, max_disruption)) 136 else: 137 asserts.explicit_pass('Test failed. Disruption Percentage = {}%. ' 138 'Max traffic disruption: {}s.'.format( 139 disruption_percentage, max_disruption)) 140 141 def pass_fail_roaming_consistency(self, results_dict): 142 """Function to evaluate roaming consistency results. 143 144 The function looks for the roams recorded in multiple runs of the same 145 attenuation waveform and checks that the DUT reliably roams to the 146 same network 147 148 Args: 149 results_dict: dict containing consistency test results 150 """ 151 test_fail = False 152 for secondary_atten, roam_stats in results_dict['roam_stats'].items(): 153 total_roams = sum(list(roam_stats.values())) 154 common_roam = max(roam_stats.keys(), key=(lambda k: roam_stats[k])) 155 common_roam_frequency = roam_stats[common_roam] / total_roams 156 self.log.info( 157 '{}dB secondary atten. Most common roam: {}. Frequency: {}'. 158 format(secondary_atten, common_roam, common_roam_frequency)) 159 if common_roam_frequency < self.testclass_params[ 160 'consistency_threshold']: 161 test_fail = True 162 self.log.info('Unstable Roams at {}dB secondary att'.format( 163 secondary_atten)) 164 self.testcase_metric_logger.add_metric('common_roam_frequency', 165 common_roam_frequency) 166 if test_fail: 167 asserts.fail('Incosistent roaming detected.') 168 else: 169 asserts.explicit_pass('Consistent roaming at all levels.') 170 171 def process_traffic_continuity_results(self, testcase_params, result): 172 """Function to process traffic results. 173 174 The function looks for traffic gaps during a roaming test 175 176 Args: 177 testcase_params: dict containing all test results and meta data 178 results_dict: dict containing consistency test results 179 """ 180 self.detect_roam_events(result) 181 current_context = context.get_current_context().get_full_output_path() 182 plot_file_path = os.path.join(current_context, 183 self.current_test_name + '.html') 184 185 if 'ping' in self.current_test_name: 186 self.detect_ping_gaps(result) 187 self.plot_ping_result(testcase_params, 188 result, 189 output_file_path=plot_file_path) 190 elif 'iperf' in self.current_test_name: 191 self.detect_iperf_gaps(result) 192 self.plot_iperf_result(testcase_params, 193 result, 194 output_file_path=plot_file_path) 195 196 results_file_path = os.path.join(current_context, 197 self.current_test_name + '.json') 198 with open(results_file_path, 'w') as results_file: 199 json.dump(wputils.serialize_dict(result), results_file, indent=4) 200 201 def process_consistency_results(self, testcase_params, results_dict): 202 """Function to process roaming consistency results. 203 204 The function looks compiles the test of roams recorded in consistency 205 tests and plots results for easy visualization. 206 207 Args: 208 testcase_params: dict containing all test results and meta data 209 results_dict: dict containing consistency test results 210 """ 211 # make figure placeholder and get relevant functions 212 if 'ping' in self.current_test_name: 213 detect_gaps = self.detect_ping_gaps 214 plot_result = self.plot_ping_result 215 primary_y_axis = 'RTT (ms)' 216 elif 'iperf' in self.current_test_name: 217 detect_gaps = self.detect_iperf_gaps 218 plot_result = self.plot_iperf_result 219 primary_y_axis = 'Throughput (Mbps)' 220 # loop over results 221 roam_stats = collections.OrderedDict() 222 current_context = context.get_current_context().get_full_output_path() 223 for secondary_atten, results_list in results_dict.items(): 224 figure = wputils.BokehFigure(title=self.current_test_name, 225 x_label='Time (ms)', 226 primary_y_label=primary_y_axis, 227 secondary_y_label='RSSI (dBm)') 228 roam_stats[secondary_atten] = collections.OrderedDict() 229 for result in results_list: 230 self.detect_roam_events(result) 231 for roam_transition, count in result['roam_counts'].items(): 232 roam_stats[secondary_atten][ 233 roam_transition] = roam_stats[secondary_atten].get( 234 roam_transition, 0) + count 235 detect_gaps(result) 236 plot_result(testcase_params, result, figure=figure) 237 # save plot 238 plot_file_name = (self.current_test_name + '_' + 239 str(secondary_atten) + '.html') 240 241 plot_file_path = os.path.join(current_context, plot_file_name) 242 figure.save_figure(plot_file_path) 243 results_dict['roam_stats'] = roam_stats 244 245 results_file_path = os.path.join(current_context, 246 self.current_test_name + '.json') 247 with open(results_file_path, 'w') as results_file: 248 json.dump(wputils.serialize_dict(result), results_file, indent=4) 249 250 def detect_roam_events(self, result): 251 """Function to process roaming results. 252 253 The function detects roams by looking at changes in BSSID and compiles 254 meta data about each roam, e.g., RSSI before and after a roam. The 255 function then calls the relevant method to process traffic results and 256 report traffic disruptions. 257 258 Args: 259 testcase_params: dict containing AP and other test params 260 result: dict containing test results 261 """ 262 roam_events = [ 263 (idx, idx + 1) 264 for idx in range(len(result['rssi_result']['bssid']) - 1) 265 if result['rssi_result']['bssid'][idx] != result['rssi_result'] 266 ['bssid'][idx + 1] 267 ] 268 269 def ignore_entry(vals): 270 for val in vals: 271 if val in {0} or math.isnan(val): 272 return True 273 return False 274 275 for roam_idx, roam_event in enumerate(roam_events): 276 # Find true roam start by scanning earlier samples for valid data 277 while ignore_entry([ 278 result['rssi_result']['frequency'][roam_event[0]], 279 result['rssi_result']['signal_poll_rssi']['data'][ 280 roam_event[0]] 281 ]): 282 roam_event = (roam_event[0] - 1, roam_event[1]) 283 roam_events[roam_idx] = roam_event 284 # Find true roam end by scanning later samples for valid data 285 while ignore_entry([ 286 result['rssi_result']['frequency'][roam_event[1]], 287 result['rssi_result']['signal_poll_rssi']['data'][ 288 roam_event[1]] 289 ]): 290 roam_event = (roam_event[0], roam_event[1] + 1) 291 roam_events[roam_idx] = roam_event 292 293 roam_events = list(set(roam_events)) 294 roam_events.sort(key=lambda event_tuple: event_tuple[1]) 295 roam_transitions = [] 296 roam_counts = {} 297 total_roams = 0 298 for event in roam_events: 299 from_bssid = next( 300 key for key, value in self.main_network.items() 301 if value['BSSID'] == result['rssi_result']['bssid'][event[0]]) 302 to_bssid = next( 303 key for key, value in self.main_network.items() 304 if value['BSSID'] == result['rssi_result']['bssid'][event[1]]) 305 curr_bssid_transition = (from_bssid, to_bssid) 306 curr_roam_transition = ( 307 (from_bssid, 308 result['rssi_result']['signal_poll_rssi']['data'][event[0]]), 309 (to_bssid, 310 result['rssi_result']['signal_poll_rssi']['data'][event[1]])) 311 roam_transitions.append(curr_roam_transition) 312 roam_counts[curr_bssid_transition] = roam_counts.get( 313 curr_bssid_transition, 0) + 1 314 total_roams = total_roams + 1 315 result['roam_events'] = roam_events 316 result['roam_transitions'] = roam_transitions 317 result['roam_counts'] = roam_counts 318 result['total_roams'] = total_roams 319 320 def detect_ping_gaps(self, result): 321 """Function to process ping results. 322 323 The function looks for gaps in iperf traffic and reports them as 324 disruptions due to roams. 325 326 Args: 327 result: dict containing test results 328 """ 329 traffic_disruption = [ 330 x for x in result['ping_result']['ping_interarrivals'] 331 if x > TRAFFIC_GAP_THRESH 332 ] 333 result['traffic_disruption'] = traffic_disruption 334 335 def detect_iperf_gaps(self, result): 336 """Function to process iperf results. 337 338 The function looks for gaps in iperf traffic and reports them as 339 disruptions due to roams. 340 341 Args: 342 result: dict containing test results 343 """ 344 tput_thresholding = [tput < 1 for tput in result['throughput']] 345 window_size = int(TRAFFIC_GAP_THRESH / IPERF_INTERVAL) 346 tput_thresholding = [ 347 any(tput_thresholding[max(0, idx - window_size):idx]) 348 for idx in range(1, 349 len(tput_thresholding) + 1) 350 ] 351 352 traffic_disruption = [] 353 current_disruption = 1 - window_size 354 for tput_low in tput_thresholding: 355 if tput_low: 356 current_disruption += 1 357 elif current_disruption > window_size: 358 traffic_disruption.append(current_disruption * IPERF_INTERVAL) 359 current_disruption = 1 - window_size 360 else: 361 current_disruption = 1 - window_size 362 result['traffic_disruption'] = traffic_disruption 363 364 def plot_ping_result(self, 365 testcase_params, 366 result, 367 figure=None, 368 output_file_path=None): 369 """Function to plot ping results. 370 371 The function plots ping RTTs along with RSSI over time during a roaming 372 test. 373 374 Args: 375 testcase_params: dict containing all test params 376 result: dict containing test results 377 figure: optional bokeh figure object to add current plot to 378 output_file_path: optional path to output file 379 """ 380 if not figure: 381 figure = wputils.BokehFigure(title=self.current_test_name, 382 x_label='Time (ms)', 383 primary_y_label='RTT (ms)', 384 secondary_y_label='RSSI (dBm)') 385 figure.add_line(x_data=result['ping_result']['time_stamp'], 386 y_data=result['ping_result']['rtt'], 387 legend='Ping RTT', 388 width=1) 389 figure.add_line( 390 x_data=result['rssi_result']['time_stamp'], 391 y_data=result['rssi_result']['signal_poll_rssi']['data'], 392 legend='RSSI', 393 y_axis='secondary') 394 figure.generate_figure(output_file_path) 395 396 def plot_iperf_result(self, 397 testcase_params, 398 result, 399 figure=None, 400 output_file_path=None): 401 """Function to plot iperf results. 402 403 The function plots iperf throughput and RSSI over time during a roaming 404 test. 405 406 Args: 407 testcase_params: dict containing all test params 408 result: dict containing test results 409 figure: optional bokeh figure object to add current plot to 410 output_file_path: optional path to output file 411 """ 412 if not figure: 413 figure = wputils.BokehFigure(title=self.current_test_name, 414 x_label='Time (s)', 415 primary_y_label='Throughput (Mbps)', 416 secondary_y_label='RSSI (dBm)') 417 iperf_time_stamps = [ 418 idx * IPERF_INTERVAL for idx in range(len(result['throughput'])) 419 ] 420 figure.add_line(iperf_time_stamps, 421 result['throughput'], 422 'Throughput', 423 width=1) 424 figure.add_line(result['rssi_result']['time_stamp'], 425 result['rssi_result']['signal_poll_rssi']['data'], 426 'RSSI', 427 y_axis='secondary') 428 429 figure.generate_figure(output_file_path) 430 431 def setup_ap(self, testcase_params): 432 """Sets up the AP and attenuator to the test configuration. 433 434 Args: 435 testcase_params: dict containing AP and other test params 436 """ 437 (primary_net_id, 438 primary_net_config) = next(net for net in self.main_network.items() 439 if net[1]['roaming_label'] == 'primary') 440 for idx, atten in enumerate(self.attenuators): 441 nets_on_port = [ 442 item["network"] for item in self.rf_map_by_atten[idx] 443 ] 444 if primary_net_id in nets_on_port: 445 atten.set_atten(0) 446 else: 447 atten.set_atten(atten.instrument.max_atten) 448 449 def setup_dut(self, testcase_params): 450 """Sets up the DUT in the configuration required by the test. 451 452 Args: 453 testcase_params: dict containing AP and other test params 454 """ 455 # Check battery level before test 456 if not wputils.health_check(self.dut, 10): 457 asserts.skip('Battery level too low. Skipping test.') 458 wutils.reset_wifi(self.dut) 459 wutils.set_wifi_country_code(self.dut, 460 self.testclass_params['country_code']) 461 (primary_net_id, 462 primary_net_config) = next(net for net in self.main_network.items() 463 if net[1]['roaming_label'] == 'primary') 464 network = primary_net_config.copy() 465 network.pop('BSSID', None) 466 self.dut.droid.wifiSetEnableAutoJoinWhenAssociated(1) 467 wutils.wifi_connect(self.dut, 468 network, 469 num_of_tries=5, 470 check_connectivity=False) 471 self.dut.droid.wifiSetEnableAutoJoinWhenAssociated(1) 472 self.dut_ip = self.dut.droid.connectivityGetIPv4Addresses('wlan0')[0] 473 if testcase_params['screen_on']: 474 self.dut.wakeup_screen() 475 self.dut.droid.wakeLockAcquireBright() 476 time.sleep(MED_SLEEP) 477 478 def setup_roaming_test(self, testcase_params): 479 """Function to set up roaming test.""" 480 self.setup_ap(testcase_params) 481 self.setup_dut(testcase_params) 482 483 def run_ping_test(self, testcase_params): 484 """Main function for ping roaming tests. 485 486 Args: 487 testcase_params: dict including all test params encoded in test 488 name 489 Returns: 490 dict containing all test results and meta data 491 """ 492 self.log.info('Starting ping test.') 493 ping_future = wputils.get_ping_stats_nb( 494 self.remote_server, self.dut_ip, 495 testcase_params['atten_waveforms']['length'], 496 testcase_params['ping_interval'], 64) 497 rssi_future = wputils.get_connected_rssi_nb( 498 self.dut, 499 int(testcase_params['atten_waveforms']['length'] / 500 testcase_params['rssi_polling_frequency']), 501 testcase_params['rssi_polling_frequency']) 502 self.run_attenuation_waveform(testcase_params) 503 return { 504 'ping_result': ping_future.result().as_dict(), 505 'rssi_result': rssi_future.result(), 506 'ap_settings': self.access_point.ap_settings, 507 } 508 509 def run_iperf_test(self, testcase_params): 510 """Main function for iperf roaming tests. 511 512 Args: 513 testcase_params: dict including all test params encoded in test 514 name 515 Returns: 516 result: dict containing all test results and meta data 517 """ 518 self.log.info('Starting iperf test.') 519 self.iperf_server.start(extra_args='-i {}'.format(IPERF_INTERVAL)) 520 self.dut_ip = self.dut.droid.connectivityGetIPv4Addresses('wlan0')[0] 521 if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 522 iperf_server_address = self.dut_ip 523 else: 524 iperf_server_address = wputils.get_server_address( 525 self.remote_server, self.dut_ip, '255.255.255.0') 526 iperf_args = '-i {} -t {} -J'.format( 527 IPERF_INTERVAL, testcase_params['atten_waveforms']['length']) 528 if not isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 529 iperf_args = iperf_args + ' -R' 530 iperf_future = wputils.start_iperf_client_nb( 531 self.iperf_client, iperf_server_address, iperf_args, 0, 532 testcase_params['atten_waveforms']['length'] + MED_SLEEP) 533 rssi_future = wputils.get_connected_rssi_nb( 534 self.dut, 535 int(testcase_params['atten_waveforms']['length'] / 536 testcase_params['rssi_polling_frequency']), 537 testcase_params['rssi_polling_frequency']) 538 self.run_attenuation_waveform(testcase_params) 539 client_output_path = iperf_future.result() 540 server_output_path = self.iperf_server.stop() 541 if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 542 iperf_file = server_output_path 543 else: 544 iperf_file = client_output_path 545 iperf_result = ipf.IPerfResult(iperf_file) 546 instantaneous_rates = [ 547 rate * 8 * (1.024**2) for rate in iperf_result.instantaneous_rates 548 ] 549 return { 550 'throughput': instantaneous_rates, 551 'rssi_result': rssi_future.result(), 552 'ap_settings': self.access_point.ap_settings, 553 } 554 555 def run_attenuation_waveform(self, testcase_params, step_duration=1): 556 """Function that generates test params based on the test name. 557 558 Args: 559 testcase_params: dict including all test params encoded in test 560 name 561 step_duration: int representing number of seconds to dwell on each 562 atten level 563 """ 564 atten_waveforms = testcase_params['atten_waveforms'] 565 for atten_idx in range(atten_waveforms['length']): 566 start_time = time.time() 567 for network, atten_waveform in atten_waveforms.items(): 568 for idx, atten in enumerate(self.attenuators): 569 nets_on_port = [ 570 item["network"] for item in self.rf_map_by_atten[idx] 571 ] 572 if network in nets_on_port: 573 atten.set_atten(atten_waveform[atten_idx]) 574 measure_time = time.time() - start_time 575 time.sleep(step_duration - measure_time) 576 577 def compile_atten_waveforms(self, waveform_params): 578 """Function to compile all attenuation waveforms for roaming test. 579 580 Args: 581 waveform_params: list of dicts representing waveforms to generate 582 """ 583 atten_waveforms = {} 584 for network in list(waveform_params[0]): 585 atten_waveforms[network] = [] 586 587 for waveform in waveform_params: 588 for network, network_waveform in waveform.items(): 589 waveform_vector = self.gen_single_atten_waveform( 590 network_waveform) 591 atten_waveforms[network] += waveform_vector 592 593 waveform_lengths = { 594 len(atten_waveforms[network]) 595 for network in atten_waveforms.keys() 596 } 597 if len(waveform_lengths) != 1: 598 raise ValueError( 599 'Attenuation waveform length should be equal for all networks.' 600 ) 601 else: 602 atten_waveforms['length'] = waveform_lengths.pop() 603 return atten_waveforms 604 605 def gen_single_atten_waveform(self, waveform_params): 606 """Function to generate a single attenuation waveform for roaming test. 607 608 Args: 609 waveform_params: dict representing waveform to generate 610 """ 611 waveform_vector = [] 612 for section in range(len(waveform_params['atten_levels']) - 1): 613 section_limits = waveform_params['atten_levels'][section:section + 614 2] 615 up_down = (1 - 2 * (section_limits[1] < section_limits[0])) 616 temp_section = list( 617 range(section_limits[0], section_limits[1] + up_down, 618 up_down * waveform_params['step_size'])) 619 temp_section = [ 620 val for val in temp_section 621 for _ in range(waveform_params['step_duration']) 622 ] 623 waveform_vector += temp_section 624 waveform_vector *= waveform_params['repetitions'] 625 return waveform_vector 626 627 def parse_test_params(self, testcase_params): 628 """Function that generates test params based on the test name. 629 630 Args: 631 test_name: current test name 632 Returns: 633 testcase_params: dict including all test params encoded in test 634 name 635 """ 636 if testcase_params["waveform_type"] == 'smooth': 637 testcase_params[ 638 'roaming_waveforms_params'] = self.testclass_params[ 639 'smooth_roaming_waveforms'] 640 elif testcase_params["waveform_type"] == 'failover': 641 testcase_params[ 642 'roaming_waveforms_params'] = self.testclass_params[ 643 'failover_roaming_waveforms'] 644 elif testcase_params["waveform_type"] == 'consistency': 645 testcase_params[ 646 'roaming_waveforms_params'] = self.testclass_params[ 647 'consistency_waveforms'] 648 return testcase_params 649 650 def _test_traffic_continuity(self, testcase_params): 651 """Test function for traffic continuity""" 652 # Compile test parameters from config and test name 653 testcase_params = self.parse_test_params(testcase_params) 654 testcase_params.update(self.testclass_params) 655 testcase_params['atten_waveforms'] = self.compile_atten_waveforms( 656 testcase_params['roaming_waveforms_params']) 657 # Run traffic test 658 self.setup_roaming_test(testcase_params) 659 if testcase_params['traffic_type'] == 'iperf': 660 result = self.run_iperf_test(testcase_params) 661 elif testcase_params['traffic_type'] == 'ping': 662 result = self.run_ping_test(testcase_params) 663 # Postprocess results 664 self.process_traffic_continuity_results(testcase_params, result) 665 self.pass_fail_traffic_continuity(result) 666 667 def _test_roam_consistency(self, testcase_params): 668 """Test function for roaming consistency""" 669 testcase_params = self.parse_test_params(testcase_params) 670 testcase_params.update(self.testclass_params) 671 # Run traffic test 672 secondary_attens = range( 673 self.testclass_params['consistency_waveforms']['secondary_loop'] 674 ['atten_levels'][0], self.testclass_params['consistency_waveforms'] 675 ['secondary_loop']['atten_levels'][1], 676 self.testclass_params['consistency_waveforms']['secondary_loop'] 677 ['step_size']) 678 results = collections.OrderedDict() 679 for secondary_atten in secondary_attens: 680 primary_waveform = self.gen_single_atten_waveform( 681 testcase_params['roaming_waveforms_params']['primary_sweep']) 682 secondary_waveform_params = { 683 'atten_levels': [secondary_atten, secondary_atten], 684 'step_size': 1, 685 'step_duration': len(primary_waveform), 686 'repetitions': 1 687 } 688 secondary_waveform = self.gen_single_atten_waveform( 689 secondary_waveform_params) 690 testcase_params['atten_waveforms'] = { 691 'length': len(primary_waveform) 692 } 693 for network_key, network_info in self.main_network.items(): 694 if 'primary' in network_info['roaming_label']: 695 testcase_params['atten_waveforms'][ 696 network_key] = primary_waveform 697 else: 698 testcase_params['atten_waveforms'][ 699 network_key] = secondary_waveform 700 results[secondary_atten] = [] 701 for run in range(self.testclass_params['consistency_num_runs']): 702 self.setup_roaming_test(testcase_params) 703 results[secondary_atten].append( 704 self.run_ping_test(testcase_params)) 705 # Postprocess results 706 self.process_consistency_results(testcase_params, results) 707 self.pass_fail_roaming_consistency(results) 708 709 def test_consistency_roaming_screen_on_ping(self): 710 testcase_params = { 711 "waveform_type": "consistency", 712 "screen_on": 1, 713 "traffic_type": "ping" 714 } 715 self._test_roam_consistency(testcase_params) 716 717 def test_smooth_roaming_screen_on_ping_continuity(self): 718 testcase_params = { 719 "waveform_type": "smooth", 720 "screen_on": 1, 721 "traffic_type": "ping" 722 } 723 self._test_traffic_continuity(testcase_params) 724 725 def test_smooth_roaming_screen_on_iperf_continuity(self): 726 testcase_params = { 727 "waveform_type": "smooth", 728 "screen_on": 1, 729 "traffic_type": "iperf" 730 } 731 self._test_traffic_continuity(testcase_params) 732 733 def test_failover_roaming_screen_on_ping_continuity(self): 734 testcase_params = { 735 "waveform_type": "failover", 736 "screen_on": 1, 737 "traffic_type": "ping" 738 } 739 self._test_traffic_continuity(testcase_params) 740 741 def test_failover_roaming_screen_on_iperf_continuity(self): 742 testcase_params = { 743 "waveform_type": "failover", 744 "screen_on": 1, 745 "traffic_type": "iperf" 746 } 747 self._test_traffic_continuity(testcase_params) 748 749 def test_smooth_roaming_screen_off_ping_continuity(self): 750 testcase_params = { 751 "waveform_type": "smooth", 752 "screen_on": 0, 753 "traffic_type": "ping" 754 } 755 self._test_traffic_continuity(testcase_params) 756 757 def test_smooth_roaming_screen_off_iperf_continuity(self): 758 testcase_params = { 759 "waveform_type": "smooth", 760 "screen_on": 0, 761 "traffic_type": "iperf" 762 } 763 self._test_traffic_continuity(testcase_params) 764 765 def test_failover_roaming_screen_off_ping_continuity(self): 766 testcase_params = { 767 "waveform_type": "failover", 768 "screen_on": 0, 769 "traffic_type": "ping" 770 } 771 self._test_traffic_continuity(testcase_params) 772 773 def test_failover_roaming_screen_off_iperf_continuity(self): 774 testcase_params = { 775 "waveform_type": "failover", 776 "screen_on": 0, 777 "traffic_type": "iperf" 778 } 779 self._test_traffic_continuity(testcase_params) 780