Add typing for WlanRvrTest
Fix existing mypy errors and enable this file for future mypy checks.
A future change will refactor the WlanRvrTest further to use Mobly's
generate_tests(), reducing redundancy in test declaration.
Bug: b/286584981
Change-Id: I83bffbc28e46aeeaa088c481aff0930f52f7dc1d
Reviewed-on: https://fuchsia-review.googlesource.com/c/antlion/+/925418
Reviewed-by: Hayden Nix <haydennix@google.com>
Commit-Queue: Auto-Submit <auto-submit@fuchsia-infra.iam.gserviceaccount.com>
Fuchsia-Auto-Submit: Sam Balana <sbalana@google.com>
diff --git a/packages/antlion/controllers/access_point.py b/packages/antlion/controllers/access_point.py
index f18d7bb..9d9d570 100755
--- a/packages/antlion/controllers/access_point.py
+++ b/packages/antlion/controllers/access_point.py
@@ -174,7 +174,7 @@
setup_bridge: bool = False,
is_nat_enabled: bool = True,
additional_parameters: dict[str, Any] | None = None,
- ) -> list[Any]:
+ ) -> list[str]:
"""Starts as an ap using a set of configurations.
This will start an ap on this host. To start an ap the controller
@@ -782,7 +782,7 @@
setup_bridge: bool = False,
is_ipv6_enabled: bool = False,
is_nat_enabled: bool = True,
-):
+) -> list[str]:
"""Creates a hostapd profile and runs it on an ap. This is a convenience
function that allows us to start an ap with a single function, without first
creating a hostapd config.
diff --git a/packages/antlion/controllers/ap_lib/hostapd_constants.py b/packages/antlion/controllers/ap_lib/hostapd_constants.py
index 59f1dc4..695e51a 100755
--- a/packages/antlion/controllers/ap_lib/hostapd_constants.py
+++ b/packages/antlion/controllers/ap_lib/hostapd_constants.py
@@ -15,11 +15,20 @@
# limitations under the License.
import itertools
-from enum import Enum, auto, unique
+from enum import Enum, StrEnum, auto, unique
from typing import TypedDict
+# TODO(http://b/286584981): Replace with BandType
BAND_2G = "2g"
BAND_5G = "5g"
+
+
+@unique
+class BandType(StrEnum):
+ BAND_2G = "2g"
+ BAND_5G = "5g"
+
+
CHANNEL_BANDWIDTH_20MHZ = 20
CHANNEL_BANDWIDTH_40MHZ = 40
CHANNEL_BANDWIDTH_80MHZ = 80
diff --git a/packages/antlion/controllers/iperf_client.py b/packages/antlion/controllers/iperf_client.py
index 6ca239e..517afab 100644
--- a/packages/antlion/controllers/iperf_client.py
+++ b/packages/antlion/controllers/iperf_client.py
@@ -19,6 +19,7 @@
import socket
import subprocess
import threading
+from abc import ABC, abstractmethod
from antlion import context
from antlion.capabilities.ssh import SSHConfig
@@ -81,7 +82,11 @@
pass
-class IPerfClientBase(object):
+class RouteNotFound(ConnectionError):
+ """Failed to find a route to the iperf server."""
+
+
+class IPerfClientBase(ABC):
"""The Base class for all IPerfClients.
This base class is responsible for synchronizing the logging to prevent
@@ -95,8 +100,19 @@
__log_file_lock = threading.Lock()
+ @property
+ @abstractmethod
+ def test_interface(self) -> str | None:
+ """Find the test interface.
+
+ Returns:
+ Name of the interface used to communicate with server_ap, or None if
+ not set.
+ """
+ ...
+
@staticmethod
- def _get_full_file_path(tag=""):
+ def _get_full_file_path(tag: str = ""):
"""Returns the full file path for the IPerfClient log file.
Note: If the directory for the file path does not exist, it will be
@@ -141,6 +157,10 @@
class IPerfClient(IPerfClientBase):
"""Class that handles iperf3 client operations."""
+ @property
+ def test_interface(self) -> str | None:
+ return None
+
def start(self, ip, iperf_args, tag, timeout=3600, iperf_binary=None):
"""Starts iperf client, and waits for completion.
@@ -181,9 +201,20 @@
test_interface: str | None = None,
):
self._ssh_provider = ssh_provider
- self.test_interface = test_interface
+ self._test_interface = test_interface
- def start(self, ip, iperf_args, tag, timeout=3600, iperf_binary=None):
+ @property
+ def test_interface(self) -> str | None:
+ return self._test_interface
+
+ def start(
+ self,
+ ip: str,
+ iperf_args: str,
+ tag: str,
+ timeout: int = 3600,
+ iperf_binary: str | None = None,
+ ):
"""Starts iperf client, and waits for completion.
Args:
@@ -226,7 +257,9 @@
class IPerfClientOverAdb(IPerfClientBase):
"""Class that handles iperf3 operations over ADB devices."""
- def __init__(self, android_device_or_serial, test_interface=None):
+ def __init__(
+ self, android_device_or_serial: object, test_interface: str | None = None
+ ):
"""Creates a new IPerfClientOverAdb object.
Args:
@@ -238,7 +271,11 @@
traffic to the iperf server.
"""
self._android_device_or_serial = android_device_or_serial
- self.test_interface = test_interface
+ self._test_interface = test_interface
+
+ @property
+ def test_interface(self) -> str | None:
+ return self._test_interface
@property
def _android_device(self):
diff --git a/pyproject.toml b/pyproject.toml
index 21461b1..ef0ec8c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -76,7 +76,6 @@
"tests/netstack/NetstackIfaceTest.py",
"tests/wlan/functional/DownloadStressTest.py",
"tests/wlan/functional/WlanWirelessNetworkManagementTest.py",
- "tests/wlan/performance/WlanRvrTest.py",
"tests/wlan/performance/WlanWmmTest.py",
# TODO(http://b/274619290): Remove the following files when the migration from ACTS
diff --git a/tests/wlan/functional/SoftApTest.py b/tests/wlan/functional/SoftApTest.py
index a5f331a..ea5757b 100644
--- a/tests/wlan/functional/SoftApTest.py
+++ b/tests/wlan/functional/SoftApTest.py
@@ -695,6 +695,8 @@
ip_client: iperf client to grab identifier from
"""
if type(ip_client) == iperf_client.IPerfClientOverAdb:
+ assert hasattr(ip_client._android_device_or_serial, "serial")
+ assert isinstance(ip_client._android_device_or_serial.serial, str)
return ip_client._android_device_or_serial.serial
if type(ip_client) == iperf_client.IPerfClientOverSsh:
return ip_client._ssh_provider.config.host_name
diff --git a/tests/wlan/functional/WlanRebootTest.py b/tests/wlan/functional/WlanRebootTest.py
index 09527ed..908d953 100644
--- a/tests/wlan/functional/WlanRebootTest.py
+++ b/tests/wlan/functional/WlanRebootTest.py
@@ -27,7 +27,7 @@
from antlion import utils
from antlion.controllers import iperf_client, iperf_server
from antlion.controllers.access_point import AccessPoint, setup_ap
-from antlion.controllers.ap_lib import hostapd_constants
+from antlion.controllers.ap_lib.hostapd_constants import AP_SSID_LENGTH_2G, BandType
from antlion.controllers.ap_lib.hostapd_security import Security, SecurityMode
from antlion.controllers.ap_lib.hostapd_utils import generate_random_password
from antlion.controllers.fuchsia_device import FuchsiaDevice
@@ -52,34 +52,28 @@
@unique
-class BandType(StrEnum):
- BAND_2G = "2g"
- BAND_5G = "5g"
-
-
-@unique
class IpVersionType(Enum):
IPV4 = auto()
IPV6 = auto()
DUAL_IPV4_IPV6 = auto()
def ipv4(self) -> bool:
- if self == IpVersionType.IPV4:
- return True
- if self == IpVersionType.IPV6:
- return False
- if self == IpVersionType.DUAL_IPV4_IPV6:
- return True
- return False
+ match self:
+ case IpVersionType.IPV4:
+ return True
+ case IpVersionType.IPV6:
+ return False
+ case IpVersionType.DUAL_IPV4_IPV6:
+ return True
def ipv6(self) -> bool:
- if self == IpVersionType.IPV4:
- return False
- if self == IpVersionType.IPV6:
- return True
- if self == IpVersionType.DUAL_IPV4_IPV6:
- return True
- return False
+ match self:
+ case IpVersionType.IPV4:
+ return False
+ case IpVersionType.IPV6:
+ return True
+ case IpVersionType.DUAL_IPV4_IPV6:
+ return True
@staticmethod
def all() -> list["IpVersionType"]:
@@ -534,7 +528,7 @@
)
assert isinstance(self.iperf_client_on_dut.test_interface, str)
- ssid = utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G)
+ ssid = utils.rand_ascii_str(AP_SSID_LENGTH_2G)
reboot_device: DeviceType = settings.reboot_device
reboot_type: RebootType = settings.reboot_type
band: BandType = settings.band
diff --git a/tests/wlan/performance/WlanRvrTest.py b/tests/wlan/performance/WlanRvrTest.py
index 926fcd3..6df2eae 100644
--- a/tests/wlan/performance/WlanRvrTest.py
+++ b/tests/wlan/performance/WlanRvrTest.py
@@ -16,22 +16,31 @@
import logging
import os
import time
+from dataclasses import dataclass
+from enum import StrEnum, auto, unique
+from typing import Tuple
-from mobly import asserts, test_runner
+from mobly import asserts, signals, test_runner
+from mobly.config_parser import TestRunConfig
from mobly.records import TestResultRecord
from antlion import context
from antlion.controllers.access_point import setup_ap
-from antlion.controllers.ap_lib import hostapd_constants
+from antlion.controllers.ap_lib.hostapd_constants import (
+ AP_DEFAULT_CHANNEL_2G,
+ AP_DEFAULT_CHANNEL_5G,
+ BandType,
+)
from antlion.controllers.ap_lib.hostapd_security import Security, SecurityMode
from antlion.controllers.ap_lib.radvd import Radvd
from antlion.controllers.ap_lib.radvd_config import RadvdConfig
-from antlion.controllers.attenuator import get_attenuators_for_device
+from antlion.controllers.attenuator import Attenuator, get_attenuators_for_device
from antlion.controllers.fuchsia_device import FuchsiaDevice
-from antlion.controllers.iperf_server import IPerfResult
+from antlion.controllers.iperf_server import IPerfResult, IPerfServerOverSsh
from antlion.test_utils.abstract_devices.wlan_device import create_wlan_device
from antlion.test_utils.wifi import base_test
from antlion.utils import rand_ascii_str
+from antlion.validation import MapValidator
AP_11ABG_PROFILE_NAME = "whirlwind_11ag_legacy"
REPORTING_SPEED_UNITS = "Mbps"
@@ -41,7 +50,29 @@
DAD_TIMEOUT_SEC = 30
-def create_rvr_graph(test_name, graph_path, graph_data):
+@unique
+class TrafficDirection(StrEnum):
+ RX = auto()
+ TX = auto()
+
+
+@unique
+class IPVersion(StrEnum):
+ V4 = auto()
+ V6 = auto()
+
+
+@dataclass
+class GraphData:
+ relative_attn: list[str]
+ throughput: list[int]
+ x_label: str
+ y_label: str
+
+
+def create_rvr_graph(
+ test_name: str, graph_path: str, graph_data: GraphData
+) -> list[object]:
"""Creates the RvR graphs
Args:
test_name: The name of test that was run. This is the title of the
@@ -52,7 +83,12 @@
A list of bokeh graph objects.
"""
try:
- from bokeh.plotting import ColumnDataSource, figure, output_file, save
+ from bokeh.plotting import ( # type: ignore
+ ColumnDataSource,
+ figure,
+ output_file,
+ save,
+ )
except ImportError:
logging.warn(
"bokeh is not installed: skipping creation of graphs. "
@@ -63,17 +99,18 @@
output_file(f"{graph_path}rvr_throughput_vs_attn_{test_name}.html", title=test_name)
throughput_vs_attn_data = ColumnDataSource(
- data=dict(
- relative_attn=graph_data["throughput_vs_attn"]["relative_attn"],
- throughput=graph_data["throughput_vs_attn"]["throughput"],
- )
+ data={
+ "relative_attn": graph_data.relative_attn,
+ "throughput": graph_data.throughput,
+ }
)
TOOLTIPS = [("Attenuation", "@relative_attn"), ("Throughput", "@throughput")]
+
throughput_vs_attn_graph = figure(
title=f"Throughput vs Relative Attenuation (Test Case: {test_name})",
- x_axis_label=graph_data["throughput_vs_attn"]["x_label"],
- y_axis_label=graph_data["throughput_vs_attn"]["y_label"],
- x_range=graph_data["throughput_vs_attn"]["relative_attn"],
+ x_axis_label=graph_data.x_label,
+ y_axis_label=graph_data.y_label,
+ x_range=graph_data.relative_attn,
tooltips=TOOLTIPS,
)
throughput_vs_attn_graph.sizing_mode = "stretch_width"
@@ -88,7 +125,7 @@
return [throughput_vs_attn_graph]
-def write_csv_rvr_data(test_name, csv_path, csv_data):
+def write_csv_rvr_data(test_name: str, csv_path: str, graph_data: GraphData) -> None:
"""Writes the CSV data for the RvR test
Args:
test_name: The name of test that was run.
@@ -96,12 +133,10 @@
csv_data: A dictionary of the data to be put in the csv file.
"""
csv_file_name = f"{csv_path}rvr_throughput_vs_attn_{test_name}.csv"
- throughput = csv_data["throughput_vs_attn"]["throughput"]
- relative_attn = csv_data["throughput_vs_attn"]["relative_attn"]
+ throughput = graph_data.throughput
+ relative_attn = graph_data.relative_attn
with open(csv_file_name, "w+") as csv_fileId:
- csv_fileId.write(
- f"{csv_data['throughput_vs_attn']['x_label']},{csv_data['throughput_vs_attn']['y_label']}\n"
- )
+ csv_fileId.write(f"{graph_data.x_label},{graph_data.y_label}\n")
for csv_loop_counter in range(0, len(relative_attn)):
csv_fileId.write(
f"{int(relative_attn[csv_loop_counter])},{throughput[csv_loop_counter]}\n"
@@ -118,17 +153,18 @@
* One Linux iPerf Server
"""
- def __init__(self, controllers):
- super().__init__(controllers)
- self.rvr_graph_summary = []
+ def __init__(self, configs: TestRunConfig) -> None:
+ super().__init__(configs)
+ self.rvr_graph_summary: list[object] = []
- def setup_class(self):
+ def setup_class(self) -> None:
super().setup_class()
self.log = logging.getLogger()
-
self.fuchsia_device: FuchsiaDevice | None = None
- device_type = self.user_params.get("dut", "fuchsia_devices")
+ params = MapValidator(self.user_params["rvr_settings"])
+
+ device_type = params.get(str, "dut", "fuchsia_devices")
if device_type == "fuchsia_devices":
self.fuchsia_device = self.fuchsia_devices[0]
self.dut = create_wlan_device(self.fuchsia_device)
@@ -140,31 +176,19 @@
'Expected "fuchsia_devices" or "android_devices".'
)
- self.starting_attn = self.user_params["rvr_settings"].get("starting_attn", 0)
+ self.starting_attn = params.get(int, "starting_attn", 0)
+ self.ending_attn = params.get(int, "ending_attn", 95)
+ self.step_size_in_db = params.get(int, "step_size_in_db", 1)
+ self.dwell_time_in_secs = params.get(int, "dwell_time_in_secs", 10)
- self.ending_attn = self.user_params["rvr_settings"].get("ending_attn", 95)
-
- self.step_size_in_db = self.user_params["rvr_settings"].get(
- "step_size_in_db", 1
+ self.reverse_rvr_after_forward = params.get(
+ bool, "reverse_rvr_after_forward", False
)
+ self.iperf_flags = params.get(str, "iperf_flags", "-i 1")
+ self.iperf_flags += f" -t {self.dwell_time_in_secs} -J"
+ self.debug_loop_count = params.get(int, "debug_loop_count", 1)
- self.dwell_time_in_secs = self.user_params["rvr_settings"].get(
- "dwell_time_in_secs", 10
- )
-
- self.reverse_rvr_after_forward = bool(
- (self.user_params["rvr_settings"].get("reverse_rvr_after_forward", None))
- )
-
- self.iperf_flags = self.user_params["rvr_settings"].get("iperf_flags", "-i 1")
-
- self.iperf_flags = f"{self.iperf_flags} -t {self.dwell_time_in_secs} -J"
-
- self.debug_loop_count = self.user_params["rvr_settings"].get(
- "debug_loop_count", 1
- )
-
- self.router_adv_daemon = None
+ self.router_adv_daemon: Radvd | None = None
if self.ending_attn == "auto":
self.use_auto_end = True
@@ -197,7 +221,7 @@
self.access_point.stop_all_aps()
- def setup_test(self):
+ def setup_test(self) -> None:
super().setup_test()
if self.iperf_server:
self.iperf_server.start()
@@ -207,15 +231,15 @@
ad.droid.wakeUpNow()
self.dut.wifi_toggle_state(True)
- def teardown_test(self):
+ def teardown_test(self) -> None:
self.cleanup_tests()
super().teardown_test()
- def teardown_class(self):
+ def teardown_class(self) -> None:
if self.router_adv_daemon:
self.router_adv_daemon.stop()
try:
- from bokeh.plotting import output_file, save
+ from bokeh.plotting import output_file, save # type: ignore
output_path = context.get_current_context().get_base_output_path()
test_class_name = context.get_current_context().test_class_name
@@ -235,11 +259,11 @@
super().teardown_class()
- def on_fail(self, record: TestResultRecord):
+ def on_fail(self, record: TestResultRecord) -> None:
super().on_fail(record)
self.cleanup_tests()
- def cleanup_tests(self):
+ def cleanup_tests(self) -> None:
"""Cleans up all the dangling pieces of the tests, for example, the
iperf server, radvd, all the currently running APs, and the various
clients running during the tests.
@@ -263,12 +287,12 @@
self.download_ap_logs()
self.access_point.stop_all_aps()
- def _wait_for_ipv4_addrs(self):
+ def _wait_for_ipv4_addrs(self) -> str:
"""Wait for an IPv4 addresses to become available on the DUT and iperf
server.
Returns:
- A string containing the private IPv4 address of the iperf server.
+ The private IPv4 address of the iperf server.
Raises:
TestFailure: If unable to acquire a IPv4 address.
@@ -281,6 +305,7 @@
self.iperf_server.test_interface
)
assert self.fuchsia_device is not None
+ assert self.dut_iperf_client.test_interface is not None
dut_ip_addresses = self.fuchsia_device.get_interface_ip_addresses(
self.dut_iperf_client.test_interface
)
@@ -306,12 +331,14 @@
ip_address_checker_counter += 1
time.sleep(1)
- asserts.fail(
+ raise signals.TestFailure(
"IPv4 addresses are not available on both the DUT and iperf server."
)
# TODO (b/258264565): Merge with fuchsia_device wait_for_ipv6_addr.
- def _wait_for_dad(self, device, test_interface):
+ def _wait_for_dad(
+ self, device: FuchsiaDevice | IPerfServerOverSsh, test_interface: str
+ ) -> str:
"""Wait for Duplicate Address Detection to resolve so that an
private-local IPv6 address is available for test.
@@ -339,26 +366,25 @@
self.log.info(f'DAD resolved with "{addr}" after {elapsed}s')
return addr
time.sleep(1)
- else:
- asserts.fail(
- "Unable to acquire a private-local IPv6 address for testing "
- f"after {elapsed}s"
- )
+
+ raise signals.TestFailure(
+ "Unable to acquire a private-local IPv6 address for testing "
+ f"after {elapsed}s"
+ )
def run_rvr(
self,
- ssid,
- security_mode=None,
- password=None,
- band="2g",
- traffic_dir="tx",
- ip_version=4,
- ):
+ ssid: str,
+ security: Security | None,
+ band: BandType,
+ traffic_dir: TrafficDirection,
+ ip_version: IPVersion,
+ ) -> GraphData:
"""Setups and runs the RvR test
Args:
ssid: The SSID for the client to associate to.
- password: Password for the network, if necessary.
+ security: Security of the AP
band: 2g or 5g
traffic_dir: rx or tx, bi is not supported by iperf3
ip_version: 4 or 6
@@ -367,14 +393,14 @@
The bokeh graph data.
"""
throughput: list[int] = []
- relative_attn: list[int] = []
+ relative_attn: list[str] = []
if band == "2g":
rvr_attenuators = self.attenuators_2g
elif band == "5g":
rvr_attenuators = self.attenuators_5g
else:
raise ValueError(f"Invalid WLAN band specified: {band}")
- if ip_version == 6:
+ if ip_version is IPVersion.V6:
self.router_adv_daemon = Radvd(
self.access_point.ssh,
self.access_point.interfaces.get_bridge_interface()[0],
@@ -391,10 +417,10 @@
while associate_counter < associate_max_attempts:
if self.dut.associate(
ssid,
- target_pwd=password,
- target_security=hostapd_constants.SECURITY_STRING_TO_DEFAULT_TARGET_SECURITY.get(
- security_mode
- ),
+ target_pwd=security.password if security else None,
+ target_security=security.security_mode
+ if security
+ else SecurityMode.OPEN,
check_connectivity=False,
):
break
@@ -405,9 +431,9 @@
f"Unable to associate at starting attenuation: {self.starting_attn}"
)
- if ip_version == 4:
+ if ip_version is IPVersion.V4:
iperf_server_ip_address = self._wait_for_ipv4_addrs()
- elif ip_version == 6:
+ elif ip_version is IPVersion.V6:
self.iperf_server.renew_test_interface_ip_address()
self.log.info(
"Waiting for iperf server to complete Duplicate "
@@ -421,6 +447,8 @@
"Waiting for DUT to complete Duplicate Address Detection "
f'on interface "{self.dut_iperf_client.test_interface}"...'
)
+ assert self.fuchsia_device is not None
+ assert self.dut_iperf_client.test_interface is not None
_ = self._wait_for_dad(
self.fuchsia_device, self.dut_iperf_client.test_interface
)
@@ -432,6 +460,9 @@
rvr_attenuators,
iperf_server_ip_address,
ip_version,
+ ssid,
+ security=security,
+ reverse=False,
throughput=throughput,
relative_attn=relative_attn,
)
@@ -442,36 +473,32 @@
iperf_server_ip_address,
ip_version,
ssid=ssid,
- security_mode=security_mode,
- password=password,
+ security=security,
reverse=True,
throughput=throughput,
relative_attn=relative_attn,
)
self.dut.disconnect()
- throughput_vs_attn = {
- "throughput": throughput,
- "relative_attn": relative_attn,
- "x_label": "Attenuation(db)",
- "y_label": f"Throughput({REPORTING_SPEED_UNITS})",
- }
- graph_data = {"throughput_vs_attn": throughput_vs_attn}
- return graph_data
+ return GraphData(
+ relative_attn=relative_attn,
+ throughput=throughput,
+ x_label="Attenuation(db)",
+ y_label=f"Throughput({REPORTING_SPEED_UNITS})",
+ )
def rvr_loop(
self,
- traffic_dir,
- rvr_attenuators,
- iperf_server_ip_address,
- ip_version,
- ssid=None,
- security_mode=None,
- password=None,
- reverse=False,
- throughput=None,
- relative_attn=None,
- ):
+ traffic_dir: TrafficDirection,
+ rvr_attenuators: list[Attenuator],
+ iperf_server_ip_address: str,
+ ip_version: IPVersion,
+ ssid: str,
+ security: Security | None,
+ reverse: bool,
+ throughput: list[int],
+ relative_attn: list[str],
+ ) -> Tuple[list[int], list[str]]:
"""The loop that goes through each attenuation level and runs the iperf
throughput pair.
Args:
@@ -492,7 +519,7 @@
relative_attn: The list of attenuation data for the test.
"""
iperf_flags = self.iperf_flags
- if traffic_dir == "rx":
+ if traffic_dir is TrafficDirection.RX:
iperf_flags = f"{self.iperf_flags} -R"
starting_attn = self.starting_attn
ending_attn = self.ending_attn
@@ -527,10 +554,10 @@
)
if self.dut.associate(
ssid,
- target_pwd=password,
- target_security=hostapd_constants.SECURITY_STRING_TO_DEFAULT_TARGET_SECURITY.get(
- security_mode
- ),
+ target_pwd=security.password if security else None,
+ target_security=security.security_mode
+ if security
+ else SecurityMode.OPEN,
check_connectivity=False,
):
associated = True
@@ -548,13 +575,13 @@
value_to_insert = f"{value_to_insert} "
else:
relative_attn.append(value_to_insert)
- attn_value_inserted = True
assert self.fuchsia_device is not None
+ assert self.dut_iperf_client.test_interface is not None
dut_ip_addresses = self.fuchsia_device.get_interface_ip_addresses(
self.dut_iperf_client.test_interface
)
- if ip_version == 4:
+ if ip_version is IPVersion.V4:
if not dut_ip_addresses["ipv4_private"]:
self.log.info(
"DUT does not have an IPv4 address. "
@@ -566,7 +593,7 @@
self.log.info(
f'DUT has the following IPv4 address: "{ipv4_private}"'
)
- elif ip_version == 6:
+ elif ip_version is IPVersion.V6:
if not dut_ip_addresses["ipv6_private_local"]:
self.log.info(
"DUT does not have an IPv6 address. "
@@ -588,12 +615,12 @@
else:
self.log.info(f'Iperf server "{iperf_server_ip_address}" is pingable.')
if server_pingable:
- if traffic_dir == "tx":
+ if traffic_dir is TrafficDirection.TX:
self.log.info(
f"Running traffic DUT to {iperf_server_ip_address} at relative "
f"attenuation of {step}"
)
- elif traffic_dir == "rx":
+ elif traffic_dir is TrafficDirection.RX:
self.log.info(
f"Running traffic {iperf_server_ip_address} to DUT at relative "
f"attenuation of {step}"
@@ -665,113 +692,21 @@
throughput.append(0)
return throughput, relative_attn
- def test_rvr_11ac_5g_80mhz_open_tx_ipv4(self):
+ def test_rvr_11ac_5g_80mhz_open_tx_ipv4(self) -> None:
ssid = rand_ascii_str(20)
setup_ap(
access_point=self.access_point,
profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_5G,
+ channel=AP_DEFAULT_CHANNEL_5G,
ssid=ssid,
setup_bridge=True,
)
- graph_data = self.run_rvr(ssid, band="5g", traffic_dir="tx", ip_version=4)
- for rvr_graph in create_rvr_graph(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- ):
- self.rvr_graph_summary.append(rvr_graph)
- write_csv_rvr_data(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- )
-
- def test_rvr_11ac_5g_80mhz_open_rx_ipv4(self):
- ssid = rand_ascii_str(20)
- setup_ap(
- access_point=self.access_point,
- profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_5G,
- ssid=ssid,
- setup_bridge=True,
- )
- graph_data = self.run_rvr(ssid, band="5g", traffic_dir="rx", ip_version=4)
- for rvr_graph in create_rvr_graph(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- ):
- self.rvr_graph_summary.append(rvr_graph)
- write_csv_rvr_data(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- )
-
- def test_rvr_11ac_5g_80mhz_open_tx_ipv6(self):
- ssid = rand_ascii_str(20)
- setup_ap(
- access_point=self.access_point,
- profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_5G,
- ssid=ssid,
- setup_bridge=True,
- )
- graph_data = self.run_rvr(ssid, band="5g", traffic_dir="tx", ip_version=6)
- for rvr_graph in create_rvr_graph(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- ):
- self.rvr_graph_summary.append(rvr_graph)
- write_csv_rvr_data(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- )
-
- def test_rvr_11ac_5g_80mhz_open_rx_ipv6(self):
- ssid = rand_ascii_str(20)
- setup_ap(
- access_point=self.access_point,
- profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_5G,
- ssid=ssid,
- setup_bridge=True,
- )
- graph_data = self.run_rvr(ssid, band="5g", traffic_dir="rx", ip_version=6)
- for rvr_graph in create_rvr_graph(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- ):
- self.rvr_graph_summary.append(rvr_graph)
- write_csv_rvr_data(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- )
-
- def test_rvr_11ac_5g_80mhz_wpa2_tx_ipv4(self):
- ssid = rand_ascii_str(20)
- password = rand_ascii_str(20)
- security_profile = Security(security_mode=SecurityMode.WPA2, password=password)
- setup_ap(
- access_point=self.access_point,
- profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_5G,
- ssid=ssid,
- security=security_profile,
- setup_bridge=True,
- )
graph_data = self.run_rvr(
ssid,
- security_mode="wpa2",
- password=password,
- band="5g",
- traffic_dir="tx",
- ip_version=4,
+ security=None,
+ band=BandType.BAND_5G,
+ traffic_dir=TrafficDirection.TX,
+ ip_version=IPVersion.V4,
)
for rvr_graph in create_rvr_graph(
self.current_test_info.name,
@@ -785,25 +720,21 @@
graph_data,
)
- def test_rvr_11ac_5g_80mhz_wpa2_rx_ipv4(self):
+ def test_rvr_11ac_5g_80mhz_open_rx_ipv4(self) -> None:
ssid = rand_ascii_str(20)
- password = rand_ascii_str(20)
- security_profile = Security(security_mode=SecurityMode.WPA2, password=password)
setup_ap(
access_point=self.access_point,
profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_5G,
+ channel=AP_DEFAULT_CHANNEL_5G,
ssid=ssid,
- security=security_profile,
setup_bridge=True,
)
graph_data = self.run_rvr(
ssid,
- security_mode="wpa2",
- password=password,
- band="5g",
- traffic_dir="rx",
- ip_version=4,
+ security=None,
+ band=BandType.BAND_5G,
+ traffic_dir=TrafficDirection.RX,
+ ip_version=IPVersion.V4,
)
for rvr_graph in create_rvr_graph(
self.current_test_info.name,
@@ -817,25 +748,21 @@
graph_data,
)
- def test_rvr_11ac_5g_80mhz_wpa2_tx_ipv6(self):
+ def test_rvr_11ac_5g_80mhz_open_tx_ipv6(self) -> None:
ssid = rand_ascii_str(20)
- password = rand_ascii_str(20)
- security_profile = Security(security_mode=SecurityMode.WPA2, password=password)
setup_ap(
access_point=self.access_point,
profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_5G,
+ channel=AP_DEFAULT_CHANNEL_5G,
ssid=ssid,
- security=security_profile,
setup_bridge=True,
)
graph_data = self.run_rvr(
ssid,
- security_mode="wpa2",
- password=password,
- band="5g",
- traffic_dir="tx",
- ip_version=6,
+ security=None,
+ band=BandType.BAND_5G,
+ traffic_dir=TrafficDirection.TX,
+ ip_version=IPVersion.V6,
)
for rvr_graph in create_rvr_graph(
self.current_test_info.name,
@@ -849,25 +776,21 @@
graph_data,
)
- def test_rvr_11ac_5g_80mhz_wpa2_rx_ipv6(self):
+ def test_rvr_11ac_5g_80mhz_open_rx_ipv6(self) -> None:
ssid = rand_ascii_str(20)
- password = rand_ascii_str(20)
- security_profile = Security(security_mode=SecurityMode.WPA2, password=password)
setup_ap(
access_point=self.access_point,
profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_5G,
+ channel=AP_DEFAULT_CHANNEL_5G,
ssid=ssid,
- security=security_profile,
setup_bridge=True,
)
graph_data = self.run_rvr(
ssid,
- security_mode="wpa2",
- password=password,
- band="5g",
- traffic_dir="rx",
- ip_version=6,
+ security=None,
+ band=BandType.BAND_5G,
+ traffic_dir=TrafficDirection.RX,
+ ip_version=IPVersion.V6,
)
for rvr_graph in create_rvr_graph(
self.current_test_info.name,
@@ -881,113 +804,25 @@
graph_data,
)
- def test_rvr_11n_2g_20mhz_open_tx_ipv4(self):
+ def test_rvr_11ac_5g_80mhz_wpa2_tx_ipv4(self) -> None:
ssid = rand_ascii_str(20)
+ security = Security(
+ security_mode=SecurityMode.WPA2, password=rand_ascii_str(20)
+ )
setup_ap(
access_point=self.access_point,
profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_2G,
+ channel=AP_DEFAULT_CHANNEL_5G,
ssid=ssid,
- setup_bridge=True,
- )
- graph_data = self.run_rvr(ssid, band="2g", traffic_dir="tx", ip_version=4)
- for rvr_graph in create_rvr_graph(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- ):
- self.rvr_graph_summary.append(rvr_graph)
- write_csv_rvr_data(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- )
-
- def test_rvr_11n_2g_20mhz_open_rx_ipv4(self):
- ssid = rand_ascii_str(20)
- setup_ap(
- access_point=self.access_point,
- profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_2G,
- ssid=ssid,
- setup_bridge=True,
- )
- graph_data = self.run_rvr(ssid, band="2g", traffic_dir="rx", ip_version=4)
- for rvr_graph in create_rvr_graph(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- ):
- self.rvr_graph_summary.append(rvr_graph)
- write_csv_rvr_data(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- )
-
- def test_rvr_11n_2g_20mhz_open_tx_ipv6(self):
- ssid = rand_ascii_str(20)
- setup_ap(
- access_point=self.access_point,
- profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_2G,
- ssid=ssid,
- setup_bridge=True,
- )
- graph_data = self.run_rvr(ssid, band="2g", traffic_dir="tx", ip_version=6)
- for rvr_graph in create_rvr_graph(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- ):
- self.rvr_graph_summary.append(rvr_graph)
- write_csv_rvr_data(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- )
-
- def test_rvr_11n_2g_20mhz_open_rx_ipv6(self):
- ssid = rand_ascii_str(20)
- setup_ap(
- access_point=self.access_point,
- profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_2G,
- ssid=ssid,
- setup_bridge=True,
- )
- graph_data = self.run_rvr(ssid, band="2g", traffic_dir="rx", ip_version=6)
- for rvr_graph in create_rvr_graph(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- ):
- self.rvr_graph_summary.append(rvr_graph)
- write_csv_rvr_data(
- self.current_test_info.name,
- context.get_current_context().get_full_output_path(),
- graph_data,
- )
-
- def test_rvr_11n_2g_20mhz_wpa2_tx_ipv4(self):
- ssid = rand_ascii_str(20)
- password = rand_ascii_str(20)
- security_profile = Security(security_mode=SecurityMode.WPA2, password=password)
- setup_ap(
- access_point=self.access_point,
- profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_2G,
- ssid=ssid,
- security=security_profile,
+ security=security,
setup_bridge=True,
)
graph_data = self.run_rvr(
ssid,
- security_mode="wpa2",
- password=password,
- band="2g",
- traffic_dir="tx",
- ip_version=4,
+ security=security,
+ band=BandType.BAND_5G,
+ traffic_dir=TrafficDirection.TX,
+ ip_version=IPVersion.V4,
)
for rvr_graph in create_rvr_graph(
self.current_test_info.name,
@@ -1001,25 +836,25 @@
graph_data,
)
- def test_rvr_11n_2g_20mhz_wpa2_rx_ipv4(self):
+ def test_rvr_11ac_5g_80mhz_wpa2_rx_ipv4(self) -> None:
ssid = rand_ascii_str(20)
- password = rand_ascii_str(20)
- security_profile = Security(security_mode=SecurityMode.WPA2, password=password)
+ security = Security(
+ security_mode=SecurityMode.WPA2, password=rand_ascii_str(20)
+ )
setup_ap(
access_point=self.access_point,
profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_2G,
+ channel=AP_DEFAULT_CHANNEL_5G,
ssid=ssid,
- security=security_profile,
+ security=security,
setup_bridge=True,
)
graph_data = self.run_rvr(
ssid,
- security_mode="wpa2",
- password=password,
- band="2g",
- traffic_dir="rx",
- ip_version=4,
+ security=security,
+ band=BandType.BAND_5G,
+ traffic_dir=TrafficDirection.RX,
+ ip_version=IPVersion.V4,
)
for rvr_graph in create_rvr_graph(
self.current_test_info.name,
@@ -1033,25 +868,25 @@
graph_data,
)
- def test_rvr_11n_2g_20mhz_wpa2_tx_ipv6(self):
+ def test_rvr_11ac_5g_80mhz_wpa2_tx_ipv6(self) -> None:
ssid = rand_ascii_str(20)
- password = rand_ascii_str(20)
- security_profile = Security(security_mode=SecurityMode.WPA2, password=password)
+ security = Security(
+ security_mode=SecurityMode.WPA2, password=rand_ascii_str(20)
+ )
setup_ap(
access_point=self.access_point,
profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_2G,
+ channel=AP_DEFAULT_CHANNEL_5G,
ssid=ssid,
- security=security_profile,
+ security=security,
setup_bridge=True,
)
graph_data = self.run_rvr(
ssid,
- security_mode="wpa2",
- password=password,
- band="2g",
- traffic_dir="tx",
- ip_version=6,
+ security=security,
+ band=BandType.BAND_5G,
+ traffic_dir=TrafficDirection.TX,
+ ip_version=IPVersion.V6,
)
for rvr_graph in create_rvr_graph(
self.current_test_info.name,
@@ -1065,25 +900,265 @@
graph_data,
)
- def test_rvr_11n_2g_20mhz_wpa2_rx_ipv6(self):
+ def test_rvr_11ac_5g_80mhz_wpa2_rx_ipv6(self) -> None:
ssid = rand_ascii_str(20)
- password = rand_ascii_str(20)
- security_profile = Security(security_mode=SecurityMode.WPA2, password=password)
+ security = Security(
+ security_mode=SecurityMode.WPA2, password=rand_ascii_str(20)
+ )
setup_ap(
access_point=self.access_point,
profile_name="whirlwind",
- channel=hostapd_constants.AP_DEFAULT_CHANNEL_2G,
+ channel=AP_DEFAULT_CHANNEL_5G,
ssid=ssid,
- security=security_profile,
+ security=security,
setup_bridge=True,
)
graph_data = self.run_rvr(
ssid,
- security_mode="wpa2",
- password=password,
- band="2g",
- traffic_dir="rx",
- ip_version=6,
+ security=security,
+ band=BandType.BAND_5G,
+ traffic_dir=TrafficDirection.RX,
+ ip_version=IPVersion.V6,
+ )
+ for rvr_graph in create_rvr_graph(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ ):
+ self.rvr_graph_summary.append(rvr_graph)
+ write_csv_rvr_data(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ )
+
+ def test_rvr_11n_2g_20mhz_open_tx_ipv4(self) -> None:
+ ssid = rand_ascii_str(20)
+ setup_ap(
+ access_point=self.access_point,
+ profile_name="whirlwind",
+ channel=AP_DEFAULT_CHANNEL_2G,
+ ssid=ssid,
+ setup_bridge=True,
+ )
+ graph_data = self.run_rvr(
+ ssid,
+ security=None,
+ band=BandType.BAND_2G,
+ traffic_dir=TrafficDirection.TX,
+ ip_version=IPVersion.V4,
+ )
+ for rvr_graph in create_rvr_graph(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ ):
+ self.rvr_graph_summary.append(rvr_graph)
+ write_csv_rvr_data(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ )
+
+ def test_rvr_11n_2g_20mhz_open_rx_ipv4(self) -> None:
+ ssid = rand_ascii_str(20)
+ setup_ap(
+ access_point=self.access_point,
+ profile_name="whirlwind",
+ channel=AP_DEFAULT_CHANNEL_2G,
+ ssid=ssid,
+ setup_bridge=True,
+ )
+ graph_data = self.run_rvr(
+ ssid,
+ security=None,
+ band=BandType.BAND_2G,
+ traffic_dir=TrafficDirection.RX,
+ ip_version=IPVersion.V4,
+ )
+ for rvr_graph in create_rvr_graph(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ ):
+ self.rvr_graph_summary.append(rvr_graph)
+ write_csv_rvr_data(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ )
+
+ def test_rvr_11n_2g_20mhz_open_tx_ipv6(self) -> None:
+ ssid = rand_ascii_str(20)
+ setup_ap(
+ access_point=self.access_point,
+ profile_name="whirlwind",
+ channel=AP_DEFAULT_CHANNEL_2G,
+ ssid=ssid,
+ setup_bridge=True,
+ )
+ graph_data = self.run_rvr(
+ ssid,
+ security=None,
+ band=BandType.BAND_2G,
+ traffic_dir=TrafficDirection.TX,
+ ip_version=IPVersion.V6,
+ )
+ for rvr_graph in create_rvr_graph(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ ):
+ self.rvr_graph_summary.append(rvr_graph)
+ write_csv_rvr_data(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ )
+
+ def test_rvr_11n_2g_20mhz_open_rx_ipv6(self) -> None:
+ ssid = rand_ascii_str(20)
+ setup_ap(
+ access_point=self.access_point,
+ profile_name="whirlwind",
+ channel=AP_DEFAULT_CHANNEL_2G,
+ ssid=ssid,
+ setup_bridge=True,
+ )
+ graph_data = self.run_rvr(
+ ssid,
+ security=None,
+ band=BandType.BAND_2G,
+ traffic_dir=TrafficDirection.RX,
+ ip_version=IPVersion.V6,
+ )
+ for rvr_graph in create_rvr_graph(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ ):
+ self.rvr_graph_summary.append(rvr_graph)
+ write_csv_rvr_data(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ )
+
+ def test_rvr_11n_2g_20mhz_wpa2_tx_ipv4(self) -> None:
+ ssid = rand_ascii_str(20)
+ security = Security(
+ security_mode=SecurityMode.WPA2, password=rand_ascii_str(20)
+ )
+ setup_ap(
+ access_point=self.access_point,
+ profile_name="whirlwind",
+ channel=AP_DEFAULT_CHANNEL_2G,
+ ssid=ssid,
+ security=security,
+ setup_bridge=True,
+ )
+ graph_data = self.run_rvr(
+ ssid,
+ security=security,
+ band=BandType.BAND_2G,
+ traffic_dir=TrafficDirection.TX,
+ ip_version=IPVersion.V4,
+ )
+ for rvr_graph in create_rvr_graph(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ ):
+ self.rvr_graph_summary.append(rvr_graph)
+ write_csv_rvr_data(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ )
+
+ def test_rvr_11n_2g_20mhz_wpa2_rx_ipv4(self) -> None:
+ ssid = rand_ascii_str(20)
+ security = Security(
+ security_mode=SecurityMode.WPA2, password=rand_ascii_str(20)
+ )
+ setup_ap(
+ access_point=self.access_point,
+ profile_name="whirlwind",
+ channel=AP_DEFAULT_CHANNEL_2G,
+ ssid=ssid,
+ security=security,
+ setup_bridge=True,
+ )
+ graph_data = self.run_rvr(
+ ssid,
+ security=security,
+ band=BandType.BAND_2G,
+ traffic_dir=TrafficDirection.RX,
+ ip_version=IPVersion.V4,
+ )
+ for rvr_graph in create_rvr_graph(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ ):
+ self.rvr_graph_summary.append(rvr_graph)
+ write_csv_rvr_data(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ )
+
+ def test_rvr_11n_2g_20mhz_wpa2_tx_ipv6(self) -> None:
+ ssid = rand_ascii_str(20)
+ security = Security(
+ security_mode=SecurityMode.WPA2, password=rand_ascii_str(20)
+ )
+ setup_ap(
+ access_point=self.access_point,
+ profile_name="whirlwind",
+ channel=AP_DEFAULT_CHANNEL_2G,
+ ssid=ssid,
+ security=security,
+ setup_bridge=True,
+ )
+ graph_data = self.run_rvr(
+ ssid,
+ security=security,
+ band=BandType.BAND_2G,
+ traffic_dir=TrafficDirection.TX,
+ ip_version=IPVersion.V6,
+ )
+ for rvr_graph in create_rvr_graph(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ ):
+ self.rvr_graph_summary.append(rvr_graph)
+ write_csv_rvr_data(
+ self.current_test_info.name,
+ context.get_current_context().get_full_output_path(),
+ graph_data,
+ )
+
+ def test_rvr_11n_2g_20mhz_wpa2_rx_ipv6(self) -> None:
+ ssid = rand_ascii_str(20)
+ security = Security(
+ security_mode=SecurityMode.WPA2, password=rand_ascii_str(20)
+ )
+ setup_ap(
+ access_point=self.access_point,
+ profile_name="whirlwind",
+ channel=AP_DEFAULT_CHANNEL_2G,
+ ssid=ssid,
+ security=security,
+ setup_bridge=True,
+ )
+ graph_data = self.run_rvr(
+ ssid,
+ security=security,
+ band=BandType.BAND_2G,
+ traffic_dir=TrafficDirection.RX,
+ ip_version=IPVersion.V6,
)
for rvr_graph in create_rvr_graph(
self.current_test_info.name,
