Migrate SoftApTest to generate_tests
Massive overhaul of SoftApTest to use Mobly's generate_test function
instead of the deprecated ACTS run_generated_testcases function.
While here, add complete typing and ensure mypy type compliance.
Fixed: b/281582459
Bug: b/274619290
Change-Id: I3b95c8731e3293c9b96d71ccc7f3710d09ad2efd
Reviewed-on: https://fuchsia-review.googlesource.com/c/antlion/+/885956
Fuchsia-Auto-Submit: Sam Balana <sbalana@google.com>
Reviewed-by: Hayden Nix <haydennix@google.com>
Commit-Queue: Auto-Submit <auto-submit@fuchsia-infra.iam.gserviceaccount.com>
diff --git a/pyproject.toml b/pyproject.toml
index 0354c05..dbf4dd2 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -79,7 +79,6 @@
"tests/netstack/NetstackIfaceTest.py",
"tests/wlan/functional/BeaconLossTest.py",
"tests/wlan/functional/DownloadStressTest.py",
- "tests/wlan/functional/SoftApTest.py",
"tests/wlan/functional/WlanRebootTest.py",
"tests/wlan/functional/WlanWirelessNetworkManagementTest.py",
"tests/wlan/performance/WlanRvrTest.py",
diff --git a/tests/wlan/functional/SoftApTest.py b/tests/wlan/functional/SoftApTest.py
index 6d8b562..28907fc 100644
--- a/tests/wlan/functional/SoftApTest.py
+++ b/tests/wlan/functional/SoftApTest.py
@@ -18,112 +18,267 @@
import multiprocessing as mp
import random
import time
-from typing import Any, Dict, Optional
+from dataclasses import dataclass
+from enum import Enum, auto, unique
+from typing import Any, Dict, List, Mapping, Optional, Type, TypeVar, Union
from mobly import asserts, signals, test_runner
+from mobly.config_parser import TestRunConfig
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, hostapd_security
+from antlion.controllers.android_device import AndroidDevice
+from antlion.controllers.ap_lib import hostapd_constants
+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
+from antlion.controllers.fuchsia_lib.wlan_ap_policy_lib import (
+ ConnectivityMode,
+ OperatingBand,
+)
from antlion.controllers.utils_lib.ssh import settings
-from antlion.test_utils.abstract_devices.wlan_device import create_wlan_device
+from antlion.controllers.utils_lib.ssh.connection import SshConnection
+from antlion.test_utils.abstract_devices.wlan_device import (
+ AndroidWlanDevice,
+ FuchsiaWlanDevice,
+ SupportsWLAN,
+ create_wlan_device,
+)
from antlion.test_utils.wifi import base_test
+from antlion.typing import StrEnum
-CONNECTIVITY_MODE_LOCAL = "local_only"
-CONNECTIVITY_MODE_UNRESTRICTED = "unrestricted"
DEFAULT_AP_PROFILE = "whirlwind"
DEFAULT_IPERF_PORT = 5201
-DEFAULT_STRESS_TEST_ITERATIONS = 10
DEFAULT_TIMEOUT = 30
DEFAULT_IPERF_TIMEOUT = 60
DEFAULT_NO_ADDR_EXPECTED_TIMEOUT = 5
-INTERFACE_ROLE_AP = "Ap"
-INTERFACE_ROLE_CLIENT = "Client"
-OPERATING_BAND_2G = "only_2_4_ghz"
-OPERATING_BAND_5G = "only_5_ghz"
-OPERATING_BAND_ANY = "any"
-SECURITY_OPEN = "none"
-SECURITY_WEP = "wep"
-SECURITY_WPA = "wpa"
-SECURITY_WPA2 = "wpa2"
-SECURITY_WPA3 = "wpa3"
STATE_UP = True
STATE_DOWN = False
-TEST_TYPE_ASSOCIATE_ONLY = "associate_only"
-TEST_TYPE_ASSOCIATE_AND_PING = "associate_and_ping"
-TEST_TYPE_ASSOCIATE_AND_PASS_TRAFFIC = "associate_and_pass_traffic"
-TEST_TYPES = {
- TEST_TYPE_ASSOCIATE_ONLY,
- TEST_TYPE_ASSOCIATE_AND_PING,
- TEST_TYPE_ASSOCIATE_AND_PASS_TRAFFIC,
-}
+
+ConfigValue = Union[str, int, bool, List["ConfigValue"], "Config"]
+Config = Dict[str, ConfigValue]
+
+T = TypeVar("T")
-def get_test_name_from_settings(settings):
- return settings["test_name"]
+def get_typed(map: Mapping[str, Any], key: str, value_type: Type[T], default: T) -> T:
+ value = map.get(key, default)
+ if not isinstance(value, value_type):
+ raise TypeError(f'"{key}" must be a {value_type.__name__}, got {type(value)}')
+ return value
-def get_ap_params_from_config_or_default(config):
- """Retrieves AP parameters from antlion config, or returns default settings.
+@unique
+class DeviceRole(Enum):
+ AP = auto()
+ CLIENT = auto()
- Args:
- config: dict, from antlion config, that may contain custom ap parameters
- Returns:
- dict, containing all AP parameters
- """
- profile = config.get("profile", DEFAULT_AP_PROFILE)
- ssid = config.get("ssid", utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G))
- channel = config.get("channel", hostapd_constants.AP_DEFAULT_CHANNEL_2G)
- security_mode = config.get("security_mode", None)
- password = config.get("password", None)
- if security_mode:
- if not password:
+@unique
+class TestType(StrEnum):
+ ASSOCIATE_ONLY = "associate_only"
+ ASSOCIATE_AND_PING = "associate_and_ping"
+ ASSOCIATE_AND_PASS_TRAFFIC = "associate_and_pass_traffic"
+
+
+@dataclass
+class TestParams:
+ test_type: TestType
+ security_type: SecurityMode
+ connectivity_mode: ConnectivityMode
+ operating_band: OperatingBand
+ ssid: str
+ password: str
+ iterations: int
+
+
+@dataclass
+class APParams:
+ profile: str
+ ssid: str
+ channel: int
+ security: Security
+ password: str
+
+ @staticmethod
+ def from_dict(d: Dict[str, Any]) -> "APParams":
+ security_mode_str = get_typed(d, "security_mode", str, SecurityMode.OPEN.value)
+ security_mode = SecurityMode.from_str(security_mode_str)
+ password = get_typed(
+ d, "password", str, generate_random_password(security_mode=security_mode)
+ )
+
+ return APParams(
+ profile=get_typed(d, "profile", str, DEFAULT_AP_PROFILE),
+ ssid=get_typed(
+ d,
+ "ssid",
+ str,
+ utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G),
+ ),
+ channel=get_typed(
+ d, "channel", int, hostapd_constants.AP_DEFAULT_CHANNEL_2G
+ ),
+ security=Security(security_mode, password),
+ password=password,
+ )
+
+ def setup_ap(
+ self, access_point: AccessPoint, timeout_sec: int = DEFAULT_TIMEOUT
+ ) -> str:
+ """Setup access_point and return the IPv4 address of its test interface."""
+ setup_ap(
+ access_point=access_point,
+ profile_name=self.profile,
+ channel=self.channel,
+ ssid=self.ssid,
+ security=self.security,
+ password=self.password,
+ )
+
+ interface = access_point.wlan_2g if self.channel < 36 else access_point.wlan_5g
+
+ end_time = time.time() + timeout_sec
+ while time.time() < end_time:
+ ips = utils.get_interface_ip_addresses(access_point.ssh, interface)
+ if len(ips["ipv4_private"]) > 0:
+ return ips["ipv4_private"][0]
+ time.sleep(1)
+ raise ConnectionError(
+ f"After {timeout_sec}s, device {access_point.identifier} still does not have "
+ f"an ipv4 address on interface {interface}."
+ )
+
+
+@dataclass
+class SoftAPParams:
+ ssid: str
+ security_type: SecurityMode
+ password: Optional[str]
+ connectivity_mode: ConnectivityMode
+ operating_band: OperatingBand
+
+ def __str__(self) -> str:
+ if self.operating_band is OperatingBand.ANY:
+ band = "any"
+ elif self.operating_band is OperatingBand.ONLY_2G:
+ band = "2g"
+ elif self.operating_band is OperatingBand.ONLY_5G:
+ band = "5g"
+ else:
+ raise TypeError(f'Unknown OperatingBand "{self.operating_band}"')
+ return f'{band}_{self.security_type.replace("/", "_")}_{self.connectivity_mode}'
+
+ @staticmethod
+ def from_dict(d: Dict[str, Any]) -> "SoftAPParams":
+ security_type = get_typed(d, "security_type", str, SecurityMode.OPEN.value)
+ security_mode = SecurityMode.from_str(security_type)
+
+ password = d.get("password")
+ if password is None and security_mode is not SecurityMode.OPEN:
password = generate_random_password(security_mode=security_mode)
- security = hostapd_security.Security(security_mode, password)
- else:
- security = None
+ if password is not None and not isinstance(password, str):
+ raise TypeError(f'"password" must be a str or None, got {type(password)}')
+ if password is not None and security_mode is SecurityMode.OPEN:
+ raise TypeError(
+ f'"password" must be None if "security_type" is "{SecurityMode.OPEN}"'
+ )
- return {
- "profile": profile,
- "ssid": ssid,
- "channel": channel,
- "security": security,
- "password": password,
- }
+ connectivity_mode = get_typed(
+ d, "connectivity_mode", str, ConnectivityMode.LOCAL_ONLY.value
+ )
+ operating_band = get_typed(
+ d, "operating_band", str, OperatingBand.ONLY_2G.value
+ )
+
+ return SoftAPParams(
+ ssid=get_typed(
+ d,
+ "ssid",
+ str,
+ utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G),
+ ),
+ security_type=security_mode,
+ password=password,
+ connectivity_mode=ConnectivityMode.from_str(connectivity_mode),
+ operating_band=OperatingBand.from_str(operating_band),
+ )
-def get_soft_ap_params_from_config_or_default(config):
- """Retrieves SoftAp parameters from antlion config or returns default settings.
+@dataclass
+class AssociationStressTestParams:
+ test_type: TestType
+ soft_ap_params: SoftAPParams
+ iterations: int
- Args:
- config: dict, from antlion config, that may contain custom soft ap
- parameters
+ def __str__(self) -> str:
+ return f"{self.soft_ap_params}_{self.test_type}_{self.iterations}_iterations"
- Returns:
- dict, containing all soft AP parameters
- """
- ssid = config.get("ssid", utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G))
- connectivity_mode = config.get("connectivity_mode", CONNECTIVITY_MODE_LOCAL)
- operating_band = config.get("operating_band", OPERATING_BAND_2G)
- security_type = config.get("security_type", SECURITY_OPEN)
- password = config.get("password", "")
+ @staticmethod
+ def from_dict(d: Dict[str, Any]) -> "AssociationStressTestParams":
+ test_type = get_typed(
+ d, "test_type", str, TestType.ASSOCIATE_AND_PASS_TRAFFIC.value
+ )
+ return AssociationStressTestParams(
+ test_type=TestType.from_str(test_type),
+ soft_ap_params=SoftAPParams.from_dict(d.get("soft_ap_params", {})),
+ iterations=get_typed(d, "iterations", int, 10),
+ )
- # The SoftAP API uses 'open' security instead of None, '' password
- # instead of None, and security_type instead of security_mode, hence
- # the difference between ap_params and soft_ap_params
- if security_type != SECURITY_OPEN and password == "":
- password = generate_random_password(security_mode=security_type)
- return {
- "ssid": ssid,
- "connectivity_mode": connectivity_mode,
- "operating_band": operating_band,
- "security_type": security_type,
- "password": password,
- }
+@dataclass
+class ClientModeAlternatingTestParams:
+ ap_params: APParams
+ soft_ap_params: SoftAPParams
+ iterations: int
+
+ def __str__(self) -> str:
+ return (
+ f"ap_{self.ap_params.security.security_mode}_"
+ f"soft_ap_{self.soft_ap_params.security_type}_"
+ f"{self.iterations}_iterations"
+ )
+
+ @staticmethod
+ def from_dict(d: Dict[str, Any]) -> "ClientModeAlternatingTestParams":
+ return ClientModeAlternatingTestParams(
+ ap_params=APParams.from_dict(d.get("ap_params", {})),
+ soft_ap_params=SoftAPParams.from_dict(d.get("soft_ap_params", {})),
+ iterations=get_typed(d, "iterations", int, 10),
+ )
+
+
+@dataclass
+class ToggleTestParams:
+ soft_ap_params: SoftAPParams
+ iterations: int
+
+ def __str__(self) -> str:
+ return f"{self.soft_ap_params}_{self.iterations}_iterations"
+
+ @staticmethod
+ def from_dict(d: Dict[str, Any]) -> "ToggleTestParams":
+ return ToggleTestParams(
+ soft_ap_params=SoftAPParams.from_dict(d.get("soft_ap_params", {})),
+ iterations=get_typed(d, "iterations", int, 10),
+ )
+
+
+@dataclass
+class ClientModeToggleTestParams:
+ ap_params: APParams
+ iterations: int
+
+ def __str__(self) -> str:
+ return f"{self.ap_params}_{self.iterations}_iterations"
+
+ @staticmethod
+ def from_dict(d: Dict[str, Any]) -> "ClientModeToggleTestParams":
+ return ClientModeToggleTestParams(
+ ap_params=APParams.from_dict(d.get("ap_params", {})),
+ iterations=get_typed(d, "iterations", int, 10),
+ )
class StressTestIterationFailure(Exception):
@@ -144,12 +299,67 @@
skipped if physical AP is not present.
"""
+ def __init__(self, configs: TestRunConfig) -> None:
+ super().__init__(configs)
+ self.log = logging.getLogger()
+ self.soft_ap_test_params = configs.user_params.get("soft_ap_test_params", {})
+
+ def pre_run(self):
+ self.generate_soft_ap_tests()
+ self.generate_association_stress_tests()
+ self.generate_soft_ap_and_client_mode_alternating_stress_tests()
+ self.generate_soft_ap_toggle_stress_tests()
+ self.generate_client_mode_toggle_stress_tests()
+ self.generate_soft_ap_toggle_stress_with_client_mode_tests()
+ self.generate_client_mode_toggle_stress_with_soft_ap_tests()
+ self.generate_soft_ap_and_client_mode_random_toggle_stress_tests()
+
+ def generate_soft_ap_tests(self):
+ tests: List[SoftAPParams] = []
+
+ for operating_band in OperatingBand.all():
+ for security_mode in [
+ SecurityMode.OPEN,
+ SecurityMode.WEP,
+ SecurityMode.WPA,
+ SecurityMode.WPA2,
+ SecurityMode.WPA3,
+ ]:
+ for connectivity_mode in ConnectivityMode.all():
+ if security_mode is SecurityMode.OPEN:
+ ssid_length = hostapd_constants.AP_SSID_LENGTH_2G
+ password = None
+ else:
+ ssid_length = hostapd_constants.AP_SSID_LENGTH_5G
+ password = generate_random_password()
+
+ tests.append(
+ SoftAPParams(
+ ssid=utils.rand_ascii_str(ssid_length),
+ security_type=security_mode,
+ password=password,
+ connectivity_mode=connectivity_mode,
+ operating_band=operating_band,
+ )
+ )
+
+ def generate_name(test: SoftAPParams) -> str:
+ return f"test_soft_ap_{test}"
+
+ self.generate_tests(
+ self.associate_with_soft_ap_test,
+ generate_name,
+ tests,
+ )
+
+ def associate_with_soft_ap_test(self, soft_ap_params: SoftAPParams):
+ self.start_soft_ap(soft_ap_params)
+ self.associate_with_soft_ap(self.primary_client, soft_ap_params)
+ self.assert_connected_to_ap(self.primary_client, self.dut, check_traffic=True)
+
def setup_class(self):
super().setup_class()
- self.log = logging.getLogger()
- self.soft_ap_test_params = self.user_params.get("soft_ap_test_params", {})
-
if len(self.fuchsia_devices) < 1:
raise signals.TestAbortClass("At least one Fuchsia device is required")
self.fuchsia_device = self.fuchsia_devices[0]
@@ -157,8 +367,8 @@
# TODO(fxb/51313): Add in device agnosticity for clients
# Create a wlan device and iperf client for each Android client
- self.clients = []
- self.iperf_clients_map: Dict[Any, iperf_client.IPerfClientBase] = {}
+ self.clients: List[SupportsWLAN] = []
+ self.iperf_clients_map: Dict[Any, Any] = {}
for device in self.android_devices:
client_wlan_device = create_wlan_device(device)
self.clients.append(client_wlan_device)
@@ -190,11 +400,10 @@
self.ap_iperf_client = iperf_client.IPerfClientOverSsh(
self.access_point.ssh_provider,
)
+ self.iperf_clients_map[self.access_point] = self.ap_iperf_client
except AttributeError:
pass
- self.iperf_clients_map[self.access_point] = self.ap_iperf_client
-
def teardown_class(self):
# Because this is using killall, it will stop all iperf processes
self.iperf_server.stop()
@@ -227,7 +436,7 @@
self.dut.disconnect()
super().teardown_test()
- def start_soft_ap(self, settings):
+ def start_soft_ap(self, params: SoftAPParams) -> None:
"""Starts a softAP on Fuchsia device.
Args:
@@ -241,25 +450,21 @@
operating_band: string, band for softAP network
- 'any', 'only_5_ghz', 'only_2_4_ghz'
"""
- ssid = settings["ssid"]
- security_type = settings["security_type"]
- password = settings.get("password", "")
- connectivity_mode = settings["connectivity_mode"]
- operating_band = settings["operating_band"]
-
- self.log.info(f"Starting SoftAP on DUT with settings: {settings}")
-
+ self.log.info(f"Starting SoftAP on DUT with settings: {params}")
response = self.fuchsia_device.sl4f.wlan_ap_policy_lib.wlanStartAccessPoint(
- ssid, security_type, password, connectivity_mode, operating_band
+ params.ssid,
+ params.security_type.fuchsia_security_type(),
+ params.password,
+ params.connectivity_mode,
+ params.operating_band,
)
if response.get("error"):
raise EnvironmentError(
f"SL4F: Failed to setup SoftAP. Err: {response['error']}"
)
+ self.log.info(f"SoftAp network ({params.ssid}) is up.")
- self.log.info(f"SoftAp network ({ssid}) is up.")
-
- def stop_soft_ap(self, settings):
+ def stop_soft_ap(self, params: SoftAPParams) -> None:
"""Stops a specific SoftAP On Fuchsia device.
Args:
@@ -269,19 +474,15 @@
Raises:
EnvironmentError, if StopSoftAP call fails.
"""
- ssid = settings["ssid"]
- security_type = settings["security_type"]
- password = settings.get("password", "")
-
response = self.fuchsia_device.sl4f.wlan_ap_policy_lib.wlanStopAccessPoint(
- ssid, security_type, password
+ params.ssid, params.security_type.fuchsia_security_type(), params.password
)
if response.get("error"):
raise EnvironmentError(
f"SL4F: Failed to stop SoftAP. Err: {response['error']}"
)
- def stop_all_soft_aps(self):
+ def stop_all_soft_aps(self) -> None:
"""Stops all SoftAPs on Fuchsia Device.
Raises:
@@ -293,42 +494,34 @@
f"SL4F: Failed to stop all SoftAPs. Err: {response['error']}"
)
- def associate_with_soft_ap(self, device, soft_ap_settings):
+ def associate_with_soft_ap(self, device: SupportsWLAN, params: SoftAPParams):
"""Associates client device with softAP on Fuchsia device.
Args:
device: wlan_device to associate with the softAP
- settings: a dict containing softAP config params (see start_soft_ap)
- for details
+ params: soft AP configuration
Raises:
- TestFailure, if association fails
+ TestFailure if association fails
"""
self.log.info(
- f"Attempting to associate client {device.identifier} with SoftAP on "
- f"FuchsiaDevice ({self.dut.identifier})."
+ f'Associating {device.identifier} to SoftAP on {self.dut.identifier} called "{params.ssid}'
)
- check_connectivity = (
- soft_ap_settings["connectivity_mode"] == CONNECTIVITY_MODE_UNRESTRICTED
- )
associated = device.associate(
- soft_ap_settings["ssid"],
- target_pwd=soft_ap_settings.get("password"),
- target_security=hostapd_constants.SECURITY_STRING_TO_DEFAULT_TARGET_SECURITY.get(
- soft_ap_settings["security_type"], None
- ),
- check_connectivity=check_connectivity,
+ params.ssid,
+ target_pwd=params.password,
+ target_security=params.security_type,
+ check_connectivity=params.connectivity_mode
+ is ConnectivityMode.UNRESTRICTED,
)
- if not associated:
- self.log.error("Failed to connect to SoftAp.")
- return False
+ asserts.assert_true(
+ associated,
+ f'Failed to associate "{device.identifier}" to SoftAP "{params.ssid}"',
+ )
- self.log.info("Client successfully associated with SoftAP.")
- return True
-
- def disconnect_from_soft_ap(self, device):
+ def disconnect_from_soft_ap(self, device: SupportsWLAN) -> None:
"""Disconnects client device from SoftAP.
Args:
@@ -337,7 +530,15 @@
self.log.info(f"Disconnecting device {device.identifier} from SoftAP.")
device.disconnect()
- def get_device_test_interface(self, device, role=None, channel=None):
+ def get_ap_test_interface(self, ap: AccessPoint, channel: int) -> str:
+ if channel < 36:
+ return ap.wlan_2g
+ else:
+ return ap.wlan_5g
+
+ def get_device_test_interface(
+ self, device: Union[SupportsWLAN, FuchsiaDevice], role: DeviceRole
+ ) -> str:
"""Retrieves test interface from a provided device, which can be the
FuchsiaDevice DUT, the AccessPoint, or an AndroidClient.
@@ -345,31 +546,27 @@
device: the device do get the test interface from. Either
FuchsiaDevice (DUT), Android client, or AccessPoint.
role: str, either "client" or "ap". Required for FuchsiaDevice (DUT)
- channel: int, channel of the ap network. Required for AccessPoint.
Returns:
String, name of test interface on given device.
"""
- if device is self.dut:
- device.device.wlan_controller.update_wlan_interfaces()
- if role == INTERFACE_ROLE_CLIENT:
- return device.device.wlan_client_test_interface_name
- elif role == INTERFACE_ROLE_AP:
- return device.device.wlan_ap_test_interface_name
- else:
- raise ValueError(f"Unsupported interface role: {role}")
- elif isinstance(device, AccessPoint):
- if not channel:
- raise ValueError("Must provide a channel to get AccessPoint interface")
- if channel < 36:
- return device.wlan_2g
- else:
- return device.wlan_5g
+ if isinstance(device, FuchsiaDevice):
+ device.wlan_controller.update_wlan_interfaces()
+ if role is DeviceRole.CLIENT:
+ return device.wlan_client_test_interface_name
+ if role is DeviceRole.AP:
+ return device.wlan_ap_test_interface_name
+ raise ValueError(f"Unsupported interface role: {role}")
else:
return device.get_default_wlan_test_interface()
- def wait_for_ipv4_address(self, device, interface_name, timeout=DEFAULT_TIMEOUT):
+ def wait_for_ipv4_address(
+ self,
+ device: Union[SupportsWLAN, AccessPoint],
+ interface_name: str,
+ timeout: int = DEFAULT_TIMEOUT,
+ ):
"""Waits for interface on a wlan_device to get an ipv4 address.
Args:
@@ -378,12 +575,18 @@
timeout: seconds to wait before raising an error
Raises:
- ValueError, if interface does not have an ipv4 address after timeout
+ ConnectionError, if interface does not have an ipv4 address after timeout
"""
+ comm_channel: Union[SshConnection, FuchsiaDevice, AndroidDevice]
if isinstance(device, AccessPoint):
comm_channel = device.ssh
- else:
+ elif isinstance(device, FuchsiaWlanDevice):
comm_channel = device.device
+ elif isinstance(device, AndroidWlanDevice):
+ comm_channel = device.device
+ else:
+ raise TypeError(f"Invalid device type {type(device)}")
+
end_time = time.time() + timeout
while time.time() < end_time:
ips = utils.get_interface_ip_addresses(comm_channel, interface_name)
@@ -400,40 +603,14 @@
f"an ipv4 address on interface {interface_name}."
)
- def device_can_ping_addr(self, device, dest_ip, timeout=DEFAULT_TIMEOUT):
- """Verify wlan_device can ping a destination ip.
-
- Args:
- device: wlan_device to initiate ping
- dest_ip: ip to ping from wlan_device
-
- Raises:
- TestFailure, if ping fails
- """
- end_time = time.time() + timeout
- while time.time() < end_time:
- with utils.SuppressLogOutput():
- ping_result = device.can_ping(dest_ip)
-
- if ping_result:
- self.log.info(
- f"Ping successful from device {device.identifier} to "
- f"dest ip {dest_ip}."
- )
- return True
- else:
- self.log.debug(
- f"Device {device.identifier} could not ping dest ip {dest_ip}. "
- "Retrying in 1 second."
- )
- time.sleep(1)
- else:
- self.log.info(
- f"Failed to ping from device {device.identifier} to dest ip {dest_ip}."
- )
- return False
-
- def run_iperf_traffic(self, ip_client, server_address, server_port=5201):
+ def run_iperf_traffic(
+ self,
+ ip_client: Union[
+ iperf_client.IPerfClientOverAdb, iperf_client.IPerfClientOverSsh
+ ],
+ server_address: str,
+ server_port: int = 5201,
+ ) -> None:
"""Runs traffic between client and ap an verifies throughput.
Args:
@@ -442,7 +619,7 @@
server_port: port of the iperf server
Raises:
- TestFailure, if no traffic passes in either direction
+ ConnectionError if no traffic passes in either direction
"""
ip_client_identifier = self.get_iperf_client_identifier(ip_client)
@@ -507,175 +684,135 @@
f"In iperf process from {self.get_iperf_client_identifier(ip_client)} to {server_address}: {err}"
)
- def get_iperf_client_identifier(self, ip_client):
- """Retrieves an indentifer string from iperf client, for logging.
+ def get_iperf_client_identifier(
+ self,
+ ip_client: Union[
+ iperf_client.IPerfClientOverAdb, iperf_client.IPerfClientOverSsh
+ ],
+ ) -> str:
+ """Retrieves an identifier string from iperf client, for logging.
Args:
ip_client: iperf client to grab identifier from
"""
if type(ip_client) == iperf_client.IPerfClientOverAdb:
return ip_client._android_device_or_serial.serial
- return ip_client._ssh_settings.hostname
+ if type(ip_client) == iperf_client.IPerfClientOverSsh:
+ return ip_client._ssh_provider.config.host_name
+ raise TypeError(f'Unknown "ip_client" type {type(ip_client)}')
- def device_is_connected_to_ap(
- self, client, ap, channel=None, check_traffic=False, timeout=DEFAULT_TIMEOUT
- ):
- """Returns whether client device can ping (and optionally pass traffic)
- to the ap device.
+ def assert_connected_to_ap(
+ self,
+ client: SupportsWLAN,
+ ap: Union[SupportsWLAN, AccessPoint],
+ channel: Optional[int] = None,
+ check_traffic: bool = False,
+ timeout_sec: int = DEFAULT_TIMEOUT,
+ ) -> None:
+ """Assert the client device has L3 connectivity to the AP."""
+ device_interface = self.get_device_test_interface(client, DeviceRole.CLIENT)
- Args:
- client: device that should be associated. Either FuchsiaDevice (DUT)
- or Android client
- ap: device acting as AP. Either FuchsiaDevice (DUT) or AccessPoint.
- channel: int, channel the AP is using. Required if ap is an
- AccessPoint object.
- check_traffic: bool, whether to attempt to pass traffic between
- client and ap devices.
- timeout: int, time in seconds to wait for devices to have ipv4
- addresses
- """
+ if isinstance(ap, AccessPoint):
+ if channel is None:
+ raise TypeError("channel must not be None when ap is an AccessPoint")
+ ap_interface = self.get_ap_test_interface(ap, channel)
+ else:
+ ap_interface = self.get_device_test_interface(ap, DeviceRole.AP)
+
+ client_ipv4 = self.wait_for_ipv4_address(
+ client, device_interface, timeout=timeout_sec
+ )
+ ap_ipv4 = self.wait_for_ipv4_address(ap, ap_interface, timeout=timeout_sec)
+
+ asserts.assert_true(
+ client.can_ping(ap_ipv4, timeout=DEFAULT_TIMEOUT * 1000),
+ "Failed to ping from client to ap",
+ )
+ asserts.assert_true(
+ ap.can_ping(client_ipv4, timeout=DEFAULT_TIMEOUT * 1000),
+ "Failed to ping from ap to client",
+ )
+
+ if not check_traffic:
+ return
+
+ if client is self.dut:
+ self.run_iperf_traffic(self.iperf_clients_map[ap], client_ipv4)
+ else:
+ self.run_iperf_traffic(self.iperf_clients_map[client], ap_ipv4)
+
+ def assert_disconnected_to_ap(
+ self,
+ client: SupportsWLAN,
+ ap: Union[SupportsWLAN, AccessPoint],
+ channel: Optional[int] = None,
+ timeout_sec: int = DEFAULT_NO_ADDR_EXPECTED_TIMEOUT,
+ ) -> None:
+ """Assert the client device does not have ping connectivity to the AP."""
+ device_interface = self.get_device_test_interface(client, DeviceRole.CLIENT)
+
+ if isinstance(ap, AccessPoint):
+ if channel is None:
+ raise TypeError("channel must not be None when ap is an AccessPoint")
+ ap_interface = self.get_ap_test_interface(ap, channel)
+ else:
+ ap_interface = self.get_device_test_interface(ap, DeviceRole.AP)
+
try:
- # Get interfaces
- client_interface = self.get_device_test_interface(
- client, INTERFACE_ROLE_CLIENT
- )
- ap_interface = self.get_device_test_interface(
- ap, role=INTERFACE_ROLE_AP, channel=channel
- )
-
- # Get addresses
client_ipv4 = self.wait_for_ipv4_address(
- client, client_interface, timeout=timeout
+ client, device_interface, timeout=timeout_sec
)
- ap_ipv4 = self.wait_for_ipv4_address(ap, ap_interface, timeout=timeout)
- except ConnectionError as err:
- self.log.error(f"Failed to retrieve interfaces and addresses. Err: {err}")
- return False
+ ap_ipv4 = self.wait_for_ipv4_address(ap, ap_interface, timeout=timeout_sec)
+ except ConnectionError:
+ # When disconnected, IP addresses aren't always available.
+ return
- if not self.device_can_ping_addr(client, ap_ipv4):
- self.log.error("Failed to ping from client to ap.")
- return False
-
- if not self.device_can_ping_addr(ap, client_ipv4):
- self.log.error("Failed to ping from ap to client.")
- return False
-
- if check_traffic:
- try:
- if client is self.dut:
- self.run_iperf_traffic(self.iperf_clients_map[ap], client_ipv4)
- else:
- self.run_iperf_traffic(self.iperf_clients_map[client], ap_ipv4)
- except ConnectionError as err:
- self.log.error(f"Failed to run traffic between DUT and AP: {err}")
- return False
- return True
-
- def verify_soft_ap_connectivity_from_state(self, state, client):
- """Verifies SoftAP state based on a client connection.
-
- Args:
- state: bool, whether SoftAP should be up
- client: SoftApClient, to verify connectivity (or lack therof)
- """
- if state == STATE_UP:
- return self.device_is_connected_to_ap(client, self.dut)
- else:
- with utils.SuppressLogOutput():
- try:
- return not self.device_is_connected_to_ap(
- client, self.dut, timeout=DEFAULT_NO_ADDR_EXPECTED_TIMEOUT
- )
- # Allow a failed to find ap interface error
- except LookupError as err:
- self.log.debug(f"Hit expected LookupError: {err}")
- return True
-
- def verify_client_mode_connectivity_from_state(self, state, channel):
- """Verifies client mode state based on DUT-AP connection.
-
- Args:
- state: bool, whether client mode should be up
- channel: int, channel of the APs network
- """
- if state == STATE_UP:
- return self.device_is_connected_to_ap(
- self.dut, self.access_point, channel=channel
- )
- else:
- with utils.SuppressLogOutput():
- try:
- return not self.device_is_connected_to_ap(
- self.dut,
- self.access_point,
- channel=channel,
- timeout=DEFAULT_NO_ADDR_EXPECTED_TIMEOUT,
- )
- # Allow a failed to find client interface error
- except LookupError as err:
- self.log.debug(f"Hit expected LookupError: {err}")
- return True
-
- # Test Types
-
- def verify_soft_ap_associate_only(self, client, soft_ap_settings):
- if not self.associate_with_soft_ap(client, soft_ap_settings):
- asserts.fail("Failed to associate client with SoftAP.")
-
- def verify_soft_ap_associate_and_ping(self, client, soft_ap_settings):
- self.verify_soft_ap_associate_only(client, soft_ap_settings)
- if not self.device_is_connected_to_ap(client, self.dut):
- asserts.fail("Client and SoftAP could not ping eachother.")
-
- def verify_soft_ap_associate_and_pass_traffic(self, client, settings):
- self.verify_soft_ap_associate_only(client, settings)
- if not self.device_is_connected_to_ap(client, self.dut, check_traffic=True):
- asserts.fail(
- "Client and SoftAP not responding to pings and passing traffic "
- "as expected."
- )
+ asserts.assert_false(
+ client.can_ping(ap_ipv4, timeout=DEFAULT_TIMEOUT * 1000),
+ "Unexpectedly succeeded to ping from client to ap",
+ )
+ asserts.assert_false(
+ ap.can_ping(client_ipv4, timeout=DEFAULT_TIMEOUT * 1000),
+ "Unexpectedly succeeded to ping from ap to client",
+ )
# Runners for Generated Test Cases
- def run_soft_ap_association_stress_test(self, settings):
- """Sets up a SoftAP, and repeatedly associates and disassociates a
- client.
-
- Args:
- settings: test configuration settings, see
- test_soft_ap_association_stress for details
- """
- client = settings["client"]
- soft_ap_params = settings["soft_ap_params"]
- test_type = settings["test_type"]
- if not test_type in TEST_TYPES:
- raise ValueError(f"Unrecognized test type {test_type}")
- iterations = settings["iterations"]
+ def run_soft_ap_association_stress_test(self, test: AssociationStressTestParams):
+ """Sets up a SoftAP, and repeatedly associates and disassociates a client."""
self.log.info(
- f"Running association stress test type {test_type} in "
- f"iteration {iterations} times"
+ f"Running association stress test type {test.test_type} in "
+ f"iteration {test.iterations} times"
)
- self.start_soft_ap(soft_ap_params)
+ self.start_soft_ap(test.soft_ap_params)
passed_count = 0
- for run in range(iterations):
+ for run in range(test.iterations):
try:
self.log.info(f"Starting SoftAp association run {str(run + 1)}")
- if test_type == TEST_TYPE_ASSOCIATE_ONLY:
- self.verify_soft_ap_associate_only(client, soft_ap_params)
+ if test.test_type == TestType.ASSOCIATE_ONLY:
+ self.associate_with_soft_ap(
+ self.primary_client, test.soft_ap_params
+ )
- elif test_type == TEST_TYPE_ASSOCIATE_AND_PING:
- self.verify_soft_ap_associate_and_ping(client, soft_ap_params)
+ elif test.test_type == TestType.ASSOCIATE_AND_PING:
+ self.associate_with_soft_ap(
+ self.primary_client, test.soft_ap_params
+ )
+ self.assert_connected_to_ap(self.primary_client, self.dut)
- elif test_type == TEST_TYPE_ASSOCIATE_AND_PASS_TRAFFIC:
- self.verify_soft_ap_associate_and_pass_traffic(
- client, soft_ap_params
+ elif test.test_type == TestType.ASSOCIATE_AND_PASS_TRAFFIC:
+ self.associate_with_soft_ap(
+ self.primary_client, test.soft_ap_params
+ )
+ self.assert_connected_to_ap(
+ self.primary_client, self.dut, check_traffic=True
)
else:
- raise AttributeError(f"Invalid test type: {test_type}")
+ raise AttributeError(f"Invalid test type: {test.test_type}")
except signals.TestFailure as err:
self.log.error(
@@ -688,148 +825,47 @@
)
passed_count += 1
- if passed_count < iterations:
+ if passed_count < test.iterations:
asserts.fail(
"SoftAp association stress test failed after "
- f"{passed_count}/{iterations} runs."
+ f"{passed_count}/{test.iterations} runs."
)
asserts.explicit_pass(
- f"SoftAp association stress test passed after {passed_count}/{iterations} "
+ f"SoftAp association stress test passed after {passed_count}/{test.iterations} "
"runs."
)
# Alternate SoftAP and Client mode test
- def run_soft_ap_and_client_mode_alternating_test(self, settings):
+ def run_soft_ap_and_client_mode_alternating_test(
+ self, test: ClientModeAlternatingTestParams
+ ):
"""Runs a single soft_ap and client alternating stress test.
See test_soft_ap_and_client_mode_alternating_stress for details.
"""
- iterations = settings["iterations"]
- pass_count = 0
- current_soft_ap_state = STATE_DOWN
- current_client_mode_state = STATE_DOWN
+ if self.access_point is None:
+ raise signals.TestSkip("No access point provided")
- self.client_mode_toggle_pre_test(settings)
- for iteration in range(iterations):
- passes = True
+ test.ap_params.setup_ap(self.access_point)
- # Attempt to toggle SoftAP on, then off. If the first toggle fails
- # to occur, exit early.
- for _ in range(2):
- (current_soft_ap_state, err) = self.run_toggle_iteration_func(
- self.soft_ap_toggle_test_iteration, settings, current_soft_ap_state
- )
- if err:
- self.log.error(f"Iteration {str(iteration + 1)} failed. Err: {err}")
- passes = False
- if current_soft_ap_state == STATE_DOWN:
- break
+ for _ in range(test.iterations):
+ # Toggle SoftAP on then off.
+ self.toggle_soft_ap(test.soft_ap_params, STATE_DOWN)
+ self.toggle_soft_ap(test.soft_ap_params, STATE_UP)
- # Attempt to toggle Client mode on, then off. If the first toggle,
- # fails to occur, exit early.
- for _ in range(2):
- (current_client_mode_state, err) = self.run_toggle_iteration_func(
- self.client_mode_toggle_test_iteration,
- settings,
- current_client_mode_state,
- )
- if err:
- self.log.error(f"Iteration {str(iteration + 1)} failed. Err: {err}")
- passes = False
- if current_client_mode_state == STATE_DOWN:
- break
-
- if passes:
- pass_count += 1
-
- if pass_count == iterations:
- asserts.explicit_pass(
- "Toggle SoftAP and client mode stress test passed "
- f"{pass_count}/{iterations} times."
- )
- else:
- asserts.fail(
- "Toggle SoftAP and client mode stress test only passed "
- f"{pass_count}/{iterations} times."
- )
+ # Toggle client mode on then off.
+ self.toggle_client_mode(self.access_point, test.ap_params, STATE_DOWN)
+ self.toggle_client_mode(self.access_point, test.ap_params, STATE_UP)
# Toggle Stress Test Helper Functions
- def run_toggle_stress_test(self, settings):
- """Runner function for toggle stress tests.
-
- Repeats some test function through stress test iterations, logging
- failures, tracking pass rate, managing states, etc.
-
- Args:
- settings: dict, stress test settings
-
- Asserts:
- PASS: if all iterations of the test function pass
- FAIL: if any iteration of the test function fails
- """
- test_runner_func = settings["test_runner_func"]
- pre_test_func = settings.get("pre_test_func", None)
- iterations = settings["iterations"]
- if pre_test_func:
- pre_test_func(settings)
-
- pass_count = 0
- current_state = STATE_DOWN
- for iteration in range(iterations):
- (current_state, err) = self.run_toggle_iteration_func(
- test_runner_func, settings, current_state
- )
- if err:
- self.log.error(f"Iteration {str(iteration + 1)} failed. Err: {err}")
- else:
- pass_count += 1
-
- if pass_count == iterations:
- asserts.explicit_pass(
- f"Stress test passed {pass_count}/{iterations} times."
- )
- else:
- asserts.fail(f"Stress test only passed {pass_count}/{iterations} times.")
-
- def run_toggle_iteration_func(self, func, settings, current_state):
- """Runs a toggle iteration function, updating the current state
- based on what the toggle iteration function raises.
-
- Used for toggle stress tests.
-
- Note on EnvironmentError vs StressTestIterationFailure:
- StressTestIterationFailure is raised by func when the toggle occurs
- but connectivty or some other post-toggle check fails (i.e. the
- next iteration should toggle to the next state.)
-
- EnvironmentError is raise by func when the toggle itself fails (i.e
- the next iteration should retry the same toggle again.)
-
- Args:
- func: toggle iteration func to run (e.g soft_ap_toggle_iteration)
- settings: dict, stress test settings
- current_state: bool, the current state of the mode being toggled
-
- Returns:
- (new_state, err):
- new_state: bool, state of the mode after toggle attempt
- err: exception, if any are raise, else None
- """
- try:
- func(settings, current_state)
- except EnvironmentError as err:
- return (current_state, err)
- except StressTestIterationFailure as err:
- return (not current_state, err)
- else:
- return (not current_state, None)
-
# Stress Test Toggle Functions
- def start_soft_ap_and_verify_connected(self, client, soft_ap_params):
+ def start_soft_ap_and_verify_connected(
+ self, client: SupportsWLAN, soft_ap_params: SoftAPParams
+ ):
"""Sets up SoftAP, associates a client, then verifies connection.
Args:
@@ -841,21 +877,10 @@
is not functioning as expected
"""
# Change SSID every time, to avoid client connection issues.
- soft_ap_params["ssid"] = utils.rand_ascii_str(
- hostapd_constants.AP_SSID_LENGTH_2G
- )
+ soft_ap_params.ssid = utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G)
self.start_soft_ap(soft_ap_params)
- associated = self.associate_with_soft_ap(client, soft_ap_params)
- if not associated:
- raise StressTestIterationFailure(
- "Failed to associated client to DUT SoftAP. "
- "Continuing with iterations."
- )
-
- if not self.verify_soft_ap_connectivity_from_state(STATE_UP, client):
- raise StressTestIterationFailure(
- "Failed to ping between client and DUT. Continuing " "with iterations."
- )
+ self.associate_with_soft_ap(client, soft_ap_params)
+ self.assert_connected_to_ap(client, self.dut)
def stop_soft_ap_and_verify_disconnected(self, client, soft_ap_params):
"""Tears down SoftAP, and verifies connection is down.
@@ -869,13 +894,11 @@
"""
self.log.info("Stopping SoftAP on DUT.")
self.stop_soft_ap(soft_ap_params)
+ self.assert_disconnected_to_ap(client, self.dut)
- if not self.verify_soft_ap_connectivity_from_state(STATE_DOWN, client):
- raise EnvironmentError(
- "Client can still ping DUT. Continuing with " "iterations."
- )
-
- def start_client_mode_and_verify_connected(self, ap_params):
+ def start_client_mode_and_verify_connected(
+ self, access_point: AccessPoint, ap_params: APParams
+ ):
"""Connects DUT to AP in client mode and verifies connection
Args:
@@ -886,53 +909,47 @@
StressTestIterationFailure, if DUT associates but connection is not
functioning as expected.
"""
- ap_ssid = ap_params["ssid"]
- ap_password = ap_params["password"]
- ap_channel = ap_params["channel"]
- ap_security = ap_params.get("security")
-
- if ap_security:
- ap_security_mode = ap_security.security_mode_string
- else:
- ap_security_mode = None
-
- self.log.info(f"Associating DUT with AP network: {ap_ssid}")
+ self.log.info(f"Associating DUT with AP network: {ap_params.ssid}")
associated = self.dut.associate(
- target_ssid=ap_ssid,
- target_pwd=ap_password,
- target_security=hostapd_constants.SECURITY_STRING_TO_DEFAULT_TARGET_SECURITY.get(
- ap_security_mode, None
- ),
+ target_ssid=ap_params.ssid,
+ target_pwd=ap_params.password,
+ target_security=ap_params.security.security_mode,
)
if not associated:
raise EnvironmentError("Failed to associate DUT in client mode.")
else:
self.log.info("Association successful.")
- if not self.verify_client_mode_connectivity_from_state(STATE_UP, ap_channel):
- raise StressTestIterationFailure("Failed to ping AP from DUT.")
+ self.assert_connected_to_ap(self.dut, access_point, channel=ap_params.channel)
- def stop_client_mode_and_verify_disconnected(self, ap_params):
+ def stop_client_mode_and_verify_disconnected(
+ self, access_point: AccessPoint, ap_params: APParams
+ ):
"""Disconnects DUT from AP and verifies connection is down.
Args:
- ap_params: dict, containing parameters of the AP network
+ ap_params: containing parameters of the AP network
Raises:
EnvironmentError, if DUT and AP can still communicate
"""
self.log.info("Disconnecting DUT from AP.")
self.dut.disconnect()
- if not self.verify_client_mode_connectivity_from_state(
- STATE_DOWN, ap_params["channel"]
- ):
- raise EnvironmentError("DUT can still ping AP.")
+ self.assert_disconnected_to_ap(
+ self.dut, access_point, channel=ap_params.channel
+ )
# Toggle Stress Test Iteration and Pre-Test Functions
# SoftAP Toggle Stress Test Helper Functions
- def soft_ap_toggle_test_iteration(self, settings, current_state):
+ def soft_ap_toggle_test(self, test: ToggleTestParams) -> None:
+ current_state = STATE_DOWN
+ for i in range(test.iterations):
+ self.toggle_soft_ap(test.soft_ap_params, current_state)
+ current_state = not current_state
+
+ def toggle_soft_ap(self, soft_ap_params: SoftAPParams, current_state: bool):
"""Runs a single iteration of SoftAP toggle stress test
Args:
@@ -945,12 +962,9 @@
functioning correctly.
EnvironmentError, if toggle fails to occur at all
"""
- soft_ap_params = settings["soft_ap_params"]
self.log.info(f"Toggling SoftAP {'down' if current_state else 'up'}.")
-
if current_state == STATE_DOWN:
self.start_soft_ap_and_verify_connected(self.primary_client, soft_ap_params)
-
else:
self.stop_soft_ap_and_verify_disconnected(
self.primary_client, soft_ap_params
@@ -958,27 +972,35 @@
# Client Mode Toggle Stress Test Helper Functions
- def client_mode_toggle_pre_test(self, settings):
- """Prepares the AP before client mode toggle tests
+ def client_mode_toggle_test(self, test: ClientModeToggleTestParams) -> None:
+ if self.access_point is None:
+ raise signals.TestSkip("No access point provided")
- Args:
- settings: dict, stress test settings
+ test.ap_params.setup_ap(self.access_point)
- Raises:
- ConnectionError, if AP setup fails
- """
- ap_params = settings["ap_params"]
- ap_channel = ap_params["channel"]
- ap_profile = ap_params.pop("profile")
- self.log.info(f"Setting up AP with params: {ap_params}")
- setup_ap(access_point=self.access_point, profile_name=ap_profile, **ap_params)
- # Confirms AP assigned itself an address
- ap_interface = self.get_device_test_interface(
- self.access_point, channel=ap_channel
- )
- self.wait_for_ipv4_address(self.access_point, ap_interface)
+ current_state = STATE_DOWN
+ for i in range(test.iterations):
+ self.log.info(
+ f"Iteration {i}: toggling client mode {'off' if current_state else 'on'}."
+ )
+ self.toggle_client_mode(self.access_point, test.ap_params, current_state)
+ current_state = not current_state
- def client_mode_toggle_test_iteration(self, settings, current_state):
+ def toggle_client_mode(
+ self, access_point: AccessPoint, ap_params: APParams, current_state: bool
+ ) -> None:
+ if current_state == STATE_DOWN:
+ self.start_client_mode_and_verify_connected(access_point, ap_params)
+ else:
+ self.stop_client_mode_and_verify_disconnected(access_point, ap_params)
+
+ # TODO: Remove
+ def client_mode_toggle_test_iteration(
+ self,
+ test: ClientModeToggleTestParams,
+ access_point: AccessPoint,
+ current_state: bool,
+ ):
"""Runs a single iteration of client mode toggle stress test
Args:
@@ -991,555 +1013,91 @@
functioning correctly.
EnvironmentError, if toggle fails to occur at all
"""
- ap_params = settings["ap_params"]
self.log.info(f"Toggling client mode {'off' if current_state else 'on'}")
-
if current_state == STATE_DOWN:
- self.start_client_mode_and_verify_connected(ap_params)
-
+ self.start_client_mode_and_verify_connected(access_point, test.ap_params)
else:
- self.stop_client_mode_and_verify_disconnected(ap_params)
+ self.stop_client_mode_and_verify_disconnected(access_point, test.ap_params)
# Toggle SoftAP with Client Mode Up Test Helper Functions
- def soft_ap_toggle_with_client_mode_pre_test(self, settings):
- """Sets up and verifies client mode before SoftAP toggle test.
- Args:
- settings: dict, stress test settings
+ def soft_ap_toggle_with_client_mode_test(
+ self, test: ClientModeAlternatingTestParams
+ ) -> None:
+ if self.access_point is None:
+ raise signals.TestSkip("No access point provided")
- Raises:
- ConnectionError, if client mode setup fails
- """
- self.client_mode_toggle_pre_test(settings)
- try:
- self.start_client_mode_and_verify_connected(settings["ap_params"])
- except StressTestIterationFailure as err:
- # This prevents it being treated as a routine error
- raise ConnectionError(
- f"Failed to set up DUT client mode before SoftAP toggle test.Err: {err}"
+ test.ap_params.setup_ap(self.access_point)
+ self.start_client_mode_and_verify_connected(self.access_point, test.ap_params)
+
+ current_state = STATE_DOWN
+ for i in range(test.iterations):
+ self.toggle_soft_ap(test.soft_ap_params, current_state)
+ self.assert_connected_to_ap(
+ self.dut, self.access_point, channel=test.ap_params.channel
)
-
- def soft_ap_toggle_with_client_mode_iteration(
- self,
- settings,
- current_state,
- ):
- """Runs single iteration of SoftAP toggle stress with client mode test.
-
- Args:
- settings: dict, containing test settings
- current_state: bool, current state of SoftAP (True if up,
- else False)
-
- Raises:
- StressTestIterationFailure, if toggle occurs but mode isn't
- functioning correctly.
- EnvironmentError, if toggle fails to occur at all
- """
- ap_params = settings["ap_params"]
- ap_channel = ap_params["channel"]
- self.soft_ap_toggle_test_iteration(settings, current_state)
- if not self.device_is_connected_to_ap(
- self.dut, self.access_point, channel=ap_channel
- ):
- raise StressTestIterationFailure(
- "DUT client mode is no longer functional after SoftAP toggle."
- )
+ current_state = not current_state
# Toggle Client Mode with SoftAP Up Test Helper Functions
- def client_mode_toggle_with_soft_ap_pre_test(self, settings):
- """Sets up and verifies softap before client mode toggle test.
- Args:
- settings: dict, stress test settings
+ def client_mode_toggle_with_soft_ap_test(
+ self, test: ClientModeAlternatingTestParams
+ ) -> None:
+ if self.access_point is None:
+ raise signals.TestSkip("No access point provided")
- Raises:
- ConnectionError, if softap setup fails
- """
- self.client_mode_toggle_pre_test(settings)
- try:
- self.start_soft_ap_and_verify_connected(
- self.primary_client, settings["soft_ap_params"]
- )
- except StressTestIterationFailure as err:
- # This prevents it being treated as a routine error
- raise ConnectionError(
- f"Failed to set up SoftAP before client mode toggle test. Err: {err}"
- )
+ test.ap_params.setup_ap(self.access_point)
+ self.start_soft_ap_and_verify_connected(
+ self.primary_client, test.soft_ap_params
+ )
- def client_mode_toggle_with_soft_ap_iteration(self, settings, current_state):
- """Runs single iteration of client mode toggle stress with SoftAP test.
-
- Args:
- settings: dict, containing test settings
- current_state: bool, current state of client mode (True if up,
- else False)
-
- Raises:
- StressTestIterationFailure, if toggle occurs but mode isn't
- functioning correctly.
- EnvironmentError, if toggle fails to occur at all
- """
- self.client_mode_toggle_test_iteration(settings, current_state)
- if not self.device_is_connected_to_ap(self.primary_client, self.dut):
- raise StressTestIterationFailure(
- "SoftAP is no longer functional after client mode toggle."
- )
+ current_state = STATE_DOWN
+ for i in range(test.iterations):
+ self.toggle_client_mode(self.access_point, test.ap_params, current_state)
+ self.assert_connected_to_ap(self.primary_client, self.dut)
+ current_state = not current_state
# Toggle SoftAP and Client Mode Randomly
- def run_soft_ap_and_client_mode_random_toggle_stress_test(self, settings):
- """Runner function for SoftAP and client mode random toggle tests.
+ def soft_ap_and_client_mode_random_toggle_test(
+ self, test: ClientModeAlternatingTestParams
+ ) -> None:
+ if self.access_point is None:
+ raise signals.TestSkip("No access point provided")
- Each iteration, randomly chooses if a mode will be toggled or not.
+ test.ap_params.setup_ap(self.access_point)
- Args:
- settings: dict, containing test settings
- """
- iterations = settings["iterations"]
- pass_count = 0
current_soft_ap_state = STATE_DOWN
current_client_mode_state = STATE_DOWN
- ap_channel = settings["ap_params"]["channel"]
-
- self.client_mode_toggle_pre_test(settings)
- for iteration in range(iterations):
- self.log.info(
- f"Starting iteration {str(iteration + 1)} out of {iterations}."
- )
- passes = True
-
+ for i in range(test.iterations):
# Randomly determine if softap, client mode, or both should
# be toggled.
rand_toggle_choice = random.randrange(0, 3)
if rand_toggle_choice <= 1:
- (current_soft_ap_state, err) = self.run_toggle_iteration_func(
- self.soft_ap_toggle_test_iteration, settings, current_soft_ap_state
- )
- if err:
- self.log.error(
- f"Iteration {str(iteration + 1)} failed toggling SoftAP. "
- f"Err: {err}"
- )
- passes = False
+ self.toggle_soft_ap(test.soft_ap_params, current_soft_ap_state)
+ current_soft_ap_state = not current_soft_ap_state
if rand_toggle_choice >= 1:
- (current_client_mode_state, err) = self.run_toggle_iteration_func(
- self.client_mode_toggle_test_iteration,
- settings,
- current_client_mode_state,
+ self.toggle_client_mode(
+ self.access_point, test.ap_params, current_client_mode_state
)
- if err:
- self.log.error(
- f"Iteration {str(iteration + 1)} failed toggling client mode. "
- f"Err: {err}"
- )
- passes = False
+ current_client_mode_state = not current_client_mode_state
- soft_ap_verified = self.verify_soft_ap_connectivity_from_state(
- current_soft_ap_state, self.primary_client
- )
- client_mode_verified = self.verify_client_mode_connectivity_from_state(
- current_client_mode_state, ap_channel
- )
+ if current_soft_ap_state == STATE_UP:
+ self.assert_connected_to_ap(self.primary_client, self.dut)
+ else:
+ self.assert_disconnected_to_ap(self.primary_client, self.dut)
- if not soft_ap_verified or not client_mode_verified:
- passes = False
- if passes:
- pass_count += 1
-
- if pass_count == iterations:
- asserts.explicit_pass(
- f"Stress test passed {pass_count}/{iterations} times."
- )
- else:
- asserts.fail(f"Stress test only passed {pass_count}/{iterations} times.")
+ if current_client_mode_state == STATE_UP:
+ self.assert_connected_to_ap(
+ self.dut, self.access_point, channel=test.ap_params.channel
+ )
+ else:
+ self.assert_disconnected_to_ap(
+ self.dut, self.access_point, channel=test.ap_params.channel
+ )
# Test Cases
- def test_soft_ap_2g_open_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G),
- "security_type": SECURITY_OPEN,
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_2G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_5g_open_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_OPEN,
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_5G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_any_open_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_OPEN,
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_ANY,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_2g_wep_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G),
- "security_type": SECURITY_WEP,
- "password": generate_random_password(security_mode=SECURITY_WEP),
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_2G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_5g_wep_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WEP,
- "password": generate_random_password(security_mode=SECURITY_WEP),
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_5G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_any_wep_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WEP,
- "password": generate_random_password(security_mode=SECURITY_WEP),
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_ANY,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client,
- )
-
- def test_soft_ap_2g_wpa_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G),
- "security_type": SECURITY_WPA,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_2G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_5g_wpa_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WPA,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_5G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_any_wpa_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WPA,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_ANY,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_2g_wpa2_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G),
- "security_type": SECURITY_WPA2,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_2G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_5g_wpa2_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WPA2,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_5G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_any_wpa2_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WPA2,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_ANY,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_2g_wpa3_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G),
- "security_type": SECURITY_WPA3,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_2G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_5g_wpa3_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WPA3,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_ANY,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_any_wpa3_local(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WPA3,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_LOCAL,
- "operating_band": OPERATING_BAND_ANY,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_2g_open_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G),
- "security_type": SECURITY_OPEN,
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_2G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_5g_open_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_OPEN,
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_5G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_any_open_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_OPEN,
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_ANY,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_2g_wep_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G),
- "security_type": SECURITY_WEP,
- "password": generate_random_password(security_mode=SECURITY_WEP),
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_2G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_5g_wep_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WEP,
- "password": generate_random_password(security_mode=SECURITY_WEP),
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_5G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_any_wep_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WEP,
- "password": generate_random_password(security_mode=SECURITY_WEP),
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_ANY,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_2g_wpa_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G),
- "security_type": SECURITY_WPA,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_2G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_5g_wpa_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WPA,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_5G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_any_wpa_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WPA,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_ANY,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_2g_wpa2_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G),
- "security_type": SECURITY_WPA2,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_2G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_5g_wpa2_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WPA2,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_5G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_any_wpa2_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WPA2,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_ANY,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_2g_wpa3_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_2G),
- "security_type": SECURITY_WPA3,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_2G,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_5g_wpa3_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WPA3,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_ANY,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
- def test_soft_ap_any_wpa3_unrestricted(self):
- soft_ap_params = {
- "ssid": utils.rand_ascii_str(hostapd_constants.AP_SSID_LENGTH_5G),
- "security_type": SECURITY_WPA3,
- "password": generate_random_password(),
- "connectivity_mode": CONNECTIVITY_MODE_UNRESTRICTED,
- "operating_band": OPERATING_BAND_ANY,
- }
- self.start_soft_ap(soft_ap_params)
- self.verify_soft_ap_associate_and_pass_traffic(
- self.primary_client, soft_ap_params
- )
-
def test_multi_client(self):
"""Tests multi-client association with a single soft AP network.
@@ -1564,17 +1122,16 @@
asserts.skip_if(len(self.clients) < 2, "Test requires at least 2 SoftAPClients")
test_params = self.soft_ap_test_params.get("multi_client_test_params", {})
- soft_ap_params = get_soft_ap_params_from_config_or_default(
- test_params.get("soft_ap_params", {})
- )
+ soft_ap_params = SoftAPParams.from_dict(test_params.get("soft_ap_params", {}))
self.start_soft_ap(soft_ap_params)
- associated = []
+ associated: List[Dict[str, Any]] = []
for client in self.clients:
# Associate new client
- self.verify_soft_ap_associate_and_ping(client, soft_ap_params)
+ self.associate_with_soft_ap(client, soft_ap_params)
+ self.assert_connected_to_ap(client, self.dut)
# Verify previously associated clients still behave as expected
for associated_client in associated:
@@ -1583,15 +1140,11 @@
f"Verifying previously associated client {id} still "
"functions correctly."
)
- if not self.device_is_connected_to_ap(
+ self.assert_connected_to_ap(
associated_client["device"], self.dut, check_traffic=True
- ):
- asserts.fail(
- f"Previously associated client {id} failed checks after "
- f"client {client.identifier} associated."
- )
+ )
- client_interface = self.get_device_test_interface(client)
+ client_interface = self.get_device_test_interface(client, DeviceRole.CLIENT)
client_ipv4 = self.wait_for_ipv4_address(client, client_interface)
associated.append({"device": client, "address": client_ipv4})
@@ -1629,13 +1182,9 @@
self.log.info(
f"Verifying still associated client {id} still functions correctly."
)
- if not self.device_is_connected_to_ap(
+ self.assert_connected_to_ap(
associated_client["device"], self.dut, check_traffic=True
- ):
- asserts.fail(
- f"Previously associated client {id} failed checks after client "
- f"{client.identifier} disassociated."
- )
+ )
self.log.info("All disassociations occurred smoothly.")
@@ -1649,36 +1198,30 @@
TestFailure: if DUT fails to pass traffic as either a client or an
AP
"""
- asserts.skip_if(not self.access_point, "No access point provided.")
+ if self.access_point is None:
+ raise signals.TestSkip("No access point provided")
self.log.info("Setting up AP using hostapd.")
test_params = self.soft_ap_test_params.get("soft_ap_and_client_test_params", {})
# Configure AP
- ap_params = get_ap_params_from_config_or_default(
- test_params.get("ap_params", {})
- )
+ ap_params = APParams.from_dict(test_params.get("ap_params", {}))
# Setup AP and associate DUT
- ap_profile = ap_params.pop("profile")
- setup_ap(access_point=self.access_point, profile_name=ap_profile, **ap_params)
+ ap_params.setup_ap(self.access_point)
try:
- self.start_client_mode_and_verify_connected(ap_params)
+ self.start_client_mode_and_verify_connected(self.access_point, ap_params)
except Exception as err:
asserts.fail(f"Failed to set up client mode. Err: {err}")
# Setup SoftAP
- soft_ap_params = get_soft_ap_params_from_config_or_default(
- test_params.get("soft_ap_params", {})
- )
+ soft_ap_params = SoftAPParams.from_dict(test_params.get("soft_ap_params", {}))
self.start_soft_ap_and_verify_connected(self.primary_client, soft_ap_params)
# Get FuchsiaDevice test interfaces
- dut_ap_interface = self.get_device_test_interface(
- self.dut, role=INTERFACE_ROLE_AP
- )
+ dut_ap_interface = self.get_device_test_interface(self.dut, role=DeviceRole.AP)
dut_client_interface = self.get_device_test_interface(
- self.dut, role=INTERFACE_ROLE_CLIENT
+ self.dut, role=DeviceRole.CLIENT
)
# Get FuchsiaDevice addresses
@@ -1701,7 +1244,7 @@
# Setup iperf processes:
# Primary client <-> SoftAP interface on FuchsiaDevice
# AP <-> Client interface on FuchsiaDevice
- process_errors = mp.Queue()
+ process_errors: mp.Queue = mp.Queue()
iperf_soft_ap = mp.Process(
target=self.run_iperf_traffic_parallel_process,
args=[
@@ -1751,61 +1294,45 @@
"simultaneously."
)
- def test_soft_ap_association_stress(self):
- """Sets up a single AP and repeatedly associate/disassociate
- a client, verifying connection every time
+ def generate_association_stress_tests(self):
+ """Repeatedly associate and disassociate a client.
- Each test creates 1 SoftAP and repeatedly associates/disassociates
- client.
+ Creates one SoftAP and uses one client.
- Example Config
- "soft_ap_test_params" : {
- "soft_ap_association_stress_tests": [
- {
- "ssid": "test_network",
- "security_type": "wpa2",
- "password": "password",
- "connectivity_mode": "local_only",
- "operating_band": "only_2_4_ghz",
- "iterations": 10
- }
- ]
- }
+ Example config:
+
+ soft_ap_test_params:
+ soft_ap_association_stress_tests:
+ - soft_ap_params:
+ ssid: "test_network"
+ security_type: "wpa2"
+ password: "password"
+ connectivity_mode: "local_only"
+ operating_band: "only_2_4_ghz"
+ iterations: 10
"""
- tests = self.soft_ap_test_params.get(
+ test_specs: List[Dict[str, Any]] = self.soft_ap_test_params.get(
"test_soft_ap_association_stress",
- [dict(test_name="test_soft_ap_association_stress_default")],
+ [],
)
- test_settings_list = []
- for config_settings in tests:
- soft_ap_params = get_soft_ap_params_from_config_or_default(
- config_settings.get("soft_ap_params", {})
- )
- test_type = config_settings.get("test_type", "associate_and_pass_traffic")
- iterations = config_settings.get(
- "iterations", DEFAULT_STRESS_TEST_ITERATIONS
- )
- test_settings = {
- "test_name": config_settings.get(
- "test_name",
- f"test_soft_ap_association_stress_{iterations}_iterations",
- ),
- "client": self.primary_client,
- "soft_ap_params": soft_ap_params,
- "test_type": test_type,
- "iterations": iterations,
- }
- test_settings_list.append(test_settings)
+ tests = [AssociationStressTestParams.from_dict(spec) for spec in test_specs]
- self.run_generated_testcases(
+ if len(tests) == 0:
+ # Add default test
+ tests.append(AssociationStressTestParams.from_dict({}))
+
+ def generate_name(test: AssociationStressTestParams) -> str:
+ return f"test_association_stress_{test}"
+
+ self.generate_tests(
self.run_soft_ap_association_stress_test,
- test_settings_list,
- name_func=get_test_name_from_settings,
+ generate_name,
+ tests,
)
- def test_soft_ap_and_client_mode_alternating_stress(self):
- """Runs tests that alternate between SoftAP and Client modes.
+ def generate_soft_ap_and_client_mode_alternating_stress_tests(self):
+ """Alternate between SoftAP and Client modes.
Each tests sets up an AP. Then, for each iteration:
- DUT starts up SoftAP, client associates with SoftAP,
@@ -1814,288 +1341,190 @@
disassociates
Example Config:
- "soft_ap_test_params": {
- "toggle_soft_ap_and_client_tests": [
- {
- "test_name": "test_wpa2_client_ap_toggle",
- "ap_params": {
- "channel": 6,
- "ssid": "test-ap-network",
- "security_mode": "wpa2",
- "password": "password"
- },
- "soft_ap_params": {
- "ssid": "test-soft-ap-network",
- "security_type": "wpa2",
- "password": "other-password",
- "connectivity_mode": "local_only",
- "operating_band": "only_2_4_ghz"
- },
- "iterations": 5
- }
- ]
- }
+
+ soft_ap_test_params:
+ toggle_soft_ap_and_client_tests:
+ - ap_params:
+ ssid: "test-ap-network"
+ security_mode: "wpa2"
+ password: "password"
+ channel: 6
+ soft_ap_params:
+ ssid: "test-soft-ap-network"
+ security_type: "wpa2"
+ password: "other-password"
+ connectivity_mode: "local_only"
+ operating_band: "only_2_4_ghz"
+ iterations: 5
"""
- asserts.skip_if(not self.access_point, "No access point provided.")
- tests = self.soft_ap_test_params.get(
- "test_soft_ap_and_client_mode_alternating_stress",
- [dict(test_name="test_soft_ap_and_client_mode_alternating_stress_default")],
+ test_specs: List[Dict[str, Any]] = self.soft_ap_test_params.get(
+ "toggle_soft_ap_and_client_tests",
+ [],
)
- test_settings_list = []
- for config_settings in tests:
- ap_params = get_ap_params_from_config_or_default(
- config_settings.get("ap_params", {})
- )
- soft_ap_params = get_soft_ap_params_from_config_or_default(
- config_settings.get("soft_ap_params", {})
- )
- iterations = config_settings.get(
- "iterations", DEFAULT_STRESS_TEST_ITERATIONS
- )
+ tests = [ClientModeAlternatingTestParams.from_dict(spec) for spec in test_specs]
- test_settings = {
- "test_name": config_settings.get(
- "test_name",
- f"test_soft_ap_and_client_mode_alternating_stress_{iterations}_iterations",
- ),
- "iterations": iterations,
- "soft_ap_params": soft_ap_params,
- "ap_params": ap_params,
- }
+ if len(tests) == 0:
+ # Add default test
+ tests.append(ClientModeAlternatingTestParams.from_dict({}))
- test_settings_list.append(test_settings)
- self.run_generated_testcases(
- test_func=self.run_soft_ap_and_client_mode_alternating_test,
- settings=test_settings_list,
- name_func=get_test_name_from_settings,
+ def generate_name(test: ClientModeAlternatingTestParams) -> str:
+ return f"test_soft_ap_and_client_mode_alternating_stress_{test}"
+
+ self.generate_tests(
+ self.run_soft_ap_and_client_mode_alternating_test,
+ generate_name,
+ tests,
)
- def test_soft_ap_toggle_stress(self):
- """Runs SoftAP toggling stress test.
-
- Each iteration toggles SoftAP to the opposite state (up or down).
+ def generate_soft_ap_toggle_stress_tests(self):
+ """Toggle SoftAP up and down.
If toggled up, a client is associated and connection is verified
If toggled down, test verifies client is not connected
Will run with default params, but custom tests can be provided in the
- ACTS config.
+ Mobly config.
Example Config
- "soft_ap_test_params" : {
- "test_soft_ap_toggle_stress": [
- "soft_ap_params": {
- "security_type": "wpa2",
- "password": "password",
- "connectivity_mode": "local_only",
- "operating_band": "only_2_4_ghz",
- },
- "iterations": 10
- ]
- }
+
+ soft_ap_test_params:
+ test_soft_ap_toggle_stress:
+ soft_ap_params:
+ security_type: "wpa2"
+ password: "password"
+ connectivity_mode: "local_only"
+ operating_band: "only_2_4_ghz"
+ iterations: 5
"""
- tests = self.soft_ap_test_params.get(
+ test_specs: List[Dict[str, Any]] = self.soft_ap_test_params.get(
"test_soft_ap_toggle_stress",
- [dict(test_name="test_soft_ap_toggle_stress_default")],
+ [],
)
- test_settings_list = []
- for config_settings in tests:
- soft_ap_params = get_soft_ap_params_from_config_or_default(
- config_settings.get("soft_ap_params", {})
- )
- iterations = config_settings.get(
- "iterations", DEFAULT_STRESS_TEST_ITERATIONS
- )
- test_settings = {
- "test_name": config_settings.get(
- "test_name", f"test_soft_ap_toggle_stress_{iterations}_iterations"
- ),
- "test_runner_func": self.soft_ap_toggle_test_iteration,
- "soft_ap_params": soft_ap_params,
- "iterations": iterations,
- }
- test_settings_list.append(test_settings)
+ tests = [ToggleTestParams.from_dict(spec) for spec in test_specs]
- self.run_generated_testcases(
- self.run_toggle_stress_test,
- test_settings_list,
- name_func=get_test_name_from_settings,
+ if len(tests) == 0:
+ # Add default test
+ tests.append(ToggleTestParams.from_dict({}))
+
+ def generate_name(test: ToggleTestParams) -> str:
+ return f"test_soft_ap_toggle_stress_{test}"
+
+ self.generate_tests(
+ self.soft_ap_toggle_test,
+ generate_name,
+ tests,
)
- def test_client_mode_toggle_stress(self):
- """Runs client mode toggling stress test.
-
- Each iteration toggles client mode to the opposite state (up or down).
+ def generate_client_mode_toggle_stress_tests(self):
+ """Toggles client mode up and down.
If toggled up, DUT associates to AP, and connection is verified
If toggled down, test verifies DUT is not connected to AP
Will run with default params, but custom tests can be provided in the
- ACTS config.
+ Mobly config.
Example Config
- "soft_ap_test_params" : {
- "test_client_mode_toggle_stress": [
- "soft_ap_params": {
- 'ssid': ssid,
- 'channel': channel,
- 'security_mode': security,
- 'password': password
- },
- "iterations": 10
- ]
- }
+
+ soft_ap_test_params:
+ test_client_mode_toggle_stress:
+ soft_ap_params:
+ security_type: "wpa2"
+ password: "password"
+ connectivity_mode: "local_only"
+ operating_band: "only_2_4_ghz"
+ iterations: 10
"""
- asserts.skip_if(not self.access_point, "No access point provided.")
- tests = self.soft_ap_test_params.get(
+ test_specs: List[Dict[str, Any]] = self.soft_ap_test_params.get(
"test_client_mode_toggle_stress",
- [dict(test_name="test_client_mode_toggle_stress_default")],
+ [],
)
- test_settings_list = []
- for config_settings in tests:
- ap_params = get_ap_params_from_config_or_default(
- config_settings.get("ap_params", {})
- )
- iterations = config_settings.get(
- "iterations", DEFAULT_STRESS_TEST_ITERATIONS
- )
- test_settings = {
- "test_name": config_settings.get(
- "test_name",
- f"test_client_mode_toggle_stress_{iterations}_iterations",
- ),
- "test_runner_func": self.client_mode_toggle_test_iteration,
- "pre_test_func": self.client_mode_toggle_pre_test,
- "ap_params": ap_params,
- "iterations": iterations,
- }
- test_settings_list.append(test_settings)
- self.run_generated_testcases(
- self.run_toggle_stress_test,
- test_settings_list,
- name_func=get_test_name_from_settings,
+ tests = [ClientModeToggleTestParams.from_dict(spec) for spec in test_specs]
+
+ if len(tests) == 0:
+ # Add default test
+ tests.append(ClientModeToggleTestParams.from_dict({}))
+
+ def generate_name(test: ClientModeToggleTestParams) -> str:
+ return f"test_client_mode_toggle_stress_{test}"
+
+ self.generate_tests(
+ self.client_mode_toggle_test,
+ generate_name,
+ tests,
)
- def test_soft_ap_toggle_stress_with_client_mode(self):
+ def generate_soft_ap_toggle_stress_with_client_mode_tests(self):
"""Same as test_soft_ap_toggle_stress, but client mode is set up
at test start and verified after every toggle."""
- asserts.skip_if(not self.access_point, "No access point provided.")
- tests = self.soft_ap_test_params.get(
+
+ test_specs: List[Dict[str, Any]] = self.soft_ap_test_params.get(
"test_soft_ap_toggle_stress_with_client_mode",
- [dict(test_name="test_soft_ap_toggle_stress_with_client_mode_default")],
+ [],
)
- test_settings_list = []
- for config_settings in tests:
- soft_ap_params = get_soft_ap_params_from_config_or_default(
- config_settings.get("soft_ap_params", {})
- )
- ap_params = get_ap_params_from_config_or_default(
- config_settings.get("ap_params", {})
- )
- iterations = config_settings.get(
- "iterations", DEFAULT_STRESS_TEST_ITERATIONS
- )
- test_settings = {
- "test_name": config_settings.get(
- "test_name",
- f"test_soft_ap_toggle_stress_with_client_mode_{iterations}_iterations",
- ),
- "test_runner_func": self.soft_ap_toggle_with_client_mode_iteration,
- "pre_test_func": self.soft_ap_toggle_with_client_mode_pre_test,
- "soft_ap_params": soft_ap_params,
- "ap_params": ap_params,
- "iterations": iterations,
- }
- test_settings_list.append(test_settings)
- self.run_generated_testcases(
- self.run_toggle_stress_test,
- test_settings_list,
- name_func=get_test_name_from_settings,
+ tests = [ClientModeAlternatingTestParams.from_dict(spec) for spec in test_specs]
+
+ if len(tests) == 0:
+ # Add default test
+ tests.append(ClientModeAlternatingTestParams.from_dict({}))
+
+ def generate_name(test: ClientModeAlternatingTestParams) -> str:
+ return f"test_soft_ap_toggle_stress_with_client_mode_{test}"
+
+ self.generate_tests(
+ self.soft_ap_toggle_with_client_mode_test,
+ generate_name,
+ tests,
)
- def test_client_mode_toggle_stress_with_soft_ap(self):
+ def generate_client_mode_toggle_stress_with_soft_ap_tests(self):
"""Same as test_client_mode_toggle_stress, but softap is set up at
test start and verified after every toggle."""
- asserts.skip_if(not self.access_point, "No access point provided.")
- tests = self.soft_ap_test_params.get(
+ test_specs: List[Dict[str, Any]] = self.soft_ap_test_params.get(
"test_client_mode_toggle_stress_with_soft_ap",
- [dict(test_name="test_client_mode_toggle_stress_with_soft_ap_default")],
+ [],
)
- test_settings_list = []
- for config_settings in tests:
- soft_ap_params = get_soft_ap_params_from_config_or_default(
- config_settings.get("soft_ap_params", {})
- )
- ap_params = get_ap_params_from_config_or_default(
- config_settings.get("ap_params", {})
- )
- iterations = config_settings.get(
- "iterations", DEFAULT_STRESS_TEST_ITERATIONS
- )
- test_settings = {
- "test_name": config_settings.get(
- "test_name",
- f"test_client_mode_toggle_stress_with_soft_ap_{iterations}_iterations",
- ),
- "test_runner_func": self.client_mode_toggle_with_soft_ap_iteration,
- "pre_test_func": self.client_mode_toggle_with_soft_ap_pre_test,
- "soft_ap_params": soft_ap_params,
- "ap_params": ap_params,
- "iterations": iterations,
- }
- test_settings_list.append(test_settings)
- self.run_generated_testcases(
- self.run_toggle_stress_test,
- test_settings_list,
- name_func=get_test_name_from_settings,
+ tests = [ClientModeAlternatingTestParams.from_dict(spec) for spec in test_specs]
+
+ if len(tests) == 0:
+ # Add default test
+ tests.append(ClientModeAlternatingTestParams.from_dict({}))
+
+ def generate_name(test: ClientModeAlternatingTestParams) -> str:
+ return f"test_client_mode_toggle_stress_with_soft_ap_{test}"
+
+ self.generate_tests(
+ self.soft_ap_toggle_with_client_mode_test,
+ generate_name,
+ tests,
)
- def test_soft_ap_and_client_mode_random_toggle_stress(self):
+ def generate_soft_ap_and_client_mode_random_toggle_stress_tests(self):
"""Same as above toggle stres tests, but each iteration, either softap,
client mode, or both are toggled, then states are verified."""
- asserts.skip_if(not self.access_point, "No access point provided.")
- tests = self.soft_ap_test_params.get(
+ test_specs: List[Dict[str, Any]] = self.soft_ap_test_params.get(
"test_soft_ap_and_client_mode_random_toggle_stress",
- [
- dict(
- test_name="test_soft_ap_and_client_mode_random_toggle_stress_default"
- )
- ],
+ [],
)
- test_settings_list = []
- for config_settings in tests:
- soft_ap_params = get_soft_ap_params_from_config_or_default(
- config_settings.get("soft_ap_params", {})
- )
- ap_params = get_ap_params_from_config_or_default(
- config_settings.get("ap_params", {})
- )
- iterations = config_settings.get(
- "iterations", DEFAULT_STRESS_TEST_ITERATIONS
- )
- test_settings = {
- "test_name": config_settings.get(
- "test_name",
- (
- "test_soft_ap_and_client_mode_random_toggle_stress_"
- f"{iterations}_iterations"
- ),
- ),
- "soft_ap_params": soft_ap_params,
- "ap_params": ap_params,
- "iterations": iterations,
- }
- test_settings_list.append(test_settings)
- self.run_generated_testcases(
- self.run_soft_ap_and_client_mode_random_toggle_stress_test,
- test_settings_list,
- name_func=get_test_name_from_settings,
+ tests = [ClientModeAlternatingTestParams.from_dict(spec) for spec in test_specs]
+
+ if len(tests) == 0:
+ # Add default test
+ tests.append(ClientModeAlternatingTestParams.from_dict({}))
+
+ def generate_name(test: ClientModeAlternatingTestParams) -> str:
+ return f"test_soft_ap_and_client_mode_random_toggle_stress_{test}"
+
+ self.generate_tests(
+ self.soft_ap_and_client_mode_random_toggle_test,
+ generate_name,
+ tests,
)
