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