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