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