blob: 6d8b56204cb359c75100244aa7e57465e2751931 [file] [log] [blame]
#!/usr/bin/env python3
#
# Copyright 2022 The Fuchsia Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
import multiprocessing as mp
import random
import time
from typing import Any, Dict, Optional
from mobly import asserts, signals, test_runner
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.ap_lib.hostapd_utils import generate_random_password
from antlion.controllers.utils_lib.ssh import settings
from antlion.test_utils.abstract_devices.wlan_device import create_wlan_device
from antlion.test_utils.wifi import base_test
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,
}
def get_test_name_from_settings(settings):
return settings["test_name"]
def get_ap_params_from_config_or_default(config):
"""Retrieves AP parameters from antlion config, or returns default settings.
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:
password = generate_random_password(security_mode=security_mode)
security = hostapd_security.Security(security_mode, password)
else:
security = None
return {
"profile": profile,
"ssid": ssid,
"channel": channel,
"security": security,
"password": password,
}
def get_soft_ap_params_from_config_or_default(config):
"""Retrieves SoftAp parameters from antlion config or returns default settings.
Args:
config: dict, from antlion config, that may contain custom soft ap
parameters
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", "")
# 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,
}
class StressTestIterationFailure(Exception):
"""Used to differentiate a subtest failure from an actual exception"""
class SoftApTest(base_test.WifiBaseTest):
"""Tests for Fuchsia SoftAP
Testbed requirement:
* One Fuchsia device
* At least one client (Android) device
* For multi-client tests, at least two client (Android) devices are
required. Test will be skipped if less than two client devices are
present.
* For any tests that exercise client-mode (e.g. toggle tests, simultaneous
tests), a physical AP (whirlwind) is also required. Those tests will be
skipped if physical AP is not present.
"""
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]
self.dut = create_wlan_device(self.fuchsia_device)
# 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] = {}
for device in self.android_devices:
client_wlan_device = create_wlan_device(device)
self.clients.append(client_wlan_device)
self.iperf_clients_map[
client_wlan_device
] = client_wlan_device.create_iperf_client()
self.primary_client = self.clients[0]
# Create an iperf server on the DUT, which will be used for any streaming.
self.iperf_server_settings = settings.from_config(
{
"user": self.fuchsia_device.ssh_username,
"host": self.fuchsia_device.ip,
"ssh_config": self.fuchsia_device.ssh_config,
}
)
self.iperf_server = iperf_server.IPerfServerOverSsh(
self.iperf_server_settings, DEFAULT_IPERF_PORT, use_killall=True
)
self.iperf_server.start()
# Attempt to create an ap iperf server. AP is only required for tests
# that use client mode.
self.access_point: Optional[AccessPoint] = None
self.ap_iperf_client: Optional[iperf_client.IPerfClientOverSsh] = None
try:
self.access_point = self.access_points[0]
self.ap_iperf_client = iperf_client.IPerfClientOverSsh(
self.access_point.ssh_provider,
)
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()
super().teardown_class()
def setup_test(self):
super().setup_test()
for ad in self.android_devices:
ad.droid.wakeLockAcquireBright()
ad.droid.wakeUpNow()
for client in self.clients:
client.disconnect()
client.reset_wifi()
client.wifi_toggle_state(True)
self.stop_all_soft_aps()
if self.access_point:
self.access_point.stop_all_aps()
self.dut.disconnect()
def teardown_test(self):
for client in self.clients:
client.disconnect()
for ad in self.android_devices:
ad.droid.wakeLockRelease()
ad.droid.goToSleepNow()
self.stop_all_soft_aps()
if self.access_point:
self.download_ap_logs()
self.access_point.stop_all_aps()
self.dut.disconnect()
super().teardown_test()
def start_soft_ap(self, settings):
"""Starts a softAP on Fuchsia device.
Args:
settings: a dict containing softAP configuration params
ssid: string, SSID of softAP network
security_type: string, security type of softAP network
- 'none', 'wep', 'wpa', 'wpa2', 'wpa3'
password: string, password if applicable
connectivity_mode: string, connecitivity_mode for softAP
- 'local_only', 'unrestricted'
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}")
response = self.fuchsia_device.sl4f.wlan_ap_policy_lib.wlanStartAccessPoint(
ssid, security_type, password, connectivity_mode, operating_band
)
if response.get("error"):
raise EnvironmentError(
f"SL4F: Failed to setup SoftAP. Err: {response['error']}"
)
self.log.info(f"SoftAp network ({ssid}) is up.")
def stop_soft_ap(self, settings):
"""Stops a specific SoftAP On Fuchsia device.
Args:
settings: a dict containing softAP config params (see start_soft_ap)
for details
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
)
if response.get("error"):
raise EnvironmentError(
f"SL4F: Failed to stop SoftAP. Err: {response['error']}"
)
def stop_all_soft_aps(self):
"""Stops all SoftAPs on Fuchsia Device.
Raises:
EnvironmentError, if StopAllAps call fails.
"""
response = self.fuchsia_device.sl4f.wlan_ap_policy_lib.wlanStopAllAccessPoint()
if response.get("error"):
raise EnvironmentError(
f"SL4F: Failed to stop all SoftAPs. Err: {response['error']}"
)
def associate_with_soft_ap(self, device, soft_ap_settings):
"""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
Raises:
TestFailure, if association fails
"""
self.log.info(
f"Attempting to associate client {device.identifier} with SoftAP on "
f"FuchsiaDevice ({self.dut.identifier})."
)
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,
)
if not associated:
self.log.error("Failed to connect to SoftAp.")
return False
self.log.info("Client successfully associated with SoftAP.")
return True
def disconnect_from_soft_ap(self, device):
"""Disconnects client device from SoftAP.
Args:
device: wlan_device to disconnect from SoftAP
"""
self.log.info(f"Disconnecting device {device.identifier} from SoftAP.")
device.disconnect()
def get_device_test_interface(self, device, role=None, channel=None):
"""Retrieves test interface from a provided device, which can be the
FuchsiaDevice DUT, the AccessPoint, or an AndroidClient.
Args:
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
else:
return device.get_default_wlan_test_interface()
def wait_for_ipv4_address(self, device, interface_name, timeout=DEFAULT_TIMEOUT):
"""Waits for interface on a wlan_device to get an ipv4 address.
Args:
device: wlan_device or AccessPoint to check interface
interface_name: name of the interface to check
timeout: seconds to wait before raising an error
Raises:
ValueError, if interface does not have an ipv4 address after timeout
"""
if isinstance(device, AccessPoint):
comm_channel = device.ssh
else:
comm_channel = device.device
end_time = time.time() + timeout
while time.time() < end_time:
ips = utils.get_interface_ip_addresses(comm_channel, interface_name)
if len(ips["ipv4_private"]) > 0:
self.log.info(
f"Device {device.identifier} interface {interface_name} has "
f"ipv4 address {ips['ipv4_private'][0]}"
)
return ips["ipv4_private"][0]
else:
time.sleep(1)
raise ConnectionError(
f"After {timeout} seconds, device {device.identifier} still does not have "
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):
"""Runs traffic between client and ap an verifies throughput.
Args:
ip_client: iperf client to use
server_address: ipv4 address of the iperf server to use
server_port: port of the iperf server
Raises:
TestFailure, if no traffic passes in either direction
"""
ip_client_identifier = self.get_iperf_client_identifier(ip_client)
self.log.info(
f"Running traffic from iperf client {ip_client_identifier} to "
f"iperf server {server_address}."
)
client_to_ap_path = ip_client.start(
server_address, f"-i 1 -t 10 -J -p {server_port}", "client_to_soft_ap"
)
client_to_ap_result = iperf_server.IPerfResult(client_to_ap_path)
if not client_to_ap_result.avg_receive_rate:
raise ConnectionError(
f"Failed to pass traffic from iperf client {ip_client_identifier} to "
f"iperf server {server_address}."
)
self.log.info(
f"Passed traffic from iperf client {ip_client_identifier} to "
f"iperf server {server_address} with avg rate of "
f"{client_to_ap_result.avg_receive_rate} MB/s."
)
self.log.info(
f"Running traffic from iperf server {server_address} to "
f"iperf client {ip_client_identifier}."
)
ap_to_client_path = ip_client.start(
server_address, f"-i 1 -t 10 -R -J -p {server_port}", "soft_ap_to_client"
)
ap_to_client_result = iperf_server.IPerfResult(ap_to_client_path)
if not ap_to_client_result.avg_receive_rate:
raise ConnectionError(
f"Failed to pass traffic from iperf server {server_address} to "
f"iperf client {ip_client_identifier}."
)
self.log.info(
f"Passed traffic from iperf server {server_address} to "
f"iperf client {ip_client_identifier} with avg rate of "
f"{ap_to_client_result.avg_receive_rate} MB/s."
)
def run_iperf_traffic_parallel_process(
self, ip_client, server_address, error_queue, server_port=5201
):
"""Executes run_iperf_traffic using a queue to capture errors. Used
when running iperf in a parallel process.
Args:
ip_client: iperf client to use
server_address: ipv4 address of the iperf server to use
error_queue: multiprocessing queue to capture errors
server_port: port of the iperf server
"""
try:
self.run_iperf_traffic(ip_client, server_address, server_port=server_port)
except ConnectionError as err:
error_queue.put(
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.
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
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.
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
"""
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
)
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
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."
)
# 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"]
self.log.info(
f"Running association stress test type {test_type} in "
f"iteration {iterations} times"
)
self.start_soft_ap(soft_ap_params)
passed_count = 0
for run in range(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)
elif test_type == TEST_TYPE_ASSOCIATE_AND_PING:
self.verify_soft_ap_associate_and_ping(client, soft_ap_params)
elif test_type == TEST_TYPE_ASSOCIATE_AND_PASS_TRAFFIC:
self.verify_soft_ap_associate_and_pass_traffic(
client, soft_ap_params
)
else:
raise AttributeError(f"Invalid test type: {test_type}")
except signals.TestFailure as err:
self.log.error(
f"SoftAp association stress run {str(run + 1)} failed. "
f"Err: {err.details}"
)
else:
self.log.info(
f"SoftAp association stress run {str(run + 1)} successful."
)
passed_count += 1
if passed_count < iterations:
asserts.fail(
"SoftAp association stress test failed after "
f"{passed_count}/{iterations} runs."
)
asserts.explicit_pass(
f"SoftAp association stress test passed after {passed_count}/{iterations} "
"runs."
)
# Alternate SoftAP and Client mode test
def run_soft_ap_and_client_mode_alternating_test(self, settings):
"""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
self.client_mode_toggle_pre_test(settings)
for iteration in range(iterations):
passes = True
# 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
# 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 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):
"""Sets up SoftAP, associates a client, then verifies connection.
Args:
client: SoftApClient, client to use to verify SoftAP
soft_ap_params: dict, containing parameters to setup softap
Raises:
StressTestIterationFailure, if toggle occurs, but connection
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
)
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."
)
def stop_soft_ap_and_verify_disconnected(self, client, soft_ap_params):
"""Tears down SoftAP, and verifies connection is down.
Args:
client: SoftApClient, client to use to verify SoftAP
soft_ap_params: dict, containing parameters of SoftAP to teardown
Raise:
EnvironmentError, if client and AP can still communicate
"""
self.log.info("Stopping SoftAP on DUT.")
self.stop_soft_ap(soft_ap_params)
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):
"""Connects DUT to AP in client mode and verifies connection
Args:
ap_params: dict, containing parameters of the AP network
Raises:
EnvironmentError, if DUT fails to associate altogether
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}")
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
),
)
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.")
def stop_client_mode_and_verify_disconnected(self, ap_params):
"""Disconnects DUT from AP and verifies connection is down.
Args:
ap_params: dict, 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.")
# Toggle Stress Test Iteration and Pre-Test Functions
# SoftAP Toggle Stress Test Helper Functions
def soft_ap_toggle_test_iteration(self, settings, current_state):
"""Runs a single iteration of SoftAP toggle stress 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
"""
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
)
# Client Mode Toggle Stress Test Helper Functions
def client_mode_toggle_pre_test(self, settings):
"""Prepares the AP before client mode toggle tests
Args:
settings: dict, stress test settings
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)
def client_mode_toggle_test_iteration(self, settings, current_state):
"""Runs a single iteration of client mode toggle stress 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
"""
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)
else:
self.stop_client_mode_and_verify_disconnected(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
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}"
)
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."
)
# 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
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}"
)
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."
)
# 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.
Each iteration, randomly chooses if a mode will be toggled or not.
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
# 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
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,
)
if err:
self.log.error(
f"Iteration {str(iteration + 1)} failed toggling client mode. "
f"Err: {err}"
)
passes = False
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 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.")
# 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.
This tests associates a variable length list of clients, verfying it can
can ping the SoftAP and pass traffic, and then verfies all previously
associated clients can still ping and pass traffic.
The same occurs in reverse for disassocations.
SoftAP parameters can be changed from default via ACTS config:
Example Config
"soft_ap_test_params" : {
"multi_client_test_params": {
"ssid": "testssid",
"security_type": "wpa2",
"password": "password",
"connectivity_mode": "local_only",
"operating_band": "only_2_4_ghz"
}
}
"""
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", {})
)
self.start_soft_ap(soft_ap_params)
associated = []
for client in self.clients:
# Associate new client
self.verify_soft_ap_associate_and_ping(client, soft_ap_params)
# Verify previously associated clients still behave as expected
for associated_client in associated:
id = associated_client["device"].identifier
self.log.info(
f"Verifying previously associated client {id} still "
"functions correctly."
)
if not self.device_is_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_ipv4 = self.wait_for_ipv4_address(client, client_interface)
associated.append({"device": client, "address": client_ipv4})
self.log.info("All devices successfully associated.")
self.log.info("Verifying all associated clients can ping eachother.")
for transmitter in associated:
for receiver in associated:
if transmitter != receiver:
if not transmitter["device"].can_ping(receiver["address"]):
asserts.fail(
"Could not ping from one associated client "
f"({transmitter['address']}) to another "
f"({receiver['address']})."
)
else:
self.log.info(
"Successfully pinged from associated client "
f"({transmitter['address']}) to another "
f"({receiver['address']})"
)
self.log.info(
"All associated clients can ping each other. Beginning disassociations."
)
while len(associated) > 0:
# Disassociate client
client = associated.pop()["device"]
self.disconnect_from_soft_ap(client)
# Verify still connected clients still behave as expected
for associated_client in associated:
id = associated_client["device"].identifier
self.log.info(
f"Verifying still associated client {id} still functions correctly."
)
if not self.device_is_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.")
def test_simultaneous_soft_ap_and_client(self):
"""Tests FuchsiaDevice DUT can act as a client and a SoftAP
simultaneously.
Raises:
ConnectionError: if DUT fails to connect as client
RuntimeError: if parallel processes fail to join
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.")
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", {})
)
# Setup AP and associate DUT
ap_profile = ap_params.pop("profile")
setup_ap(access_point=self.access_point, profile_name=ap_profile, **ap_params)
try:
self.start_client_mode_and_verify_connected(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", {})
)
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_client_interface = self.get_device_test_interface(
self.dut, role=INTERFACE_ROLE_CLIENT
)
# Get FuchsiaDevice addresses
dut_ap_ipv4 = self.wait_for_ipv4_address(self.dut, dut_ap_interface)
dut_client_ipv4 = self.wait_for_ipv4_address(self.dut, dut_client_interface)
# Set up secondary iperf server of FuchsiaDevice
self.log.info("Setting up second iperf server on FuchsiaDevice DUT.")
secondary_iperf_server = iperf_server.IPerfServerOverSsh(
self.iperf_server_settings, DEFAULT_IPERF_PORT + 1, use_killall=True
)
secondary_iperf_server.start()
# Set up iperf client on AP
self.log.info("Setting up iperf client on AP.")
ap_iperf_client = iperf_client.IPerfClientOverSsh(
self.access_point.ssh_provider,
)
# Setup iperf processes:
# Primary client <-> SoftAP interface on FuchsiaDevice
# AP <-> Client interface on FuchsiaDevice
process_errors = mp.Queue()
iperf_soft_ap = mp.Process(
target=self.run_iperf_traffic_parallel_process,
args=[
self.iperf_clients_map[self.primary_client],
dut_ap_ipv4,
process_errors,
],
)
iperf_fuchsia_client = mp.Process(
target=self.run_iperf_traffic_parallel_process,
args=[ap_iperf_client, dut_client_ipv4, process_errors],
kwargs={"server_port": 5202},
)
# Run iperf processes simultaneously
self.log.info(
"Running simultaneous iperf traffic: between AP and DUT "
"client interface, and DUT AP interface and client."
)
iperf_soft_ap.start()
iperf_fuchsia_client.start()
# Block until processes can join or timeout
for proc in [iperf_soft_ap, iperf_fuchsia_client]:
proc.join(timeout=DEFAULT_IPERF_TIMEOUT)
if proc.is_alive():
proc.terminate()
proc.join()
raise RuntimeError(f"Failed to join process {proc}")
# Stop iperf server (also stopped in teardown class as failsafe)
secondary_iperf_server.stop()
# Check errors from parallel processes
if process_errors.empty():
asserts.explicit_pass(
"FuchsiaDevice was successfully able to pass traffic as a "
"client and an AP simultaneously."
)
else:
while not process_errors.empty():
self.log.error(f"Error in iperf process: {process_errors.get()}")
asserts.fail(
"FuchsiaDevice failed to pass traffic as a client and an AP "
"simultaneously."
)
def test_soft_ap_association_stress(self):
"""Sets up a single AP and repeatedly associate/disassociate
a client, verifying connection every time
Each test creates 1 SoftAP and repeatedly associates/disassociates
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
}
]
}
"""
tests = 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)
self.run_generated_testcases(
self.run_soft_ap_association_stress_test,
test_settings_list,
name_func=get_test_name_from_settings,
)
def test_soft_ap_and_client_mode_alternating_stress(self):
"""Runs tests that alternate between SoftAP and Client modes.
Each tests sets up an AP. Then, for each iteration:
- DUT starts up SoftAP, client associates with SoftAP,
connection is verified, then disassociates
- DUT associates to the AP, connection is verified, then
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
}
]
}
"""
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_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
)
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,
}
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 test_soft_ap_toggle_stress(self):
"""Runs SoftAP toggling stress test.
Each iteration toggles SoftAP to the opposite state (up or 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.
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
]
}
"""
tests = 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)
self.run_generated_testcases(
self.run_toggle_stress_test,
test_settings_list,
name_func=get_test_name_from_settings,
)
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).
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.
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
]
}
"""
asserts.skip_if(not self.access_point, "No access point provided.")
tests = 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,
)
def test_soft_ap_toggle_stress_with_client_mode(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_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,
)
def test_client_mode_toggle_stress_with_soft_ap(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_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,
)
def test_soft_ap_and_client_mode_random_toggle_stress(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_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,
)
if __name__ == "__main__":
test_runner.main()