1#!/usr/bin/env python3 2# 3# Copyright 2020 - 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 operator 18import time 19 20from bokeh.palettes import Category10 21from bokeh.plotting import ColumnDataSource, figure, output_file, save 22from bokeh.models import Span, Label 23 24from acts import asserts 25from acts import context 26from acts import utils 27from acts.controllers.access_point import setup_ap 28from acts.controllers.ap_lib import hostapd_constants 29from acts.controllers.ap_lib import hostapd_security 30from acts_contrib.test_utils.abstract_devices import wmm_transceiver 31from acts_contrib.test_utils.fuchsia import wmm_test_cases 32from acts_contrib.test_utils.abstract_devices.wlan_device import create_wlan_device 33from acts_contrib.test_utils.abstract_devices.wlan_device_lib.AbstractDeviceWlanDeviceBaseTest import AbstractDeviceWlanDeviceBaseTest 34 35DEFAULT_N_CAPABILITIES_20_MHZ = [ 36 hostapd_constants.N_CAPABILITY_LDPC, hostapd_constants.N_CAPABILITY_SGI20, 37 hostapd_constants.N_CAPABILITY_TX_STBC, 38 hostapd_constants.N_CAPABILITY_RX_STBC1, 39 hostapd_constants.N_CAPABILITY_HT20 40] 41 42DEFAULT_AP_PARAMS = { 43 'profile_name': 'whirlwind', 44 'channel': hostapd_constants.AP_DEFAULT_CHANNEL_2G, 45 'n_capabilities': DEFAULT_N_CAPABILITIES_20_MHZ, 46 'ac_capabilities': None 47} 48 49DEFAULT_BW_PERCENTAGE = 1 50DEFAULT_STREAM_TIMEOUT = 60 51DEFAULT_STREAM_TIME = 10 52 53OPERATORS = { 54 '>': operator.gt, 55 '>=': operator.ge, 56 '<': operator.lt, 57 '<=': operator.le, 58 '==': operator.eq 59} 60 61GRAPH_COLOR_LEN = 10 62GRAPH_DEFAULT_LINE_WIDTH = 2 63GRAPH_DEFAULT_CIRCLE_SIZE = 10 64 65 66def eval_operator(operator_string, 67 actual_value, 68 expected_value, 69 max_bw, 70 rel_tolerance=0, 71 abs_tolerance=0, 72 max_bw_rel_tolerance=0): 73 """ 74 Determines if an inequality evaluates to True, given relative and absolute 75 tolerance. 76 77 Args: 78 operator_string: string, the operator to use for the comparison 79 actual_value: the value to compare to some expected value 80 expected_value: the value the actual value is compared to 81 rel_tolerance: decimal representing the percent tolerance, relative to 82 the expected value. E.g. (101 <= 100) w/ rel_tol=0.01 is True 83 abs_tolerance: the lowest actual (not percent) tolerance for error. 84 E.g. (101 == 100) w/ rel_tol=0.005 is False, but 85 (101 == 100) w/ rel_tol=0.005 and abs_tol=1 is True 86 max_bw_rel_tolerance: decimal representing the percent tolerance, 87 relative to the maximimum allowed bandwidth. 88 E.g. (101 <= max bw of 100) w/ max_bw_rel_tol=0.01 is True 89 90 91 Returns: 92 True, if inequality evaluates to True within tolerances 93 False, otherwise 94 """ 95 op = OPERATORS[operator_string] 96 if op(actual_value, expected_value): 97 return True 98 99 error = abs(actual_value - expected_value) 100 accepted_error = max(expected_value * rel_tolerance, abs_tolerance, 101 max_bw * max_bw_rel_tolerance) 102 return error <= accepted_error 103 104 105class WlanWmmTest(AbstractDeviceWlanDeviceBaseTest): 106 """Tests WMM QoS Functionality (Station only) 107 108 Testbed Requirements: 109 * One ACTS compatible wlan_device (staut) 110 * One Whirlwind Access Point 111 * For some tests, One additional ACTS compatible device (secondary_sta) 112 113 For accurate results, must be performed in an RF isolated environment. 114 """ 115 def setup_class(self): 116 super().setup_class() 117 118 try: 119 self.wmm_test_params = self.user_params['wmm_test_params'] 120 self._wmm_transceiver_configs = self.wmm_test_params[ 121 'wmm_transceivers'] 122 except KeyError: 123 raise AttributeError('Must provide at least 2 WmmTransceivers in ' 124 '"wmm_test_params" field of ACTS config.') 125 126 if len(self._wmm_transceiver_configs) < 2: 127 raise AttributeError( 128 'At least 2 WmmTransceivers must be provided.') 129 130 self.android_devices = getattr(self, 'android_devices', []) 131 self.fuchsia_devices = getattr(self, 'fuchsia_devices', []) 132 133 self.wlan_devices = [ 134 create_wlan_device(device) 135 for device in self.android_devices + self.fuchsia_devices 136 ] 137 138 # Create STAUT transceiver 139 if 'staut' not in self._wmm_transceiver_configs: 140 raise AttributeError( 141 'Must provide a WmmTransceiver labeled "staut" with a ' 142 'wlan_device.') 143 self.staut = wmm_transceiver.create( 144 self._wmm_transceiver_configs['staut'], 145 identifier='staut', 146 wlan_devices=self.wlan_devices) 147 148 # Required to for automated power cycling 149 self.dut = self.staut.wlan_device 150 151 # Create AP transceiver 152 if 'access_point' not in self._wmm_transceiver_configs: 153 raise AttributeError( 154 'Must provide a WmmTransceiver labeled "access_point" with a ' 155 'access_point.') 156 self.access_point_transceiver = wmm_transceiver.create( 157 self._wmm_transceiver_configs['access_point'], 158 identifier='access_point', 159 access_points=self.access_points) 160 161 self.wmm_transceivers = [self.staut, self.access_point_transceiver] 162 163 # Create secondary station transceiver, if present 164 if 'secondary_sta' in self._wmm_transceiver_configs: 165 self.secondary_sta = wmm_transceiver.create( 166 self._wmm_transceiver_configs['secondary_sta'], 167 identifier='secondary_sta', 168 wlan_devices=self.wlan_devices) 169 self.wmm_transceivers.append(self.secondary_sta) 170 else: 171 self.secondary_sta = None 172 173 self.wmm_transceiver_map = { 174 tc.identifier: tc 175 for tc in self.wmm_transceivers 176 } 177 178 def setup_test(self): 179 for tc in self.wmm_transceivers: 180 if tc.wlan_device: 181 tc.wlan_device.wifi_toggle_state(True) 182 tc.wlan_device.disconnect() 183 if tc.access_point: 184 tc.access_point.stop_all_aps() 185 186 def teardown_test(self): 187 for tc in self.wmm_transceivers: 188 tc.cleanup_asynchronous_streams() 189 if tc.wlan_device: 190 tc.wlan_device.disconnect() 191 tc.wlan_device.reset_wifi() 192 if tc.access_point: 193 self.download_ap_logs() 194 tc.access_point.stop_all_aps() 195 196 def teardown_class(self): 197 for tc in self.wmm_transceivers: 198 tc.destroy_resources() 199 super().teardown_class() 200 201 def on_fail(self, test_name, begin_time): 202 for wlan_device in self.wlan_devices: 203 super().on_device_fail(wlan_device.device, test_name, begin_time) 204 205 def start_ap_with_wmm_params(self, ap_parameters, wmm_parameters): 206 """Sets up WMM network on AP. 207 208 Args: 209 ap_parameters: a dictionary of kwargs to set up on ap 210 wmm_parameters: a dictionary of wmm_params to set up on ap 211 212 Returns: 213 String, subnet of the network setup (e.g. '192.168.1.0/24') 214 """ 215 # Defaults for required parameters 216 ap_parameters['force_wmm'] = True 217 if 'ssid' not in ap_parameters: 218 ap_parameters['ssid'] = utils.rand_ascii_str( 219 hostapd_constants.AP_SSID_LENGTH_2G) 220 221 if 'profile_name' not in ap_parameters: 222 ap_parameters['profile_name'] = 'whirlwind' 223 224 if 'channel' not in ap_parameters: 225 ap_parameters['channel'] = 6 226 227 if 'n_capabilities' not in ap_parameters: 228 ap_parameters['n_capabilities'] = DEFAULT_N_CAPABILITIES_20_MHZ 229 230 if 'additional_ap_parameters' in ap_parameters: 231 ap_parameters['additional_ap_parameters'].update(wmm_parameters) 232 else: 233 ap_parameters['additional_ap_parameters'] = wmm_parameters 234 235 # Optional security 236 security_config = ap_parameters.get('security_config', None) 237 if security_config: 238 ap_parameters['security'] = hostapd_security.Security( 239 **security_config) 240 ap_parameters.pop('security_config') 241 242 # Start AP with kwargs 243 self.log.info('Setting up WMM network: %s' % ap_parameters['ssid']) 244 setup_ap(self.access_point_transceiver.access_point, **ap_parameters) 245 self.log.info('Network (%s) is up.' % ap_parameters['ssid']) 246 247 # Return subnet 248 if ap_parameters['channel'] < hostapd_constants.LOWEST_5G_CHANNEL: 249 return self.access_point_transceiver.access_point._AP_2G_SUBNET_STR 250 else: 251 return self.access_point_transceiver.access_point._AP_5G_SUBNET_STR 252 253 def associate_transceiver(self, wmm_transceiver, ap_params): 254 """Associates a WmmTransceiver that has a wlan_device. 255 256 Args: 257 wmm_transceiver: transceiver to associate 258 ap_params: dict, contains ssid and password, if any, for network 259 """ 260 if not wmm_transceiver.wlan_device: 261 raise AttributeError( 262 'Cannot associate a WmmTransceiver that does not have a ' 263 'WlanDevice.') 264 ssid = ap_params['ssid'] 265 password = None 266 target_security = None 267 security = ap_params.get('security') 268 if security: 269 password = security.password 270 target_security = hostapd_constants.SECURITY_STRING_TO_DEFAULT_TARGET_SECURITY.get( 271 security.security_mode_string) 272 associated = wmm_transceiver.wlan_device.associate( 273 target_ssid=ssid, 274 target_pwd=password, 275 target_security=target_security) 276 if not associated: 277 raise ConnectionError('Failed to associate WmmTransceiver %s.' % 278 wmm_transceiver.identifier) 279 self.log.info('WmmTransceiver %s associated.' % 280 wmm_transceiver.identifier) 281 282 def validate_streams_in_phase(self, phase_id, phases, max_bw): 283 """Validates any stream in a phase that has validation criteria. 284 285 Args: 286 phase_id: identifier of the phase to check 287 phases: dictionary containing phases for retrieving stream 288 transmitters, expected bandwidths, etc. 289 max_bw: the max link bandwidth, measured in the test 290 291 Returns: 292 True, if ALL validation criteria for ALL streams in phase pass 293 False, otherwise 294 """ 295 pass_val = True 296 for stream_id, stream in phases[phase_id].items(): 297 if 'validation' in stream: 298 transmitter = stream['transmitter'] 299 uuid = stream['uuid'] 300 actual_bw = transmitter.get_results(uuid).avg_rate 301 if not actual_bw: 302 raise ConnectionError( 303 '(Phase: %s, Stream: %s) - Stream results show ' 304 'bandwidth: None' % (phase_id, stream_id)) 305 for check in stream['validation']: 306 operator_str = check['operator'] 307 rel_tolerance = check.get('rel_tolerance', 0) 308 abs_tolerance = check.get('abs_tolerance', 0) 309 max_bw_rel_tolerance = check.get('max_bw_rel_tolerance', 0) 310 expected_bw_percentage = check.get('bandwidth_percentage', 311 DEFAULT_BW_PERCENTAGE) 312 # Explicit Bandwidth Validation 313 if 'bandwidth' in check: 314 comp_bw = check['bandwidth'] 315 log_msg = ( 316 'Expected Bandwidth: %s (explicit validation ' 317 'bandwidth [%s] x expected bandwidth ' 318 'percentage [%s])' % 319 (expected_bw_percentage * comp_bw, comp_bw, 320 expected_bw_percentage)) 321 322 # Stream Comparison Validation 323 elif 'phase' in check and 'stream' in check: 324 comp_phase_id = check['phase'] 325 comp_stream_id = check['stream'] 326 comp_stream = phases[comp_phase_id][comp_stream_id] 327 comp_transmitter = comp_stream['transmitter'] 328 comp_uuid = comp_stream['uuid'] 329 comp_bw = comp_transmitter.get_results( 330 comp_uuid).avg_rate 331 log_msg = ( 332 'Expected Bandwidth: %s (bandwidth for phase: %s, ' 333 'stream: %s [%s] x expected bandwidth percentage ' 334 '[%s])' % 335 (expected_bw_percentage * comp_bw, comp_phase_id, 336 comp_stream_id, comp_bw, expected_bw_percentage)) 337 338 # Expected Bandwidth Validation 339 else: 340 if 'bandwidth' in stream: 341 comp_bw = stream['bandwidth'] 342 log_msg = ( 343 'Expected Bandwidth: %s (expected stream ' 344 'bandwidth [%s] x expected bandwidth ' 345 'percentage [%s])' % 346 (expected_bw_percentage * comp_bw, comp_bw, 347 expected_bw_percentage)) 348 else: 349 max_bw_percentage = stream.get( 350 'max_bandwidth_percentage', 351 DEFAULT_BW_PERCENTAGE) 352 comp_bw = max_bw * max_bw_percentage 353 log_msg = ( 354 'Expected Bandwidth: %s (max bandwidth [%s] x ' 355 'stream bandwidth percentage [%s] x expected ' 356 'bandwidth percentage [%s])' % 357 (expected_bw_percentage * comp_bw, max_bw, 358 max_bw_percentage, expected_bw_percentage)) 359 360 self.log.info( 361 'Validation criteria - Stream: %s, ' 362 'Actual Bandwidth: %s, Operator: %s, %s, ' 363 'Relative Tolerance: %s, Absolute Tolerance: %s, Max ' 364 'Bandwidth Relative Tolerance: %s' % 365 (stream_id, actual_bw, operator_str, log_msg, 366 rel_tolerance, abs_tolerance, max_bw_rel_tolerance)) 367 368 if eval_operator( 369 operator_str, 370 actual_bw, 371 comp_bw * expected_bw_percentage, 372 max_bw, 373 rel_tolerance=rel_tolerance, 374 abs_tolerance=abs_tolerance, 375 max_bw_rel_tolerance=max_bw_rel_tolerance): 376 self.log.info( 377 '(Phase: %s, Stream: %s) - PASSES validation check!' 378 % (phase_id, stream_id)) 379 else: 380 self.log.info( 381 '(Phase: %s, Stream: %s) - Stream FAILS validation ' 382 'check.' % (phase_id, stream_id)) 383 pass_val = False 384 if pass_val: 385 self.log.info( 386 '(Phase %s) - All streams\' validation criteria were met.' % 387 phase_id) 388 return True 389 else: 390 self.log.error( 391 '(Phase %s) - At least one stream validation criterion was not ' 392 'met.' % phase_id) 393 return False 394 395 def graph_test(self, phases, max_bw): 396 """ Outputs a bokeh html graph of the streams. Saves to ACTS log 397 directory. 398 399 Args: 400 phases: dictionary containing phases for retrieving stream 401 transmitters, expected bandwidths, etc. 402 max_bw: the max link bandwidth, measured in the test 403 404 """ 405 406 output_path = context.get_current_context().get_base_output_path() 407 output_file_name = '%s/WlanWmmTest/%s.html' % (output_path, 408 self.test_name) 409 output_file(output_file_name) 410 411 start_time = 0 412 graph_lines = [] 413 414 # Used for scaling 415 highest_stream_bw = 0 416 lowest_stream_bw = 100000 417 418 for phase_id, phase in phases.items(): 419 longest_stream_time = 0 420 for stream_id, stream in phase.items(): 421 transmitter = stream['transmitter'] 422 uuid = stream['uuid'] 423 424 if 'bandwidth' in stream: 425 stream_bw = "{:.3f}".format(stream['bandwidth']) 426 stream_bw_formula_str = '%sMb/s' % stream_bw 427 elif 'max_bandwidth_percentage' in stream: 428 max_bw_percentage = stream['max_bandwidth_percentage'] 429 stream_bw = "{:.3f}".format(max_bw * max_bw_percentage) 430 stream_bw_formula_str = '%sMb/s (%s%% of max bandwidth)' % ( 431 stream_bw, str(max_bw_percentage * 100)) 432 else: 433 raise AttributeError( 434 'Stream %s must have either a bandwidth or ' 435 'max_bandwidth_percentage parameter.' % stream_id) 436 437 stream_time = stream.get('time', DEFAULT_STREAM_TIME) 438 longest_stream_time = max(longest_stream_time, stream_time) 439 440 avg_rate = transmitter.get_results(uuid).avg_rate 441 442 instantaneous_rates = transmitter.get_results( 443 uuid).instantaneous_rates 444 highest_stream_bw = max(highest_stream_bw, 445 max(instantaneous_rates)) 446 lowest_stream_bw = min(lowest_stream_bw, 447 min(instantaneous_rates)) 448 449 stream_data = ColumnDataSource( 450 dict(time=[ 451 x for x in range(start_time, start_time + stream_time) 452 ], 453 instantaneous_bws=instantaneous_rates, 454 avg_bw=[avg_rate for _ in range(stream_time)], 455 stream_id=[stream_id for _ in range(stream_time)], 456 attempted_bw=[ 457 stream_bw_formula_str for _ in range(stream_time) 458 ])) 459 line = { 460 'x_axis': 'time', 461 'y_axis': 'instantaneous_bws', 462 'source': stream_data, 463 'line_width': GRAPH_DEFAULT_LINE_WIDTH, 464 'legend_label': '%s:%s' % (phase_id, stream_id) 465 } 466 graph_lines.append(line) 467 468 start_time = start_time + longest_stream_time 469 TOOLTIPS = [('Time', '@time'), 470 ('Attempted Bandwidth', '@attempted_bw'), 471 ('Instantaneous Bandwidth', '@instantaneous_bws'), 472 ('Stream Average Bandwidth', '@avg_bw'), 473 ('Stream', '@stream_id')] 474 475 # Create and scale graph appropriately 476 time_vs_bandwidth_graph = figure( 477 title=('Bandwidth for %s' % self.test_name), 478 x_axis_label='Time', 479 y_axis_label='Bandwidth', 480 tooltips=TOOLTIPS, 481 y_range=(lowest_stream_bw - 482 (0.5 * (highest_stream_bw - lowest_stream_bw)), 483 1.05 * max_bw)) 484 time_vs_bandwidth_graph.sizing_mode = 'stretch_both' 485 time_vs_bandwidth_graph.title.align = 'center' 486 colors = Category10[GRAPH_COLOR_LEN] 487 color_ind = 0 488 489 # Draw max bandwidth line 490 max_bw_span = Span(location=max_bw, 491 dimension='width', 492 line_color='black', 493 line_dash='dashed', 494 line_width=GRAPH_DEFAULT_LINE_WIDTH) 495 max_bw_label = Label(x=(0.5 * start_time), 496 y=max_bw, 497 text=('Max Bandwidth: %sMb/s' % max_bw), 498 text_align='center') 499 time_vs_bandwidth_graph.add_layout(max_bw_span) 500 time_vs_bandwidth_graph.add_layout(max_bw_label) 501 502 # Draw stream lines 503 for line in graph_lines: 504 time_vs_bandwidth_graph.line(line['x_axis'], 505 line['y_axis'], 506 source=line['source'], 507 line_width=line['line_width'], 508 legend_label=line['legend_label'], 509 color=colors[color_ind]) 510 time_vs_bandwidth_graph.circle(line['x_axis'], 511 line['y_axis'], 512 source=line['source'], 513 size=GRAPH_DEFAULT_CIRCLE_SIZE, 514 legend_label=line['legend_label'], 515 color=colors[color_ind]) 516 color_ind = (color_ind + 1) % GRAPH_COLOR_LEN 517 time_vs_bandwidth_graph.legend.location = "top_left" 518 time_vs_bandwidth_graph.legend.click_policy = "hide" 519 graph_file = save([time_vs_bandwidth_graph]) 520 self.log.info('Saved graph to %s' % graph_file) 521 522 def run_wmm_test(self, 523 phases, 524 ap_parameters=DEFAULT_AP_PARAMS, 525 wmm_parameters=hostapd_constants. 526 WMM_PHYS_11A_11G_11N_11AC_DEFAULT_PARAMS, 527 stream_timeout=DEFAULT_STREAM_TIMEOUT): 528 """Runs a WMM test case. 529 530 Args: 531 phases: dictionary of phases of streams to run in parallel, 532 including any validation critera (see example below). 533 ap_parameters: dictionary of custom kwargs to setup on AP (see 534 start_ap_with_wmm_parameters) 535 wmm_parameters: dictionary of WMM AC parameters 536 stream_timeout: int, time in seconds to wait before force joining 537 parallel streams 538 539 Asserts: 540 PASS, if all validation criteria for all phases are met 541 FAIL, otherwise 542 """ 543 # Setup AP 544 subnet_str = self.start_ap_with_wmm_params(ap_parameters, 545 wmm_parameters) 546 # Determine transmitters and receivers used in test case 547 transmitters = set() 548 receivers = set() 549 for phase in phases.values(): 550 for stream in phase.values(): 551 transmitter = self.wmm_transceiver_map[ 552 stream['transmitter_str']] 553 transmitters.add(transmitter) 554 stream['transmitter'] = transmitter 555 receiver = self.wmm_transceiver_map[stream['receiver_str']] 556 receivers.add(receiver) 557 stream['receiver'] = receiver 558 transceivers = transmitters.union(receivers) 559 560 # Associate all transceivers with wlan_devices 561 for tc in transceivers: 562 if tc.wlan_device: 563 self.associate_transceiver(tc, ap_parameters) 564 565 # Determine link max bandwidth 566 self.log.info('Determining link maximum bandwidth.') 567 uuid = self.staut.run_synchronous_traffic_stream( 568 {'receiver': self.access_point_transceiver}, subnet_str) 569 max_bw = self.staut.get_results(uuid).avg_send_rate 570 self.log.info('Link maximum bandwidth: %s Mb/s' % max_bw) 571 572 # Run parallel phases 573 pass_test = True 574 for phase_id, phase in phases.items(): 575 self.log.info('Setting up phase: %s' % phase_id) 576 577 for stream_id, stream in phase.items(): 578 579 transmitter = stream['transmitter'] 580 receiver = stream['receiver'] 581 access_category = stream.get('access_category', None) 582 stream_time = stream.get('time', DEFAULT_STREAM_TIME) 583 584 # Determine stream type 585 if 'bandwidth' in stream: 586 bw = stream['bandwidth'] 587 elif 'max_bandwidth_percentage' in stream: 588 max_bw_percentage = stream['max_bandwidth_percentage'] 589 bw = max_bw * max_bw_percentage 590 else: 591 raise AttributeError( 592 'Stream %s must have either a bandwidth or ' 593 'max_bandwidth_percentage parameter.' % stream_id) 594 595 stream_params = { 596 'receiver': receiver, 597 'access_category': access_category, 598 'bandwidth': bw, 599 'time': stream_time 600 } 601 602 uuid = transmitter.prepare_asynchronous_stream( 603 stream_params, subnet_str) 604 stream['uuid'] = uuid 605 606 # Start all streams in phase 607 start_time = time.time() + 5 608 for transmitter in transmitters: 609 transmitter.start_asynchronous_streams(start_time=start_time) 610 611 # Wait for streams to join 612 for transmitter in transmitters: 613 end_time = time.time() + stream_timeout 614 while transmitter.has_active_streams: 615 if time.time() > end_time: 616 raise ConnectionError( 617 'Transmitter\'s (%s) active streams are not finishing.' 618 % transmitter.identifier) 619 time.sleep(1) 620 621 # Cleanup all streams 622 for transmitter in transmitters: 623 transmitter.cleanup_asynchronous_streams() 624 625 # Validate streams 626 pass_test = pass_test and self.validate_streams_in_phase( 627 phase_id, phases, max_bw) 628 629 self.graph_test(phases, max_bw) 630 if pass_test: 631 asserts.explicit_pass( 632 'Validation criteria met for all streams in all phases.') 633 else: 634 asserts.fail( 635 'At least one stream failed to meet validation criteria.') 636 637# Test Cases 638 639# Internal Traffic Differentiation 640 641 def test_internal_traffic_diff_VO_VI(self): 642 self.run_wmm_test(wmm_test_cases.test_internal_traffic_diff_VO_VI) 643 644 def test_internal_traffic_diff_VO_BE(self): 645 self.run_wmm_test(wmm_test_cases.test_internal_traffic_diff_VO_BE) 646 647 def test_internal_traffic_diff_VO_BK(self): 648 self.run_wmm_test(wmm_test_cases.test_internal_traffic_diff_VO_BK) 649 650 def test_internal_traffic_diff_VI_BE(self): 651 self.run_wmm_test(wmm_test_cases.test_internal_traffic_diff_VI_BE) 652 653 def test_internal_traffic_diff_VI_BK(self): 654 self.run_wmm_test(wmm_test_cases.test_internal_traffic_diff_VI_BK) 655 656 def test_internal_traffic_diff_BE_BK(self): 657 self.run_wmm_test(wmm_test_cases.test_internal_traffic_diff_BE_BK) 658 659# External Traffic Differentiation 660 661 """Single station, STAUT transmits high priority""" 662 def test_external_traffic_diff_staut_VO_ap_VI(self): 663 self.run_wmm_test( 664 wmm_test_cases.test_external_traffic_diff_staut_VO_ap_VI) 665 666 def test_external_traffic_diff_staut_VO_ap_BE(self): 667 self.run_wmm_test( 668 wmm_test_cases.test_external_traffic_diff_staut_VO_ap_BE) 669 670 def test_external_traffic_diff_staut_VO_ap_BK(self): 671 self.run_wmm_test( 672 wmm_test_cases.test_external_traffic_diff_staut_VO_ap_BK) 673 674 def test_external_traffic_diff_staut_VI_ap_BE(self): 675 self.run_wmm_test( 676 wmm_test_cases.test_external_traffic_diff_staut_VI_ap_BE) 677 678 def test_external_traffic_diff_staut_VI_ap_BK(self): 679 self.run_wmm_test( 680 wmm_test_cases.test_external_traffic_diff_staut_VI_ap_BK) 681 682 def test_external_traffic_diff_staut_BE_ap_BK(self): 683 self.run_wmm_test( 684 wmm_test_cases.test_external_traffic_diff_staut_BE_ap_BK) 685 686 """Single station, STAUT transmits low priority""" 687 688 def test_external_traffic_diff_staut_VI_ap_VO(self): 689 self.run_wmm_test( 690 wmm_test_cases.test_external_traffic_diff_staut_VI_ap_VO) 691 692 def test_external_traffic_diff_staut_BE_ap_VO(self): 693 self.run_wmm_test( 694 wmm_test_cases.test_external_traffic_diff_staut_BE_ap_VO) 695 696 def test_external_traffic_diff_staut_BK_ap_VO(self): 697 self.run_wmm_test( 698 wmm_test_cases.test_external_traffic_diff_staut_BK_ap_VO) 699 700 def test_external_traffic_diff_staut_BE_ap_VI(self): 701 self.run_wmm_test( 702 wmm_test_cases.test_external_traffic_diff_staut_BE_ap_VI) 703 704 def test_external_traffic_diff_staut_BK_ap_VI(self): 705 self.run_wmm_test( 706 wmm_test_cases.test_external_traffic_diff_staut_BK_ap_VI) 707 708 def test_external_traffic_diff_staut_BK_ap_BE(self): 709 self.run_wmm_test( 710 wmm_test_cases.test_external_traffic_diff_staut_BK_ap_BE) 711 712# # Dual Internal/External Traffic Differentiation (Single station) 713 714 def test_dual_traffic_diff_staut_VO_VI_ap_VI(self): 715 self.run_wmm_test( 716 wmm_test_cases.test_dual_traffic_diff_staut_VO_VI_ap_VI) 717 718 def test_dual_traffic_diff_staut_VO_BE_ap_BE(self): 719 self.run_wmm_test( 720 wmm_test_cases.test_dual_traffic_diff_staut_VO_BE_ap_BE) 721 722 def test_dual_traffic_diff_staut_VO_BK_ap_BK(self): 723 self.run_wmm_test( 724 wmm_test_cases.test_dual_traffic_diff_staut_VO_BK_ap_BK) 725 726 def test_dual_traffic_diff_staut_VI_BE_ap_BE(self): 727 self.run_wmm_test( 728 wmm_test_cases.test_dual_traffic_diff_staut_VI_BE_ap_BE) 729 730 def test_dual_traffic_diff_staut_VI_BK_ap_BK(self): 731 self.run_wmm_test( 732 wmm_test_cases.test_dual_traffic_diff_staut_VI_BK_ap_BK) 733 734 def test_dual_traffic_diff_staut_BE_BK_ap_BK(self): 735 self.run_wmm_test( 736 wmm_test_cases.test_dual_traffic_diff_staut_BE_BK_ap_BK) 737 738# ACM Bit Conformance Tests (Single station, as WFA test below uses two) 739 740 def test_acm_bit_on_VI(self): 741 wmm_params_VI_ACM = utils.merge_dicts( 742 hostapd_constants.WMM_PHYS_11A_11G_11N_11AC_DEFAULT_PARAMS, 743 hostapd_constants.WMM_ACM_VI) 744 self.run_wmm_test(wmm_test_cases.test_acm_bit_on_VI, 745 wmm_parameters=wmm_params_VI_ACM) 746 747# AC Parameter Modificiation Tests (Single station, as WFA test below uses two) 748 749 def test_ac_param_degrade_VO(self): 750 self.run_wmm_test( 751 wmm_test_cases.test_ac_param_degrade_VO, 752 wmm_parameters=hostapd_constants.WMM_DEGRADED_VO_PARAMS) 753 754 def test_ac_param_degrade_VI(self): 755 self.run_wmm_test( 756 wmm_test_cases.test_ac_param_degrade_VI, 757 wmm_parameters=hostapd_constants.WMM_DEGRADED_VI_PARAMS) 758 759 def test_ac_param_improve_BE(self): 760 self.run_wmm_test( 761 wmm_test_cases.test_ac_param_improve_BE, 762 wmm_parameters=hostapd_constants.WMM_IMPROVE_BE_PARAMS) 763 764 def test_ac_param_improve_BK(self): 765 self.run_wmm_test( 766 wmm_test_cases.test_ac_param_improve_BK, 767 wmm_parameters=hostapd_constants.WMM_IMPROVE_BK_PARAMS) 768 769 770# WFA Test Plan Tests 771 772 """Traffic Differentiation in Single BSS (Single Station)""" 773 def test_wfa_traffic_diff_single_station_staut_BE_ap_VI_BE(self): 774 self.run_wmm_test( 775 wmm_test_cases. 776 test_wfa_traffic_diff_single_station_staut_BE_ap_VI_BE) 777 778 def test_wfa_traffic_diff_single_station_staut_VI_BE(self): 779 self.run_wmm_test( 780 wmm_test_cases.test_wfa_traffic_diff_single_station_staut_VI_BE) 781 782 def test_wfa_traffic_diff_single_station_staut_VI_BE_ap_BE(self): 783 self.run_wmm_test( 784 wmm_test_cases. 785 test_wfa_traffic_diff_single_station_staut_VI_BE_ap_BE) 786 787 def test_wfa_traffic_diff_single_station_staut_BE_BK_ap_BK(self): 788 self.run_wmm_test( 789 wmm_test_cases. 790 test_wfa_traffic_diff_single_station_staut_BE_BK_ap_BK) 791 792 def test_wfa_traffic_diff_single_station_staut_VO_VI_ap_VI(self): 793 self.run_wmm_test( 794 wmm_test_cases. 795 test_wfa_traffic_diff_single_station_staut_VO_VI_ap_VI) 796 797 """Traffic Differentiation in Single BSS (Two Stations)""" 798 799 def test_wfa_traffic_diff_two_stations_staut_BE_secondary_VI_BE(self): 800 asserts.skip_if(not self.secondary_sta, 'No secondary station.') 801 self.run_wmm_test( 802 wmm_test_cases. 803 test_wfa_traffic_diff_two_stations_staut_BE_secondary_VI_BE) 804 805 def test_wfa_traffic_diff_two_stations_staut_VI_secondary_BE(self): 806 asserts.skip_if(not self.secondary_sta, 'No secondary station.') 807 self.run_wmm_test( 808 wmm_test_cases. 809 test_wfa_traffic_diff_two_stations_staut_VI_secondary_BE) 810 811 def test_wfa_traffic_diff_two_stations_staut_BK_secondary_BE_BK(self): 812 asserts.skip_if(not self.secondary_sta, 'No secondary station.') 813 self.run_wmm_test( 814 wmm_test_cases. 815 test_wfa_traffic_diff_two_stations_staut_BK_secondary_BE_BK) 816 817 def test_wfa_traffic_diff_two_stations_staut_VI_secondary_VO_VI(self): 818 asserts.skip_if(not self.secondary_sta, 'No secondary station.') 819 self.run_wmm_test( 820 wmm_test_cases. 821 test_wfa_traffic_diff_two_stations_staut_VI_secondary_VO_VI) 822 823 """Test ACM Bit Conformance (Two Stations)""" 824 825 def test_wfa_acm_bit_on_VI(self): 826 asserts.skip_if(not self.secondary_sta, 'No secondary station.') 827 wmm_params_VI_ACM = utils.merge_dicts( 828 hostapd_constants.WMM_PHYS_11A_11G_11N_11AC_DEFAULT_PARAMS, 829 hostapd_constants.WMM_ACM_VI) 830 self.run_wmm_test(wmm_test_cases.test_wfa_acm_bit_on_VI, 831 wmm_parameters=wmm_params_VI_ACM) 832 833 """Test the AC Parameter Modification""" 834 835 def test_wfa_ac_param_degrade_VI(self): 836 asserts.skip_if(not self.secondary_sta, 'No secondary station.') 837 self.run_wmm_test( 838 wmm_test_cases.test_wfa_ac_param_degrade_VI, 839 wmm_parameters=hostapd_constants.WMM_DEGRADED_VI_PARAMS) 840