Omar El Ayach | b503ae8 | 2019-02-13 09:36:36 -0800 | [diff] [blame] | 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 | |
| 17 | import collections |
| 18 | import json |
| 19 | import logging |
| 20 | import math |
| 21 | import os |
| 22 | import time |
| 23 | from acts import asserts |
| 24 | from acts import base_test |
| 25 | from acts import context |
| 26 | from acts import utils |
| 27 | from acts.controllers import iperf_server as ipf |
| 28 | from acts.controllers.utils_lib import ssh |
| 29 | from acts.test_utils.wifi import wifi_performance_test_utils as wputils |
| 30 | from acts.test_utils.wifi import wifi_retail_ap as retail_ap |
| 31 | from acts.test_utils.wifi import wifi_test_utils as wutils |
| 32 | |
| 33 | SHORT_SLEEP = 1 |
| 34 | MED_SLEEP = 5 |
| 35 | TRAFFIC_GAP_THRESH = 0.5 |
| 36 | IPERF_INTERVAL = 0.25 |
| 37 | |
| 38 | |
| 39 | class 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) |
Omar El Ayach | 68395c4 | 2019-03-01 11:25:47 -0800 | [diff] [blame^] | 145 | current_context = context.get_current_context().get_full_output_path() |
| 146 | plot_file_path = os.path.join(current_context, |
Omar El Ayach | b503ae8 | 2019-02-13 09:36:36 -0800 | [diff] [blame] | 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 | |
Omar El Ayach | 68395c4 | 2019-03-01 11:25:47 -0800 | [diff] [blame^] | 158 | results_file_path = os.path.join(current_context, |
Omar El Ayach | b503ae8 | 2019-02-13 09:36:36 -0800 | [diff] [blame] | 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() |
Omar El Ayach | 68395c4 | 2019-03-01 11:25:47 -0800 | [diff] [blame^] | 184 | current_context = context.get_current_context().get_full_output_path() |
Omar El Ayach | b503ae8 | 2019-02-13 09:36:36 -0800 | [diff] [blame] | 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') |
Omar El Ayach | 68395c4 | 2019-03-01 11:25:47 -0800 | [diff] [blame^] | 203 | |
| 204 | plot_file_path = os.path.join(current_context, plot_file_name) |
Omar El Ayach | b503ae8 | 2019-02-13 09:36:36 -0800 | [diff] [blame] | 205 | figure.save_figure(plot_file_path) |
| 206 | results_dict['roam_stats'] = roam_stats |
| 207 | |
Omar El Ayach | 68395c4 | 2019-03-01 11:25:47 -0800 | [diff] [blame^] | 208 | results_file_path = os.path.join(current_context, |
Omar El Ayach | b503ae8 | 2019-02-13 09:36:36 -0800 | [diff] [blame] | 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() |