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