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 copy 19import json 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.metrics.loggers.blackbox import BlackboxMappedMetricLogger 30from acts_contrib.test_utils.wifi import wifi_performance_test_utils as wputils 31from acts_contrib.test_utils.wifi import wifi_retail_ap as retail_ap 32from acts_contrib.test_utils.wifi import wifi_test_utils as wutils 33 34SHORT_SLEEP = 1 35MED_SLEEP = 5 36TRAFFIC_GAP_THRESH = 0.5 37IPERF_INTERVAL = 0.25 38 39 40class WifiRoamingPerformanceTest(base_test.BaseTestClass): 41 """Class for ping-based Wifi performance tests. 42 43 This class implements WiFi ping performance tests such as range and RTT. 44 The class setups up the AP in the desired configurations, configures 45 and connects the phone to the AP, and runs For an example config file to 46 run this test class see example_connectivity_performance_ap_sta.json. 47 """ 48 def __init__(self, controllers): 49 base_test.BaseTestClass.__init__(self, controllers) 50 self.testcase_metric_logger = ( 51 BlackboxMappedMetricLogger.for_test_case()) 52 self.testclass_metric_logger = ( 53 BlackboxMappedMetricLogger.for_test_class()) 54 self.publish_testcase_metrics = True 55 56 def setup_class(self): 57 """Initializes common test hardware and parameters. 58 59 This function initializes hardwares and compiles parameters that are 60 common to all tests in this class. 61 """ 62 self.dut = self.android_devices[-1] 63 req_params = [ 64 'RetailAccessPoints', 'roaming_test_params', 'testbed_params' 65 ] 66 opt_params = ['main_network', 'RemoteServer'] 67 self.unpack_userparams(req_params, opt_params) 68 self.testclass_params = self.roaming_test_params 69 self.num_atten = self.attenuators[0].instrument.num_atten 70 self.remote_server = ssh.connection.SshConnection( 71 ssh.settings.from_config(self.RemoteServer[0]['ssh_config'])) 72 self.remote_server.setup_master_ssh() 73 self.iperf_server = self.iperf_servers[0] 74 self.iperf_client = self.iperf_clients[0] 75 self.access_points = retail_ap.create(self.RetailAccessPoints) 76 77 # Get RF connection map 78 self.log.info("Getting RF connection map.") 79 wutils.wifi_toggle_state(self.dut, True) 80 self.rf_map_by_network, self.rf_map_by_atten = ( 81 wputils.get_full_rf_connection_map(self.attenuators, self.dut, 82 self.remote_server, 83 self.main_network, True)) 84 self.log.info("RF Map (by Network): {}".format(self.rf_map_by_network)) 85 self.log.info("RF Map (by Atten): {}".format(self.rf_map_by_atten)) 86 87 #Turn WiFi ON 88 if self.testclass_params.get('airplane_mode', 1): 89 self.log.info('Turning on airplane mode.') 90 asserts.assert_true(utils.force_airplane_mode(self.dut, True), 91 "Can not turn on airplane mode.") 92 wutils.wifi_toggle_state(self.dut, True) 93 94 def pass_fail_traffic_continuity(self, result): 95 """Pass fail check for traffic continuity 96 97 Currently, the function only reports test results and implicitly passes 98 the test. A pass fail criterion is current being researched. 99 100 Args: 101 result: dict containing test results 102 """ 103 self.log.info('Detected {} roam transitions:'.format( 104 len(result['roam_transitions']))) 105 for event in result['roam_transitions']: 106 self.log.info('Roam @ {0:.2f}s: {1} -> {2})'.format( 107 event[0], event[1], event[2])) 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 roam_candidates = copy.deepcopy(self.main_network) 299 roam_candidates['disconnected'] = {'BSSID': 'disconnected'} 300 for event in roam_events: 301 from_bssid = next( 302 key for key, value in roam_candidates.items() 303 if value['BSSID'] == result['rssi_result']['bssid'][event[0]]) 304 to_bssid = next( 305 key for key, value in roam_candidates.items() 306 if value['BSSID'] == result['rssi_result']['bssid'][event[1]]) 307 if from_bssid == to_bssid: 308 continue 309 curr_bssid_transition = (from_bssid, to_bssid) 310 curr_roam_transition = ( 311 result['rssi_result']['time_stamp'][event[0]], 312 (from_bssid, 313 result['rssi_result']['signal_poll_rssi']['data'][event[0]]), 314 (to_bssid, 315 result['rssi_result']['signal_poll_rssi']['data'][event[1]])) 316 roam_transitions.append(curr_roam_transition) 317 roam_counts[curr_bssid_transition] = roam_counts.get( 318 curr_bssid_transition, 0) + 1 319 total_roams = total_roams + 1 320 result['roam_events'] = roam_events 321 result['roam_transitions'] = roam_transitions 322 result['roam_counts'] = roam_counts 323 result['total_roams'] = total_roams 324 325 def detect_ping_gaps(self, result): 326 """Function to process ping results. 327 328 The function looks for gaps in iperf traffic and reports them as 329 disruptions due to roams. 330 331 Args: 332 result: dict containing test results 333 """ 334 traffic_disruption = [ 335 x for x in result['ping_result']['ping_interarrivals'] 336 if x > TRAFFIC_GAP_THRESH 337 ] 338 result['traffic_disruption'] = traffic_disruption 339 340 def detect_iperf_gaps(self, result): 341 """Function to process iperf results. 342 343 The function looks for gaps in iperf traffic and reports them as 344 disruptions due to roams. 345 346 Args: 347 result: dict containing test results 348 """ 349 tput_thresholding = [tput < 1 for tput in result['throughput']] 350 window_size = int(TRAFFIC_GAP_THRESH / IPERF_INTERVAL) 351 tput_thresholding = [ 352 any(tput_thresholding[max(0, idx - window_size):idx]) 353 for idx in range(1, 354 len(tput_thresholding) + 1) 355 ] 356 357 traffic_disruption = [] 358 current_disruption = 1 - window_size 359 for tput_low in tput_thresholding: 360 if tput_low: 361 current_disruption += 1 362 elif current_disruption > window_size: 363 traffic_disruption.append(current_disruption * IPERF_INTERVAL) 364 current_disruption = 1 - window_size 365 else: 366 current_disruption = 1 - window_size 367 result['traffic_disruption'] = traffic_disruption 368 369 def plot_ping_result(self, 370 testcase_params, 371 result, 372 figure=None, 373 output_file_path=None): 374 """Function to plot ping results. 375 376 The function plots ping RTTs along with RSSI over time during a roaming 377 test. 378 379 Args: 380 testcase_params: dict containing all test params 381 result: dict containing test results 382 figure: optional bokeh figure object to add current plot to 383 output_file_path: optional path to output file 384 """ 385 if not figure: 386 figure = wputils.BokehFigure(title=self.current_test_name, 387 x_label='Time (ms)', 388 primary_y_label='RTT (ms)', 389 secondary_y_label='RSSI (dBm)') 390 figure.add_line(x_data=result['ping_result']['time_stamp'], 391 y_data=result['ping_result']['rtt'], 392 legend='Ping RTT', 393 width=1) 394 figure.add_line( 395 x_data=result['rssi_result']['time_stamp'], 396 y_data=result['rssi_result']['signal_poll_rssi']['data'], 397 legend='RSSI', 398 y_axis='secondary') 399 try: 400 figure.generate_figure(output_file_path) 401 except: 402 pass 403 404 def plot_iperf_result(self, 405 testcase_params, 406 result, 407 figure=None, 408 output_file_path=None): 409 """Function to plot iperf results. 410 411 The function plots iperf throughput and RSSI over time during a roaming 412 test. 413 414 Args: 415 testcase_params: dict containing all test params 416 result: dict containing test results 417 figure: optional bokeh figure object to add current plot to 418 output_file_path: optional path to output file 419 """ 420 if not figure: 421 figure = wputils.BokehFigure(title=self.current_test_name, 422 x_label='Time (s)', 423 primary_y_label='Throughput (Mbps)', 424 secondary_y_label='RSSI (dBm)') 425 iperf_time_stamps = [ 426 idx * IPERF_INTERVAL for idx in range(len(result['throughput'])) 427 ] 428 figure.add_line(iperf_time_stamps, 429 result['throughput'], 430 'Throughput', 431 width=1) 432 figure.add_line(result['rssi_result']['time_stamp'], 433 result['rssi_result']['signal_poll_rssi']['data'], 434 'RSSI', 435 y_axis='secondary') 436 try: 437 figure.generate_figure(output_file_path) 438 except: 439 pass 440 441 def setup_ap(self, testcase_params): 442 """Sets up the AP and attenuator to the test configuration. 443 444 Args: 445 testcase_params: dict containing AP and other test params 446 """ 447 (primary_net_id, 448 primary_net_config) = next(net for net in self.main_network.items() 449 if net[1]['roaming_label'] == 'primary') 450 for idx, atten in enumerate(self.attenuators): 451 nets_on_port = [ 452 item["network"] for item in self.rf_map_by_atten[idx] 453 ] 454 if primary_net_id in nets_on_port: 455 atten.set_atten(0) 456 else: 457 atten.set_atten(atten.instrument.max_atten) 458 459 def setup_dut(self, testcase_params): 460 """Sets up the DUT in the configuration required by the test. 461 462 Args: 463 testcase_params: dict containing AP and other test params 464 """ 465 # Check battery level before test 466 if not wputils.health_check(self.dut, 10): 467 asserts.skip('Battery level too low. Skipping test.') 468 wutils.reset_wifi(self.dut) 469 wutils.set_wifi_country_code(self.dut, 470 self.testclass_params['country_code']) 471 (primary_net_id, 472 primary_net_config) = next(net for net in self.main_network.items() 473 if net[1]['roaming_label'] == 'primary') 474 network = primary_net_config.copy() 475 network.pop('BSSID', None) 476 self.dut.droid.wifiSetEnableAutoJoinWhenAssociated(1) 477 wutils.wifi_connect(self.dut, 478 network, 479 num_of_tries=5, 480 check_connectivity=True) 481 self.dut.droid.wifiSetEnableAutoJoinWhenAssociated(1) 482 self.dut_ip = self.dut.droid.connectivityGetIPv4Addresses('wlan0')[0] 483 if testcase_params['screen_on']: 484 self.dut.wakeup_screen() 485 self.dut.droid.wakeLockAcquireBright() 486 time.sleep(MED_SLEEP) 487 488 def setup_roaming_test(self, testcase_params): 489 """Function to set up roaming test.""" 490 self.setup_ap(testcase_params) 491 self.setup_dut(testcase_params) 492 493 def run_ping_test(self, testcase_params): 494 """Main function for ping roaming tests. 495 496 Args: 497 testcase_params: dict including all test params encoded in test 498 name 499 Returns: 500 dict containing all test results and meta data 501 """ 502 self.log.info('Starting ping test.') 503 if testcase_params.get('ping_to_dut', False): 504 ping_future = wputils.get_ping_stats_nb( 505 self.remote_server, self.dut_ip, 506 testcase_params['atten_waveforms']['length'], 507 testcase_params['ping_interval'], 64) 508 else: 509 if testcase_params.get('lan_traffic_only', False): 510 ping_address = wputils.get_server_address( 511 self.remote_server, self.dut_ip, '255.255.255.0') 512 else: 513 ping_address = wputils.get_server_address( 514 self.remote_server, self.dut_ip, 'public') 515 ping_future = wputils.get_ping_stats_nb( 516 self.dut, ping_address, 517 testcase_params['atten_waveforms']['length'], 518 testcase_params['ping_interval'], 64) 519 rssi_future = wputils.get_connected_rssi_nb( 520 self.dut, 521 int(testcase_params['atten_waveforms']['length'] / 522 testcase_params['rssi_polling_frequency']), 523 testcase_params['rssi_polling_frequency']) 524 self.run_attenuation_waveform(testcase_params) 525 return { 526 'ping_result': ping_future.result().as_dict(), 527 'rssi_result': rssi_future.result(), 528 'ap_settings': [ap.ap_settings for ap in self.access_points], 529 } 530 531 def run_iperf_test(self, testcase_params): 532 """Main function for iperf roaming tests. 533 534 Args: 535 testcase_params: dict including all test params encoded in test 536 name 537 Returns: 538 result: dict containing all test results and meta data 539 """ 540 self.log.info('Starting iperf test.') 541 self.iperf_server.start(extra_args='-i {}'.format(IPERF_INTERVAL)) 542 self.dut_ip = self.dut.droid.connectivityGetIPv4Addresses('wlan0')[0] 543 if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 544 iperf_server_address = self.dut_ip 545 else: 546 if testcase_params.get('lan_traffic_only', False): 547 iperf_server_address = wputils.get_server_address( 548 self.remote_server, self.dut_ip, '255.255.255.0') 549 else: 550 iperf_server_address = wputils.get_server_address( 551 self.remote_server, self.dut_ip, 'public') 552 iperf_args = '-i {} -t {} -J'.format( 553 IPERF_INTERVAL, testcase_params['atten_waveforms']['length']) 554 if not isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 555 iperf_args = iperf_args + ' -R' 556 iperf_future = wputils.start_iperf_client_nb( 557 self.iperf_client, iperf_server_address, iperf_args, 0, 558 testcase_params['atten_waveforms']['length'] + MED_SLEEP) 559 rssi_future = wputils.get_connected_rssi_nb( 560 self.dut, 561 int(testcase_params['atten_waveforms']['length'] / 562 testcase_params['rssi_polling_frequency']), 563 testcase_params['rssi_polling_frequency']) 564 self.run_attenuation_waveform(testcase_params) 565 client_output_path = iperf_future.result() 566 server_output_path = self.iperf_server.stop() 567 if isinstance(self.iperf_server, ipf.IPerfServerOverAdb): 568 iperf_file = server_output_path 569 else: 570 iperf_file = client_output_path 571 iperf_result = ipf.IPerfResult(iperf_file) 572 instantaneous_rates = [ 573 rate * 8 * (1.024**2) for rate in iperf_result.instantaneous_rates 574 ] 575 return { 576 'throughput': instantaneous_rates, 577 'rssi_result': rssi_future.result(), 578 'ap_settings': [ap.ap_settings for ap in self.access_points], 579 } 580 581 def run_attenuation_waveform(self, testcase_params, step_duration=1): 582 """Function that generates test params based on the test name. 583 584 Args: 585 testcase_params: dict including all test params encoded in test 586 name 587 step_duration: int representing number of seconds to dwell on each 588 atten level 589 """ 590 atten_waveforms = testcase_params['atten_waveforms'] 591 for atten_idx in range(atten_waveforms['length']): 592 start_time = time.time() 593 for network, atten_waveform in atten_waveforms.items(): 594 for idx, atten in enumerate(self.attenuators): 595 nets_on_port = [ 596 item["network"] for item in self.rf_map_by_atten[idx] 597 ] 598 if network in nets_on_port: 599 atten.set_atten(atten_waveform[atten_idx]) 600 measure_time = time.time() - start_time 601 time.sleep(step_duration - measure_time) 602 603 def compile_atten_waveforms(self, waveform_params): 604 """Function to compile all attenuation waveforms for roaming test. 605 606 Args: 607 waveform_params: list of dicts representing waveforms to generate 608 """ 609 atten_waveforms = {} 610 for network in self.main_network: 611 atten_waveforms[network] = [] 612 613 for waveform in waveform_params: 614 for network_name, network in self.main_network.items(): 615 waveform_vector = self.gen_single_atten_waveform( 616 waveform[network['roaming_label']]) 617 atten_waveforms[network_name] += waveform_vector 618 619 waveform_lengths = { 620 len(atten_waveforms[network]) 621 for network in atten_waveforms.keys() 622 } 623 if len(waveform_lengths) != 1: 624 raise ValueError( 625 'Attenuation waveform length should be equal for all networks.' 626 ) 627 else: 628 atten_waveforms['length'] = waveform_lengths.pop() 629 return atten_waveforms 630 631 def gen_single_atten_waveform(self, waveform_params): 632 """Function to generate a single attenuation waveform for roaming test. 633 634 Args: 635 waveform_params: dict representing waveform to generate 636 """ 637 waveform_vector = [] 638 for section in range(len(waveform_params['atten_levels']) - 1): 639 section_limits = waveform_params['atten_levels'][section:section + 640 2] 641 up_down = (1 - 2 * (section_limits[1] < section_limits[0])) 642 temp_section = list( 643 range(section_limits[0], section_limits[1] + up_down, 644 up_down * waveform_params['step_size'])) 645 temp_section = [ 646 val for val in temp_section 647 for _ in range(waveform_params['step_duration']) 648 ] 649 waveform_vector += temp_section 650 waveform_vector *= waveform_params['repetitions'] 651 return waveform_vector 652 653 def parse_test_params(self, testcase_params): 654 """Function that generates test params based on the test name. 655 656 Args: 657 test_name: current test name 658 Returns: 659 testcase_params: dict including all test params encoded in test 660 name 661 """ 662 if testcase_params["waveform_type"] == 'smooth': 663 testcase_params[ 664 'roaming_waveforms_params'] = self.testclass_params[ 665 'smooth_roaming_waveforms'] 666 elif testcase_params["waveform_type"] == 'failover': 667 testcase_params[ 668 'roaming_waveforms_params'] = self.testclass_params[ 669 'failover_roaming_waveforms'] 670 elif testcase_params["waveform_type"] == 'consistency': 671 testcase_params[ 672 'roaming_waveforms_params'] = self.testclass_params[ 673 'consistency_waveforms'] 674 return testcase_params 675 676 def _test_traffic_continuity(self, testcase_params): 677 """Test function for traffic continuity""" 678 # Compile test parameters from config and test name 679 testcase_params = self.parse_test_params(testcase_params) 680 testcase_params.update(self.testclass_params) 681 testcase_params['atten_waveforms'] = self.compile_atten_waveforms( 682 testcase_params['roaming_waveforms_params']) 683 # Run traffic test 684 self.setup_roaming_test(testcase_params) 685 if testcase_params['traffic_type'] == 'iperf': 686 result = self.run_iperf_test(testcase_params) 687 elif testcase_params['traffic_type'] == 'ping': 688 result = self.run_ping_test(testcase_params) 689 # Postprocess results 690 self.process_traffic_continuity_results(testcase_params, result) 691 self.pass_fail_traffic_continuity(result) 692 693 def _test_roam_consistency(self, testcase_params): 694 """Test function for roaming consistency""" 695 testcase_params = self.parse_test_params(testcase_params) 696 testcase_params.update(self.testclass_params) 697 # Run traffic test 698 secondary_attens = range( 699 self.testclass_params['consistency_waveforms']['secondary_loop'] 700 ['atten_levels'][0], self.testclass_params['consistency_waveforms'] 701 ['secondary_loop']['atten_levels'][1], 702 self.testclass_params['consistency_waveforms']['secondary_loop'] 703 ['step_size']) 704 results = collections.OrderedDict() 705 for secondary_atten in secondary_attens: 706 primary_waveform = self.gen_single_atten_waveform( 707 testcase_params['roaming_waveforms_params']['primary_sweep']) 708 secondary_waveform_params = { 709 'atten_levels': [secondary_atten, secondary_atten], 710 'step_size': 1, 711 'step_duration': len(primary_waveform), 712 'repetitions': 1 713 } 714 secondary_waveform = self.gen_single_atten_waveform( 715 secondary_waveform_params) 716 testcase_params['atten_waveforms'] = { 717 'length': len(primary_waveform) 718 } 719 for network_key, network_info in self.main_network.items(): 720 if 'primary' in network_info['roaming_label']: 721 testcase_params['atten_waveforms'][ 722 network_key] = primary_waveform 723 else: 724 testcase_params['atten_waveforms'][ 725 network_key] = secondary_waveform 726 results[secondary_atten] = [] 727 for run in range(self.testclass_params['consistency_num_runs']): 728 self.setup_roaming_test(testcase_params) 729 results[secondary_atten].append( 730 self.run_ping_test(testcase_params)) 731 # Postprocess results 732 self.process_consistency_results(testcase_params, results) 733 self.pass_fail_roaming_consistency(results) 734 735 def test_consistency_roaming_screen_on_ping(self): 736 testcase_params = { 737 "waveform_type": "consistency", 738 "screen_on": 1, 739 "traffic_type": "ping" 740 } 741 self._test_roam_consistency(testcase_params) 742 743 def test_smooth_roaming_screen_on_ping_continuity(self): 744 testcase_params = { 745 "waveform_type": "smooth", 746 "screen_on": 1, 747 "traffic_type": "ping" 748 } 749 self._test_traffic_continuity(testcase_params) 750 751 def test_smooth_roaming_screen_on_iperf_continuity(self): 752 testcase_params = { 753 "waveform_type": "smooth", 754 "screen_on": 1, 755 "traffic_type": "iperf" 756 } 757 self._test_traffic_continuity(testcase_params) 758 759 def test_failover_roaming_screen_on_ping_continuity(self): 760 testcase_params = { 761 "waveform_type": "failover", 762 "screen_on": 1, 763 "traffic_type": "ping" 764 } 765 self._test_traffic_continuity(testcase_params) 766 767 def test_failover_roaming_screen_on_iperf_continuity(self): 768 testcase_params = { 769 "waveform_type": "failover", 770 "screen_on": 1, 771 "traffic_type": "iperf" 772 } 773 self._test_traffic_continuity(testcase_params) 774 775 def test_smooth_roaming_screen_off_ping_continuity(self): 776 testcase_params = { 777 "waveform_type": "smooth", 778 "screen_on": 0, 779 "traffic_type": "ping" 780 } 781 self._test_traffic_continuity(testcase_params) 782 783 def test_smooth_roaming_screen_off_iperf_continuity(self): 784 testcase_params = { 785 "waveform_type": "smooth", 786 "screen_on": 0, 787 "traffic_type": "iperf" 788 } 789 self._test_traffic_continuity(testcase_params) 790 791 def test_failover_roaming_screen_off_ping_continuity(self): 792 testcase_params = { 793 "waveform_type": "failover", 794 "screen_on": 0, 795 "traffic_type": "ping" 796 } 797 self._test_traffic_continuity(testcase_params) 798 799 def test_failover_roaming_screen_off_iperf_continuity(self): 800 testcase_params = { 801 "waveform_type": "failover", 802 "screen_on": 0, 803 "traffic_type": "iperf" 804 } 805 self._test_traffic_continuity(testcase_params) 806