blob: 573c65c52acd8d4e4f12994aaef3b2f1de7cf94a [file] [log] [blame]
#!/usr/bin/env python3
#
# Copyright 2022 The Fuchsia Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
import operator
import time
from mobly import asserts, test_runner
from mobly.records import TestResultRecord
from antlion import context, utils
from antlion.controllers.access_point import setup_ap
from antlion.controllers.ap_lib import hostapd_constants, hostapd_security
from antlion.test_utils.abstract_devices import wmm_transceiver
from antlion.test_utils.abstract_devices.wlan_device import create_wlan_device
from antlion.test_utils.fuchsia import wmm_test_cases
from antlion.test_utils.wifi import base_test
DEFAULT_N_CAPABILITIES_20_MHZ = [
hostapd_constants.N_CAPABILITY_LDPC,
hostapd_constants.N_CAPABILITY_SGI20,
hostapd_constants.N_CAPABILITY_TX_STBC,
hostapd_constants.N_CAPABILITY_RX_STBC1,
hostapd_constants.N_CAPABILITY_HT20,
]
DEFAULT_AP_PARAMS = {
"profile_name": "whirlwind",
"channel": hostapd_constants.AP_DEFAULT_CHANNEL_2G,
"n_capabilities": DEFAULT_N_CAPABILITIES_20_MHZ,
"ac_capabilities": None,
}
DEFAULT_BW_PERCENTAGE = 1
DEFAULT_STREAM_TIMEOUT = 60
DEFAULT_STREAM_TIME = 10
OPERATORS = {
">": operator.gt,
">=": operator.ge,
"<": operator.lt,
"<=": operator.le,
"==": operator.eq,
}
GRAPH_COLOR_LEN = 10
GRAPH_DEFAULT_LINE_WIDTH = 2
GRAPH_DEFAULT_CIRCLE_SIZE = 10
def eval_operator(
operator_string,
actual_value,
expected_value,
max_bw,
rel_tolerance=0,
abs_tolerance=0,
max_bw_rel_tolerance=0,
):
"""
Determines if an inequality evaluates to True, given relative and absolute
tolerance.
Args:
operator_string: string, the operator to use for the comparison
actual_value: the value to compare to some expected value
expected_value: the value the actual value is compared to
rel_tolerance: decimal representing the percent tolerance, relative to
the expected value. E.g. (101 <= 100) w/ rel_tol=0.01 is True
abs_tolerance: the lowest actual (not percent) tolerance for error.
E.g. (101 == 100) w/ rel_tol=0.005 is False, but
(101 == 100) w/ rel_tol=0.005 and abs_tol=1 is True
max_bw_rel_tolerance: decimal representing the percent tolerance,
relative to the maximimum allowed bandwidth.
E.g. (101 <= max bw of 100) w/ max_bw_rel_tol=0.01 is True
Returns:
True, if inequality evaluates to True within tolerances
False, otherwise
"""
op = OPERATORS[operator_string]
if op(actual_value, expected_value):
return True
error = abs(actual_value - expected_value)
accepted_error = max(
expected_value * rel_tolerance, abs_tolerance, max_bw * max_bw_rel_tolerance
)
return error <= accepted_error
class WlanWmmTest(base_test.WifiBaseTest):
"""Tests WMM QoS Functionality (Station only)
Testbed Requirements:
* One ACTS compatible wlan_device (staut)
* One Whirlwind Access Point
* For some tests, One additional ACTS compatible device (secondary_sta)
For accurate results, must be performed in an RF isolated environment.
"""
def setup_class(self):
super().setup_class()
self.log = logging.getLogger()
try:
self.wmm_test_params = self.user_params["wmm_test_params"]
self._wmm_transceiver_configs = self.wmm_test_params["wmm_transceivers"]
except KeyError:
raise AttributeError(
"Must provide at least 2 WmmTransceivers in "
'"wmm_test_params" field of ACTS config.'
)
if len(self._wmm_transceiver_configs) < 2:
raise AttributeError("At least 2 WmmTransceivers must be provided.")
self.android_devices = getattr(self, "android_devices", [])
self.fuchsia_devices = getattr(self, "fuchsia_devices", [])
self.wlan_devices = [
create_wlan_device(device)
for device in self.android_devices + self.fuchsia_devices
]
# Create STAUT transceiver
if "staut" not in self._wmm_transceiver_configs:
raise AttributeError(
'Must provide a WmmTransceiver labeled "staut" with a ' "wlan_device."
)
self.staut = wmm_transceiver.create(
self._wmm_transceiver_configs["staut"],
identifier="staut",
wlan_devices=self.wlan_devices,
)
# Required to for automated power cycling
self.dut = self.staut.wlan_device
# Create AP transceiver
if "access_point" not in self._wmm_transceiver_configs:
raise AttributeError(
'Must provide a WmmTransceiver labeled "access_point" with a '
"access_point."
)
self.access_point_transceiver = wmm_transceiver.create(
self._wmm_transceiver_configs["access_point"],
identifier="access_point",
access_points=self.access_points,
)
self.wmm_transceivers = [self.staut, self.access_point_transceiver]
# Create secondary station transceiver, if present
if "secondary_sta" in self._wmm_transceiver_configs:
self.secondary_sta = wmm_transceiver.create(
self._wmm_transceiver_configs["secondary_sta"],
identifier="secondary_sta",
wlan_devices=self.wlan_devices,
)
self.wmm_transceivers.append(self.secondary_sta)
else:
self.secondary_sta = None
self.wmm_transceiver_map = {tc.identifier: tc for tc in self.wmm_transceivers}
def setup_test(self):
for tc in self.wmm_transceivers:
if tc.wlan_device:
tc.wlan_device.wifi_toggle_state(True)
tc.wlan_device.disconnect()
if tc.access_point:
tc.access_point.stop_all_aps()
def teardown_test(self):
for tc in self.wmm_transceivers:
tc.cleanup_asynchronous_streams()
if tc.wlan_device:
tc.wlan_device.disconnect()
tc.wlan_device.reset_wifi()
if tc.access_point:
self.download_ap_logs()
tc.access_point.stop_all_aps()
def teardown_class(self):
for tc in self.wmm_transceivers:
tc.destroy_resources()
super().teardown_class()
def on_fail(self, record: TestResultRecord):
for wlan_device in self.wlan_devices:
super().on_device_fail(wlan_device.device, record)
def start_ap_with_wmm_params(self, ap_parameters, wmm_parameters):
"""Sets up WMM network on AP.
Args:
ap_parameters: a dictionary of kwargs to set up on ap
wmm_parameters: a dictionary of wmm_params to set up on ap
Returns:
String, subnet of the network setup (e.g. '192.168.1.0/24')
"""
# Defaults for required parameters
ap_parameters["force_wmm"] = True
if "ssid" not in ap_parameters:
ap_parameters["ssid"] = utils.rand_ascii_str(
hostapd_constants.AP_SSID_LENGTH_2G
)
if "profile_name" not in ap_parameters:
ap_parameters["profile_name"] = "whirlwind"
if "channel" not in ap_parameters:
ap_parameters["channel"] = 6
if "n_capabilities" not in ap_parameters:
ap_parameters["n_capabilities"] = DEFAULT_N_CAPABILITIES_20_MHZ
if "additional_ap_parameters" in ap_parameters:
ap_parameters["additional_ap_parameters"].update(wmm_parameters)
else:
ap_parameters["additional_ap_parameters"] = wmm_parameters
# Optional security
security_config = ap_parameters.get("security_config", None)
if security_config:
ap_parameters["security"] = hostapd_security.Security(**security_config)
ap_parameters.pop("security_config")
# Start AP with kwargs
self.log.info("Setting up WMM network: %s" % ap_parameters["ssid"])
setup_ap(self.access_point_transceiver.access_point, **ap_parameters)
self.log.info("Network (%s) is up." % ap_parameters["ssid"])
# Return subnet
if ap_parameters["channel"] < hostapd_constants.LOWEST_5G_CHANNEL:
return self.access_point_transceiver.access_point._AP_2G_SUBNET_STR
else:
return self.access_point_transceiver.access_point._AP_5G_SUBNET_STR
def associate_transceiver(self, wmm_transceiver, ap_params):
"""Associates a WmmTransceiver that has a wlan_device.
Args:
wmm_transceiver: transceiver to associate
ap_params: dict, contains ssid and password, if any, for network
"""
if not wmm_transceiver.wlan_device:
raise AttributeError(
"Cannot associate a WmmTransceiver that does not have a " "WlanDevice."
)
ssid = ap_params["ssid"]
password = None
target_security = None
security = ap_params.get("security")
if security:
password = security.password
target_security = (
hostapd_constants.SECURITY_STRING_TO_DEFAULT_TARGET_SECURITY.get(
security.security_mode_string
)
)
associated = wmm_transceiver.wlan_device.associate(
target_ssid=ssid, target_pwd=password, target_security=target_security
)
if not associated:
raise ConnectionError(
"Failed to associate WmmTransceiver %s." % wmm_transceiver.identifier
)
self.log.info("WmmTransceiver %s associated." % wmm_transceiver.identifier)
def validate_streams_in_phase(self, phase_id, phases, max_bw):
"""Validates any stream in a phase that has validation criteria.
Args:
phase_id: identifier of the phase to check
phases: dictionary containing phases for retrieving stream
transmitters, expected bandwidths, etc.
max_bw: the max link bandwidth, measured in the test
Returns:
True, if ALL validation criteria for ALL streams in phase pass
False, otherwise
"""
pass_val = True
for stream_id, stream in phases[phase_id].items():
if "validation" in stream:
transmitter = stream["transmitter"]
uuid = stream["uuid"]
actual_bw = transmitter.get_results(uuid).avg_rate
if not actual_bw:
raise ConnectionError(
"(Phase: %s, Stream: %s) - Stream results show "
"bandwidth: None" % (phase_id, stream_id)
)
for check in stream["validation"]:
operator_str = check["operator"]
rel_tolerance = check.get("rel_tolerance", 0)
abs_tolerance = check.get("abs_tolerance", 0)
max_bw_rel_tolerance = check.get("max_bw_rel_tolerance", 0)
expected_bw_percentage = check.get(
"bandwidth_percentage", DEFAULT_BW_PERCENTAGE
)
# Explicit Bandwidth Validation
if "bandwidth" in check:
comp_bw = check["bandwidth"]
log_msg = (
"Expected Bandwidth: %s (explicit validation "
"bandwidth [%s] x expected bandwidth "
"percentage [%s])"
% (
expected_bw_percentage * comp_bw,
comp_bw,
expected_bw_percentage,
)
)
# Stream Comparison Validation
elif "phase" in check and "stream" in check:
comp_phase_id = check["phase"]
comp_stream_id = check["stream"]
comp_stream = phases[comp_phase_id][comp_stream_id]
comp_transmitter = comp_stream["transmitter"]
comp_uuid = comp_stream["uuid"]
comp_bw = comp_transmitter.get_results(comp_uuid).avg_rate
log_msg = (
"Expected Bandwidth: %s (bandwidth for phase: %s, "
"stream: %s [%s] x expected bandwidth percentage "
"[%s])"
% (
expected_bw_percentage * comp_bw,
comp_phase_id,
comp_stream_id,
comp_bw,
expected_bw_percentage,
)
)
# Expected Bandwidth Validation
else:
if "bandwidth" in stream:
comp_bw = stream["bandwidth"]
log_msg = (
"Expected Bandwidth: %s (expected stream "
"bandwidth [%s] x expected bandwidth "
"percentage [%s])"
% (
expected_bw_percentage * comp_bw,
comp_bw,
expected_bw_percentage,
)
)
else:
max_bw_percentage = stream.get(
"max_bandwidth_percentage", DEFAULT_BW_PERCENTAGE
)
comp_bw = max_bw * max_bw_percentage
log_msg = (
"Expected Bandwidth: %s (max bandwidth [%s] x "
"stream bandwidth percentage [%s] x expected "
"bandwidth percentage [%s])"
% (
expected_bw_percentage * comp_bw,
max_bw,
max_bw_percentage,
expected_bw_percentage,
)
)
self.log.info(
"Validation criteria - Stream: %s, "
"Actual Bandwidth: %s, Operator: %s, %s, "
"Relative Tolerance: %s, Absolute Tolerance: %s, Max "
"Bandwidth Relative Tolerance: %s"
% (
stream_id,
actual_bw,
operator_str,
log_msg,
rel_tolerance,
abs_tolerance,
max_bw_rel_tolerance,
)
)
if eval_operator(
operator_str,
actual_bw,
comp_bw * expected_bw_percentage,
max_bw,
rel_tolerance=rel_tolerance,
abs_tolerance=abs_tolerance,
max_bw_rel_tolerance=max_bw_rel_tolerance,
):
self.log.info(
"(Phase: %s, Stream: %s) - PASSES validation check!"
% (phase_id, stream_id)
)
else:
self.log.info(
"(Phase: %s, Stream: %s) - Stream FAILS validation "
"check." % (phase_id, stream_id)
)
pass_val = False
if pass_val:
self.log.info(
"(Phase %s) - All streams' validation criteria were met." % phase_id
)
return True
else:
self.log.error(
"(Phase %s) - At least one stream validation criterion was not "
"met." % phase_id
)
return False
def graph_test(self, phases, max_bw):
"""Outputs a bokeh html graph of the streams. Saves to ACTS log
directory.
Args:
phases: dictionary containing phases for retrieving stream
transmitters, expected bandwidths, etc.
max_bw: the max link bandwidth, measured in the test
"""
try:
from bokeh.models import Label, Span
from bokeh.palettes import Category10
from bokeh.plotting import ColumnDataSource, figure, output_file, save
except ImportError:
self.log.warn(
"bokeh is not installed: skipping creation of graphs. "
"Note CSV files are still available. If graphs are "
'desired, install antlion with the "bokeh" feature.'
)
return
output_path = context.get_current_context().get_base_output_path()
output_file_name = "%s/WlanWmmTest/%s.html" % (output_path, self.test_name)
output_file(output_file_name)
start_time = 0
graph_lines = []
# Used for scaling
highest_stream_bw = 0
lowest_stream_bw = 100000
for phase_id, phase in phases.items():
longest_stream_time = 0
for stream_id, stream in phase.items():
transmitter = stream["transmitter"]
uuid = stream["uuid"]
if "bandwidth" in stream:
stream_bw = "{:.3f}".format(stream["bandwidth"])
stream_bw_formula_str = "%sMb/s" % stream_bw
elif "max_bandwidth_percentage" in stream:
max_bw_percentage = stream["max_bandwidth_percentage"]
stream_bw = "{:.3f}".format(max_bw * max_bw_percentage)
stream_bw_formula_str = "%sMb/s (%s%% of max bandwidth)" % (
stream_bw,
str(max_bw_percentage * 100),
)
else:
raise AttributeError(
"Stream %s must have either a bandwidth or "
"max_bandwidth_percentage parameter." % stream_id
)
stream_time = stream.get("time", DEFAULT_STREAM_TIME)
longest_stream_time = max(longest_stream_time, stream_time)
avg_rate = transmitter.get_results(uuid).avg_rate
instantaneous_rates = transmitter.get_results(uuid).instantaneous_rates
highest_stream_bw = max(highest_stream_bw, max(instantaneous_rates))
lowest_stream_bw = min(lowest_stream_bw, min(instantaneous_rates))
stream_data = ColumnDataSource(
dict(
time=[x for x in range(start_time, start_time + stream_time)],
instantaneous_bws=instantaneous_rates,
avg_bw=[avg_rate for _ in range(stream_time)],
stream_id=[stream_id for _ in range(stream_time)],
attempted_bw=[
stream_bw_formula_str for _ in range(stream_time)
],
)
)
line = {
"x_axis": "time",
"y_axis": "instantaneous_bws",
"source": stream_data,
"line_width": GRAPH_DEFAULT_LINE_WIDTH,
"legend_label": "%s:%s" % (phase_id, stream_id),
}
graph_lines.append(line)
start_time = start_time + longest_stream_time
TOOLTIPS = [
("Time", "@time"),
("Attempted Bandwidth", "@attempted_bw"),
("Instantaneous Bandwidth", "@instantaneous_bws"),
("Stream Average Bandwidth", "@avg_bw"),
("Stream", "@stream_id"),
]
# Create and scale graph appropriately
time_vs_bandwidth_graph = figure(
title=("Bandwidth for %s" % self.test_name),
x_axis_label="Time",
y_axis_label="Bandwidth",
tooltips=TOOLTIPS,
y_range=(
lowest_stream_bw - (0.5 * (highest_stream_bw - lowest_stream_bw)),
1.05 * max_bw,
),
)
time_vs_bandwidth_graph.sizing_mode = "stretch_both"
time_vs_bandwidth_graph.title.align = "center"
colors = Category10[GRAPH_COLOR_LEN]
color_ind = 0
# Draw max bandwidth line
max_bw_span = Span(
location=max_bw,
dimension="width",
line_color="black",
line_dash="dashed",
line_width=GRAPH_DEFAULT_LINE_WIDTH,
)
max_bw_label = Label(
x=(0.5 * start_time),
y=max_bw,
text=("Max Bandwidth: %sMb/s" % max_bw),
text_align="center",
)
time_vs_bandwidth_graph.add_layout(max_bw_span)
time_vs_bandwidth_graph.add_layout(max_bw_label)
# Draw stream lines
for line in graph_lines:
time_vs_bandwidth_graph.line(
line["x_axis"],
line["y_axis"],
source=line["source"],
line_width=line["line_width"],
legend_label=line["legend_label"],
color=colors[color_ind],
)
time_vs_bandwidth_graph.circle(
line["x_axis"],
line["y_axis"],
source=line["source"],
size=GRAPH_DEFAULT_CIRCLE_SIZE,
legend_label=line["legend_label"],
color=colors[color_ind],
)
color_ind = (color_ind + 1) % GRAPH_COLOR_LEN
time_vs_bandwidth_graph.legend.location = "top_left"
time_vs_bandwidth_graph.legend.click_policy = "hide"
graph_file = save([time_vs_bandwidth_graph])
self.log.info("Saved graph to %s" % graph_file)
def run_wmm_test(
self,
phases,
ap_parameters=DEFAULT_AP_PARAMS,
wmm_parameters=hostapd_constants.WMM_PHYS_11A_11G_11N_11AC_DEFAULT_PARAMS,
stream_timeout=DEFAULT_STREAM_TIMEOUT,
):
"""Runs a WMM test case.
Args:
phases: dictionary of phases of streams to run in parallel,
including any validation critera (see example below).
ap_parameters: dictionary of custom kwargs to setup on AP (see
start_ap_with_wmm_parameters)
wmm_parameters: dictionary of WMM AC parameters
stream_timeout: int, time in seconds to wait before force joining
parallel streams
Asserts:
PASS, if all validation criteria for all phases are met
FAIL, otherwise
"""
# Setup AP
subnet_str = self.start_ap_with_wmm_params(ap_parameters, wmm_parameters)
# Determine transmitters and receivers used in test case
transmitters = set()
receivers = set()
for phase in phases.values():
for stream in phase.values():
transmitter = self.wmm_transceiver_map[stream["transmitter_str"]]
transmitters.add(transmitter)
stream["transmitter"] = transmitter
receiver = self.wmm_transceiver_map[stream["receiver_str"]]
receivers.add(receiver)
stream["receiver"] = receiver
transceivers = transmitters.union(receivers)
# Associate all transceivers with wlan_devices
for tc in transceivers:
if tc.wlan_device:
self.associate_transceiver(tc, ap_parameters)
# Determine link max bandwidth
self.log.info("Determining link maximum bandwidth.")
uuid = self.staut.run_synchronous_traffic_stream(
{"receiver": self.access_point_transceiver}, subnet_str
)
max_bw = self.staut.get_results(uuid).avg_send_rate
self.log.info("Link maximum bandwidth: %s Mb/s" % max_bw)
# Run parallel phases
pass_test = True
for phase_id, phase in phases.items():
self.log.info("Setting up phase: %s" % phase_id)
for stream_id, stream in phase.items():
transmitter = stream["transmitter"]
receiver = stream["receiver"]
access_category = stream.get("access_category", None)
stream_time = stream.get("time", DEFAULT_STREAM_TIME)
# Determine stream type
if "bandwidth" in stream:
bw = stream["bandwidth"]
elif "max_bandwidth_percentage" in stream:
max_bw_percentage = stream["max_bandwidth_percentage"]
bw = max_bw * max_bw_percentage
else:
raise AttributeError(
"Stream %s must have either a bandwidth or "
"max_bandwidth_percentage parameter." % stream_id
)
stream_params = {
"receiver": receiver,
"access_category": access_category,
"bandwidth": bw,
"time": stream_time,
}
uuid = transmitter.prepare_asynchronous_stream(
stream_params, subnet_str
)
stream["uuid"] = uuid
# Start all streams in phase
start_time = time.time() + 5
for transmitter in transmitters:
transmitter.start_asynchronous_streams(start_time=start_time)
# Wait for streams to join
for transmitter in transmitters:
end_time = time.time() + stream_timeout
while transmitter.has_active_streams:
if time.time() > end_time:
raise ConnectionError(
"Transmitter's (%s) active streams are not finishing."
% transmitter.identifier
)
time.sleep(1)
# Cleanup all streams
for transmitter in transmitters:
transmitter.cleanup_asynchronous_streams()
# Validate streams
pass_test = pass_test and self.validate_streams_in_phase(
phase_id, phases, max_bw
)
self.graph_test(phases, max_bw)
if pass_test:
asserts.explicit_pass(
"Validation criteria met for all streams in all phases."
)
else:
asserts.fail("At least one stream failed to meet validation criteria.")
# Test Cases
# Internal Traffic Differentiation
def test_internal_traffic_diff_VO_VI(self):
self.run_wmm_test(wmm_test_cases.test_internal_traffic_diff_VO_VI)
def test_internal_traffic_diff_VO_BE(self):
self.run_wmm_test(wmm_test_cases.test_internal_traffic_diff_VO_BE)
def test_internal_traffic_diff_VO_BK(self):
self.run_wmm_test(wmm_test_cases.test_internal_traffic_diff_VO_BK)
def test_internal_traffic_diff_VI_BE(self):
self.run_wmm_test(wmm_test_cases.test_internal_traffic_diff_VI_BE)
def test_internal_traffic_diff_VI_BK(self):
self.run_wmm_test(wmm_test_cases.test_internal_traffic_diff_VI_BK)
def test_internal_traffic_diff_BE_BK(self):
self.run_wmm_test(wmm_test_cases.test_internal_traffic_diff_BE_BK)
# External Traffic Differentiation
"""Single station, STAUT transmits high priority"""
def test_external_traffic_diff_staut_VO_ap_VI(self):
self.run_wmm_test(wmm_test_cases.test_external_traffic_diff_staut_VO_ap_VI)
def test_external_traffic_diff_staut_VO_ap_BE(self):
self.run_wmm_test(wmm_test_cases.test_external_traffic_diff_staut_VO_ap_BE)
def test_external_traffic_diff_staut_VO_ap_BK(self):
self.run_wmm_test(wmm_test_cases.test_external_traffic_diff_staut_VO_ap_BK)
def test_external_traffic_diff_staut_VI_ap_BE(self):
self.run_wmm_test(wmm_test_cases.test_external_traffic_diff_staut_VI_ap_BE)
def test_external_traffic_diff_staut_VI_ap_BK(self):
self.run_wmm_test(wmm_test_cases.test_external_traffic_diff_staut_VI_ap_BK)
def test_external_traffic_diff_staut_BE_ap_BK(self):
self.run_wmm_test(wmm_test_cases.test_external_traffic_diff_staut_BE_ap_BK)
"""Single station, STAUT transmits low priority"""
def test_external_traffic_diff_staut_VI_ap_VO(self):
self.run_wmm_test(wmm_test_cases.test_external_traffic_diff_staut_VI_ap_VO)
def test_external_traffic_diff_staut_BE_ap_VO(self):
self.run_wmm_test(wmm_test_cases.test_external_traffic_diff_staut_BE_ap_VO)
def test_external_traffic_diff_staut_BK_ap_VO(self):
self.run_wmm_test(wmm_test_cases.test_external_traffic_diff_staut_BK_ap_VO)
def test_external_traffic_diff_staut_BE_ap_VI(self):
self.run_wmm_test(wmm_test_cases.test_external_traffic_diff_staut_BE_ap_VI)
def test_external_traffic_diff_staut_BK_ap_VI(self):
self.run_wmm_test(wmm_test_cases.test_external_traffic_diff_staut_BK_ap_VI)
def test_external_traffic_diff_staut_BK_ap_BE(self):
self.run_wmm_test(wmm_test_cases.test_external_traffic_diff_staut_BK_ap_BE)
# # Dual Internal/External Traffic Differentiation (Single station)
def test_dual_traffic_diff_staut_VO_VI_ap_VI(self):
self.run_wmm_test(wmm_test_cases.test_dual_traffic_diff_staut_VO_VI_ap_VI)
def test_dual_traffic_diff_staut_VO_BE_ap_BE(self):
self.run_wmm_test(wmm_test_cases.test_dual_traffic_diff_staut_VO_BE_ap_BE)
def test_dual_traffic_diff_staut_VO_BK_ap_BK(self):
self.run_wmm_test(wmm_test_cases.test_dual_traffic_diff_staut_VO_BK_ap_BK)
def test_dual_traffic_diff_staut_VI_BE_ap_BE(self):
self.run_wmm_test(wmm_test_cases.test_dual_traffic_diff_staut_VI_BE_ap_BE)
def test_dual_traffic_diff_staut_VI_BK_ap_BK(self):
self.run_wmm_test(wmm_test_cases.test_dual_traffic_diff_staut_VI_BK_ap_BK)
def test_dual_traffic_diff_staut_BE_BK_ap_BK(self):
self.run_wmm_test(wmm_test_cases.test_dual_traffic_diff_staut_BE_BK_ap_BK)
# ACM Bit Conformance Tests (Single station, as WFA test below uses two)
def test_acm_bit_on_VI(self):
wmm_params_VI_ACM = utils.merge_dicts(
hostapd_constants.WMM_PHYS_11A_11G_11N_11AC_DEFAULT_PARAMS,
hostapd_constants.WMM_ACM_VI,
)
self.run_wmm_test(
wmm_test_cases.test_acm_bit_on_VI, wmm_parameters=wmm_params_VI_ACM
)
# AC Parameter Modificiation Tests (Single station, as WFA test below uses two)
def test_ac_param_degrade_VO(self):
self.run_wmm_test(
wmm_test_cases.test_ac_param_degrade_VO,
wmm_parameters=hostapd_constants.WMM_DEGRADED_VO_PARAMS,
)
def test_ac_param_degrade_VI(self):
self.run_wmm_test(
wmm_test_cases.test_ac_param_degrade_VI,
wmm_parameters=hostapd_constants.WMM_DEGRADED_VI_PARAMS,
)
def test_ac_param_improve_BE(self):
self.run_wmm_test(
wmm_test_cases.test_ac_param_improve_BE,
wmm_parameters=hostapd_constants.WMM_IMPROVE_BE_PARAMS,
)
def test_ac_param_improve_BK(self):
self.run_wmm_test(
wmm_test_cases.test_ac_param_improve_BK,
wmm_parameters=hostapd_constants.WMM_IMPROVE_BK_PARAMS,
)
# WFA Test Plan Tests
"""Traffic Differentiation in Single BSS (Single Station)"""
def test_wfa_traffic_diff_single_station_staut_BE_ap_VI_BE(self):
self.run_wmm_test(
wmm_test_cases.test_wfa_traffic_diff_single_station_staut_BE_ap_VI_BE
)
def test_wfa_traffic_diff_single_station_staut_VI_BE(self):
self.run_wmm_test(
wmm_test_cases.test_wfa_traffic_diff_single_station_staut_VI_BE
)
def test_wfa_traffic_diff_single_station_staut_VI_BE_ap_BE(self):
self.run_wmm_test(
wmm_test_cases.test_wfa_traffic_diff_single_station_staut_VI_BE_ap_BE
)
def test_wfa_traffic_diff_single_station_staut_BE_BK_ap_BK(self):
self.run_wmm_test(
wmm_test_cases.test_wfa_traffic_diff_single_station_staut_BE_BK_ap_BK
)
def test_wfa_traffic_diff_single_station_staut_VO_VI_ap_VI(self):
self.run_wmm_test(
wmm_test_cases.test_wfa_traffic_diff_single_station_staut_VO_VI_ap_VI
)
"""Traffic Differentiation in Single BSS (Two Stations)"""
def test_wfa_traffic_diff_two_stations_staut_BE_secondary_VI_BE(self):
asserts.skip_if(not self.secondary_sta, "No secondary station.")
self.run_wmm_test(
wmm_test_cases.test_wfa_traffic_diff_two_stations_staut_BE_secondary_VI_BE
)
def test_wfa_traffic_diff_two_stations_staut_VI_secondary_BE(self):
asserts.skip_if(not self.secondary_sta, "No secondary station.")
self.run_wmm_test(
wmm_test_cases.test_wfa_traffic_diff_two_stations_staut_VI_secondary_BE
)
def test_wfa_traffic_diff_two_stations_staut_BK_secondary_BE_BK(self):
asserts.skip_if(not self.secondary_sta, "No secondary station.")
self.run_wmm_test(
wmm_test_cases.test_wfa_traffic_diff_two_stations_staut_BK_secondary_BE_BK
)
def test_wfa_traffic_diff_two_stations_staut_VI_secondary_VO_VI(self):
asserts.skip_if(not self.secondary_sta, "No secondary station.")
self.run_wmm_test(
wmm_test_cases.test_wfa_traffic_diff_two_stations_staut_VI_secondary_VO_VI
)
"""Test ACM Bit Conformance (Two Stations)"""
def test_wfa_acm_bit_on_VI(self):
asserts.skip_if(not self.secondary_sta, "No secondary station.")
wmm_params_VI_ACM = utils.merge_dicts(
hostapd_constants.WMM_PHYS_11A_11G_11N_11AC_DEFAULT_PARAMS,
hostapd_constants.WMM_ACM_VI,
)
self.run_wmm_test(
wmm_test_cases.test_wfa_acm_bit_on_VI, wmm_parameters=wmm_params_VI_ACM
)
"""Test the AC Parameter Modification"""
def test_wfa_ac_param_degrade_VI(self):
asserts.skip_if(not self.secondary_sta, "No secondary station.")
self.run_wmm_test(
wmm_test_cases.test_wfa_ac_param_degrade_VI,
wmm_parameters=hostapd_constants.WMM_DEGRADED_VI_PARAMS,
)
if __name__ == "__main__":
test_runner.main()