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