Refactor wlan_controller and wlan_lib for Honeydew
To enable Antlion to use the Honeydew fuchsia device and it's SL4F
controller we have to refactor our Antlion library and controller for
WLAN to match the API for Honeydew. This also involves shifting some of
the methods around from wlan_controller to our fuchsia_device and
refactoring wlan tests to match updated naming.
Change-Id: Ieb69e4a97e521df3cd099530d463ebf91b780e5c
Reviewed-on: https://fuchsia-review.googlesource.com/c/antlion/+/976512
Fuchsia-Auto-Submit: Patrick Lu <patricklu@google.com>
Reviewed-by: Sam Balana <sbalana@google.com>
Reviewed-by: Anthonio Saw <anthonio@google.com>
Commit-Queue: Auto-Submit <auto-submit@fuchsia-infra.iam.gserviceaccount.com>
diff --git a/BUILD.gn b/BUILD.gn
index ce2c77a..b38fdbc 100644
--- a/BUILD.gn
+++ b/BUILD.gn
@@ -110,8 +110,6 @@
"controllers/fuchsia_lib/lib_controllers/__init__.py",
"controllers/fuchsia_lib/lib_controllers/wlan_controller.py",
"controllers/fuchsia_lib/lib_controllers/wlan_policy_controller.py",
- "controllers/fuchsia_lib/location/__init__.py",
- "controllers/fuchsia_lib/location/regulatory_region_lib.py",
"controllers/fuchsia_lib/logging_lib.py",
"controllers/fuchsia_lib/netstack/__init__.py",
"controllers/fuchsia_lib/netstack/netstack_lib.py",
@@ -221,9 +219,9 @@
"validation.py",
]
library_deps = [
+ "third_party/github.com/jd/tenacity",
"//src/testing/end_to_end/honeydew",
"//third_party/mobly",
"//third_party/pyyaml:yaml",
- "third_party/github.com/jd/tenacity",
]
}
diff --git a/packages/antlion/controllers/fuchsia_device.py b/packages/antlion/controllers/fuchsia_device.py
index 705745b..79a0a28 100644
--- a/packages/antlion/controllers/fuchsia_device.py
+++ b/packages/antlion/controllers/fuchsia_device.py
@@ -16,16 +16,20 @@
from __future__ import annotations
+import logging
import os
import re
import textwrap
import time
from typing import Any, Mapping
+import fuchsia_controller_py as fuchsia_controller
import honeydew
from honeydew.interfaces.device_classes.fuchsia_device import (
FuchsiaDevice as HdFuchsiaDevice,
)
+from honeydew.typing.custom_types import TRANSPORT, FFXConfig
+from honeydew.typing.wlan import CountryCode
from antlion import context
from antlion import logger as acts_logger
@@ -283,11 +287,22 @@
"if use_honeydew is True"
)
- self.honeydew_fd = honeydew.create_device(
- device_name=self.mdns_name, ssh_private_key=self.ssh_priv_key
+ hd_ffx_config = FFXConfig(
+ binary_path=self.ffx_binary_path,
+ isolate_dir=fuchsia_controller.IsolateDir(None),
+ # TODO(http://b/324454126): Remove type ignore
+ logs_dir=f"{logging.log_path}/ffx/", # type: ignore[attr-defined]
+ logs_level="None",
+ mdns_enabled=False,
+ subtools_search_path=None,
)
- self.init_controllers()
+ self.honeydew_fd = honeydew.create_device(
+ device_name=self.mdns_name,
+ transport=TRANSPORT.FUCHSIA_CONTROLLER_PREFERRED,
+ ffx_config=hd_ffx_config,
+ ssh_private_key=self.ssh_priv_key,
+ )
@cached_property
def sl4f(self) -> SL4F:
@@ -349,6 +364,10 @@
def wlan_policy_controller(self) -> WlanPolicyController:
return WlanPolicyController(self.sl4f, self.ssh)
+ @cached_property
+ def wlan_controller(self) -> WlanController:
+ return WlanController(self.sl4f)
+
def _generate_ssh_config(self, file_path: str) -> None:
"""Generate and write an SSH config for Fuchsia to disk.
@@ -379,10 +398,6 @@
with open(file_path, "w", encoding="utf-8") as file:
file.write(content)
- def init_controllers(self) -> None:
- # Contains WLAN core functions
- self.wlan_controller = WlanController(self)
-
def start_package_server(self) -> None:
if not self.packages_archive_path:
self.log.warn(
@@ -404,6 +419,24 @@
self.package_server.start()
self.package_server.configure_device(self.ssh)
+ def update_wlan_interfaces(self) -> None:
+ """Retrieves WLAN interfaces from device and sets the FuchsiaDevice
+ attributes.
+ """
+ wlan_interfaces = self.wlan_controller.get_interfaces_by_role()
+ self.wlan_client_interfaces = wlan_interfaces.client
+ self.wlan_ap_interfaces = wlan_interfaces.ap
+
+ # Set test interfaces to value from config, else the first found
+ # interface, else None
+ if self.wlan_client_test_interface_name is None:
+ self.wlan_client_test_interface_name = next(
+ iter(self.wlan_client_interfaces), None
+ )
+
+ if self.wlan_ap_test_interface_name is None:
+ self.wlan_ap_test_interface_name = next(iter(self.wlan_ap_interfaces), None)
+
def configure_wlan(
self,
association_mechanism: str | None = None,
@@ -423,7 +456,7 @@
Raises:
FuchsiaDeviceError, if configuration fails
"""
- self.wlan_controller.set_country_code(self.config_country_code)
+ self.wlan_controller.set_country_code(CountryCode(self.config_country_code))
# If args aren't provided, use the defaults, which can be set in the
# config.
@@ -465,7 +498,7 @@
self.wlan_policy_controller.configure_wlan(preserve_saved_networks)
# Retrieve WLAN client and AP interfaces
- self.wlan_controller.update_wlan_interfaces()
+ self.update_wlan_interfaces()
def deconfigure_wlan(self) -> None:
"""
@@ -750,38 +783,6 @@
else:
raise ConnectionError("Device failed to get an ipv6 address.")
- def check_connect_response(self, connect_response: dict[str, object]) -> bool:
- if connect_response.get("error") is None:
- # Checks the response from SL4F and if there is no error, check
- # the result.
- connection_result = connect_response.get("result")
- if not connection_result:
- # Ideally the error would be present but just outputting a log
- # message until available.
- self.log.debug("Connect call failed, aborting!")
- return False
- else:
- # Returns True if connection was successful.
- return True
- else:
- # the response indicates an error - log and raise failure
- self.log.debug(
- "Aborting! - Connect call failed with error: %s"
- % connect_response.get("error")
- )
- return False
-
- def check_disconnect_response(self, disconnect_response: dict[str, object]) -> bool:
- if disconnect_response.get("error") is None:
- # Returns True if disconnect was successful.
- return True
- else:
- # the response indicates an error - log and raise failure
- self.log.debug(
- f"Disconnect call failed with error: {disconnect_response.get('error')}"
- )
- return False
-
def stop_services(self) -> None:
"""Stops all host-side clients to the Fuchsia device.
@@ -790,6 +791,7 @@
"""
self.log.info("Stopping host device services.")
del self.wlan_policy_controller
+ del self.wlan_controller
del self.sl4f
del self.ssh
del self.ffx
diff --git a/packages/antlion/controllers/fuchsia_lib/lib_controllers/wlan_controller.py b/packages/antlion/controllers/fuchsia_lib/lib_controllers/wlan_controller.py
index 9104d13..8791c9f 100644
--- a/packages/antlion/controllers/fuchsia_lib/lib_controllers/wlan_controller.py
+++ b/packages/antlion/controllers/fuchsia_lib/lib_controllers/wlan_controller.py
@@ -16,33 +16,21 @@
import time
from dataclasses import dataclass
-from typing import TYPE_CHECKING
+
+from honeydew.typing.wlan import CountryCode, QueryIfaceResponse, WlanMacRole
from antlion import logger, signals, utils
-from antlion.controllers.fuchsia_lib.wlan_lib import (
- FuchsiaWlanLib,
- QueryIfaceResult,
- WlanMacRole,
-)
+from antlion.controllers.fuchsia_lib.sl4f import SL4F
from antlion.validation import MapValidator
-if TYPE_CHECKING:
- from antlion.controllers.fuchsia_device import FuchsiaDevice
-
TIME_TO_SLEEP_BETWEEN_RETRIES = 1
TIME_TO_WAIT_FOR_COUNTRY_CODE = 10
-@dataclass
-class GetCountryCodeFromPhyParams:
- wlan_lib: FuchsiaWlanLib
- id: str
-
-
@dataclass(frozen=True)
class WlanInterfaces:
- client: dict[str, QueryIfaceResult]
- ap: dict[str, QueryIfaceResult]
+ client: dict[str, QueryIfaceResponse]
+ ap: dict[str, QueryIfaceResponse]
class WlanControllerError(signals.ControllerError):
@@ -52,59 +40,24 @@
class WlanController:
"""Contains methods related to wlan core, to be used in FuchsiaDevice object"""
- def __init__(self, fuchsia_device: "FuchsiaDevice") -> None:
- self.device = fuchsia_device
+ def __init__(self, sl4f: SL4F) -> None:
+ self.sl4f = sl4f
self.log = logger.create_tagged_trace_logger(
- f"WlanController for FuchsiaDevice | {self.device.ip}"
+ f"WlanController | {self.sl4f.address}"
)
- # TODO(70501): Wrap wlan_lib functions and setup from FuchsiaDevice here
- # (similar to how WlanPolicyController does it) to prevent FuchsiaDevice
- # from growing too large.
- def _configure_wlan(self) -> None:
- pass
-
- def _deconfigure_wlan(self) -> None:
- pass
-
- def update_wlan_interfaces(self) -> None:
- """Retrieves WLAN interfaces from device and sets the FuchsiaDevice
- attributes.
- """
- wlan_interfaces = self.get_interfaces_by_role()
- self.device.wlan_client_interfaces = wlan_interfaces.client
- self.device.wlan_ap_interfaces = wlan_interfaces.ap
-
- # Set test interfaces to value from config, else the first found
- # interface, else None
- if self.device.wlan_client_test_interface_name is None:
- self.device.wlan_client_test_interface_name = next(
- iter(self.device.wlan_client_interfaces), None
- )
-
- if self.device.wlan_ap_test_interface_name is None:
- self.device.wlan_ap_test_interface_name = next(
- iter(self.device.wlan_ap_interfaces), None
- )
-
def get_interfaces_by_role(self) -> WlanInterfaces:
"""Retrieves WLAN interface information."""
# Retrieve WLAN interface IDs
- response = self.device.sl4f.wlan_lib.wlanGetIfaceIdList()
- if response.get("error"):
- raise WlanControllerError(
- f"Failed to get WLAN iface ids: {response['error']}"
- )
-
- wlan_iface_ids = response.get("result", [])
+ wlan_iface_ids = self.sl4f.wlan_lib.get_iface_id_list()
if len(wlan_iface_ids) < 1:
return WlanInterfaces(client={}, ap={})
# Use IDs to get WLAN interface info and mac addresses
- wlan_ifaces_by_mac: dict[str, QueryIfaceResult] = {}
+ wlan_ifaces_by_mac: dict[str, QueryIfaceResponse] = {}
for id in wlan_iface_ids:
- result = self.device.sl4f.wlan_lib.wlanQueryInterface(id)
+ result = self.sl4f.wlan_lib.query_iface(id)
mac = utils.mac_address_list_to_str(bytes(result.sta_addr))
wlan_ifaces_by_mac[mac] = result
@@ -116,15 +69,15 @@
# because only netstack has that information. The bug linked here is
# to reconcile some of the information between the two perspectives, at
# which point we can eliminate step.
- net_ifaces_response = self.device.sl4f.netstack_lib.netstackListInterfaces()
+ net_ifaces_response = self.sl4f.netstack_lib.netstackListInterfaces()
if net_ifaces_response.get("error"):
raise WlanControllerError(
f"Failed to get network interfaces list: {net_ifaces_response['error']}"
)
net_ifaces = net_ifaces_response["result"]
- client: dict[str, QueryIfaceResult] = {}
- ap: dict[str, QueryIfaceResult] = {}
+ client: dict[str, QueryIfaceResponse] = {}
+ ap: dict[str, QueryIfaceResponse] = {}
for iface in net_ifaces:
iface = MapValidator(iface)
try:
@@ -147,7 +100,7 @@
return WlanInterfaces(client, ap)
- def set_country_code(self, country_code: str) -> None:
+ def set_country_code(self, country_code: CountryCode) -> None:
"""Sets country code through the regulatory region service and waits
for the code to be applied to WLAN PHY.
@@ -159,47 +112,19 @@
ConnectionError - failure to query PHYs
"""
self.log.info(f"Setting DUT country code to {country_code}")
- country_code_response = self.device.sl4f.regulatory_region_lib.setRegion(
- country_code
- )
- if country_code_response.get("error"):
- raise EnvironmentError(
- "Failed to set country code (%s) on DUT. Error: %s"
- % (country_code, country_code_response["error"])
- )
+ self.sl4f.wlan_lib.set_region(country_code)
self.log.info(
f"Verifying DUT country code was correctly set to {country_code}."
)
- phy_ids_response = self.device.sl4f.wlan_lib.wlanPhyIdList()
- if phy_ids_response.get("error"):
- raise ConnectionError(
- "Failed to get phy ids from DUT while setting country code to "
- f'{country_code}. Error: {phy_ids_response["error"]}'
- )
+ phy_ids_response = self.sl4f.wlan_lib.get_phy_id_list()
end_time = time.time() + TIME_TO_WAIT_FOR_COUNTRY_CODE
while time.time() < end_time:
- for id in phy_ids_response["result"]:
- get_country_response = self.device.sl4f.wlan_lib.wlanGetCountry(id)
- if get_country_response.get("error"):
- raise ConnectionError(
- "Failed to query PHY ID (%s) for country. Error: %s"
- % (id, get_country_response["error"])
- )
-
- set_code = "".join(
- [chr(ascii_char) for ascii_char in get_country_response["result"]]
- )
- if set_code != country_code:
- self.log.debug(
- "PHY (id: %s) has incorrect country code set. "
- "Expected: %s, Got: %s" % (id, country_code, set_code)
- )
- break
- else:
- self.log.info(f"All PHYs have expected country code ({country_code})")
- break
- time.sleep(TIME_TO_SLEEP_BETWEEN_RETRIES)
+ for id in phy_ids_response:
+ resp = self.sl4f.wlan_lib.get_country(id)
+ if resp == country_code:
+ return
+ time.sleep(TIME_TO_SLEEP_BETWEEN_RETRIES)
else:
raise EnvironmentError(f"Failed to set DUT country code to {country_code}.")
diff --git a/packages/antlion/controllers/fuchsia_lib/location/__init__.py b/packages/antlion/controllers/fuchsia_lib/location/__init__.py
deleted file mode 100644
index e69de29..0000000
--- a/packages/antlion/controllers/fuchsia_lib/location/__init__.py
+++ /dev/null
diff --git a/packages/antlion/controllers/fuchsia_lib/location/regulatory_region_lib.py b/packages/antlion/controllers/fuchsia_lib/location/regulatory_region_lib.py
deleted file mode 100644
index 54d9e44..0000000
--- a/packages/antlion/controllers/fuchsia_lib/location/regulatory_region_lib.py
+++ /dev/null
@@ -1,38 +0,0 @@
-#!/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.
-
-from antlion.controllers.fuchsia_lib.base_lib import BaseLib
-
-
-class FuchsiaRegulatoryRegionLib(BaseLib):
- def __init__(self, addr: str) -> None:
- super().__init__(addr, "location_regulatory_region")
-
- # TODO(fxb/46727): Provide an analagous call to check the region
- # configured into the driver.
- def setRegion(self, region_code):
- """Set regulatory region.
-
- Args:
- region_code: 2-byte ASCII string.
-
- Returns:
- Dictionary, None if success, error if error.
- """
- test_cmd = "location_regulatory_region_facade.set_region"
- test_args = {"region": region_code}
-
- return self.send_command(test_cmd, test_args)
diff --git a/packages/antlion/controllers/fuchsia_lib/sl4f.py b/packages/antlion/controllers/fuchsia_lib/sl4f.py
index 71330c5..1c8da0c 100644
--- a/packages/antlion/controllers/fuchsia_lib/sl4f.py
+++ b/packages/antlion/controllers/fuchsia_lib/sl4f.py
@@ -25,9 +25,6 @@
from antlion.controllers.fuchsia_lib.hardware_power_statecontrol_lib import (
FuchsiaHardwarePowerStatecontrolLib,
)
-from antlion.controllers.fuchsia_lib.location.regulatory_region_lib import (
- FuchsiaRegulatoryRegionLib,
-)
from antlion.controllers.fuchsia_lib.logging_lib import FuchsiaLoggingLib
from antlion.controllers.fuchsia_lib.netstack.netstack_lib import FuchsiaNetstackLib
from antlion.controllers.fuchsia_lib.ssh import FuchsiaSSHProvider, SSHError
@@ -107,9 +104,6 @@
self.address
)
- # Grab commands from FuchsiaRegulatoryRegionLib
- self.regulatory_region_lib = FuchsiaRegulatoryRegionLib(self.address)
-
# Grab commands from FuchsiaLoggingLib
self.logging_lib = FuchsiaLoggingLib(self.address)
diff --git a/packages/antlion/controllers/fuchsia_lib/wlan_lib.py b/packages/antlion/controllers/fuchsia_lib/wlan_lib.py
index 5714c43..e37c20b 100644
--- a/packages/antlion/controllers/fuchsia_lib/wlan_lib.py
+++ b/packages/antlion/controllers/fuchsia_lib/wlan_lib.py
@@ -13,251 +13,452 @@
# 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.
-from dataclasses import dataclass
+from dataclasses import asdict
from enum import StrEnum
-from typing import Any
+from honeydew import errors
+from honeydew.interfaces.device_classes.fuchsia_device import (
+ FuchsiaDevice as HdFuchsiaDevice,
+)
+from honeydew.typing.wlan import (
+ BssDescription,
+ BssType,
+ ChannelBandwidth,
+ ClientStatusConnected,
+ ClientStatusConnecting,
+ ClientStatusIdle,
+ ClientStatusResponse,
+ CountryCode,
+ Protection,
+ QueryIfaceResponse,
+ WlanChannel,
+ WlanMacRole,
+)
from mobly import signals
from antlion.controllers.fuchsia_lib.base_lib import BaseLib
from antlion.validation import MapValidator
-COMMAND_SCAN_FOR_BSS_INFO = "wlan.scan_for_bss_info"
-COMMAND_CONNECT = "wlan.connect"
-COMMAND_DISCONNECT = "wlan.disconnect"
-COMMAND_STATUS = "wlan.status"
-COMMAND_GET_IFACE_ID_LIST = "wlan.get_iface_id_list"
-COMMAND_GET_PHY_ID_LIST = "wlan.get_phy_id_list"
-COMMAND_CREATE_IFACE = "wlan.create_iface"
-COMMAND_DESTROY_IFACE = "wlan.destroy_iface"
-COMMAND_GET_COUNTRY = "wlan_phy.get_country"
-COMMAND_GET_DEV_PATH = "wlan_phy.get_dev_path"
-COMMAND_QUERY_IFACE = "wlan.query_iface"
+STATUS_IDLE_KEY = "Idle"
+STATUS_CONNECTING_KEY = "Connecting"
+
+# We need to convert the string we receive from the wlan facade to an intEnum
+# because serde gives us a string.
+string_to_int_enum_map: dict[str, Protection] = {
+ "Unknown": Protection.UNKNOWN,
+ "Open": Protection.OPEN,
+ "Wep": Protection.WEP,
+ "Wpa1": Protection.WPA1,
+ "Wpa1Wpa2PersonalTkipOnly": Protection.WPA1_WPA2_PERSONAL_TKIP_ONLY,
+ "Wpa2PersonalTkipOnly": Protection.WPA2_PERSONAL_TKIP_ONLY,
+ "Wpa1Wpa2Personal": Protection.WPA1_WPA2_PERSONAL,
+ "Wpa2Personal": Protection.WPA2_PERSONAL,
+ "Wpa2Wpa3Personal": Protection.WPA2_WPA3_PERSONAL,
+ "Wpa3Personal": Protection.WPA3_PERSONAL,
+ "Wpa2Enterprise": Protection.WPA2_ENTERPRISE,
+ "Wpa3Enterprise": Protection.WPA3_ENTERPRISE,
+}
-class WlanError(signals.ControllerError):
- """SL4F server responded to wlan request with an error."""
+class WlanFailure(signals.TestFailure):
+ """Exception for SL4F commands executed by WLAN lib."""
-class WlanMacRole(StrEnum):
- """Role of the WLAN MAC interface.
+class Command(StrEnum):
+ """Sl4f Server Commands."""
- Loosely matches the fuchsia.wlan.common.WlanMacRole FIDL enum.
- See https://cs.opensource.google/fuchsia/fuchsia/+/main:src/testing/sl4f/src/wlan/types.rs
- """
-
- CLIENT = "Client"
- AP = "Ap"
- MESH = "Mesh"
- UNKNOWN = "Unknown"
-
-
-@dataclass(frozen=True)
-class CreateIfaceResult:
- interface_id: int
-
-
-@dataclass(frozen=True)
-class QueryIfaceResult:
- """Result of the wlan.query_iface command.
-
- Matches fuchsia.wlan.device.service.QueryIfaceResponse.
- See https://cs.opensource.google/fuchsia/fuchsia/+/main:sdk/fidl/fuchsia.wlan.device.service/service.fidl
- """
-
- role: WlanMacRole
- """The role the iface is currently operating in, e.g., client role."""
- id: int
- """The iface's global ID."""
- phy_id: int
- """Iface's PHY ID."""
- phy_assigned_id: int
- """Local ID assigned by this iface's PHY."""
- sta_addr: list[int]
- """The iface's MAC."""
+ SCAN_FOR_BSS_INFO = "wlan.scan_for_bss_info"
+ CONNECT = "wlan.connect"
+ DISCONNECT = "wlan.disconnect"
+ STATUS = "wlan.status"
+ GET_IFACE_ID_LIST = "wlan.get_iface_id_list"
+ GET_PHY_ID_LIST = "wlan.get_phy_id_list"
+ CREATE_IFACE = "wlan.create_iface"
+ DESTROY_IFACE = "wlan.destroy_iface"
+ GET_COUNTRY = "wlan_phy.get_country"
+ QUERY_IFACE = "wlan.query_iface"
+ SET_REGION = "location_regulatory_region_facade.set_region"
class FuchsiaWlanLib(BaseLib):
- def __init__(self, addr: str) -> None:
+ def __init__(self, addr: str, honeydew_fd: HdFuchsiaDevice | None = None) -> None:
super().__init__(addr, "wlan")
+ self.honeydew_fd = honeydew_fd
- def wlanScanForBSSInfo(self):
+ def _check_response_error(
+ self, cmd: Command, response_json: dict[str, object]
+ ) -> object | None:
+ """Helper method to process errors from SL4F calls.
+
+ Args:
+ cmd: SL4F command sent.
+ response_json: Response from SL4F server.
+
+ Returns:
+ Response json or None if error.
+
+ Raises:
+ WlanFailure if the response_json has something in the 'error' field.
+ """
+ resp = MapValidator(response_json)
+ error = resp.get(str, "error", None)
+ if error:
+ # We sometimes expect to catch WlanFailure so we include a log here for
+ # when we do retries.
+ self.log.debug(f"SL4F call: {cmd} failed with Error: '{error}'.")
+ raise WlanFailure(f"SL4F call: {cmd} failed with Error: '{error}'.")
+ else:
+ return response_json.get("result")
+
+ def scan_for_bss_info(self) -> dict[str, list[BssDescription]]:
"""Scans and returns BSS info
Returns:
A dict mapping each seen SSID to a list of BSS Description IE
blocks, one for each BSS observed in the network
+
+ Raises:
+ WlanFailure: Sl4f run command failed.
"""
- test_cmd = COMMAND_SCAN_FOR_BSS_INFO
+ if self.honeydew_fd:
+ try:
+ return self.honeydew_fd.wlan.scan_for_bss_info()
+ except errors.Sl4fError as e:
+ raise WlanFailure(
+ f"SL4F call {Command.SCAN_FOR_BSS_INFO} failed."
+ ) from e
+ else:
+ resp = self.send_command(Command.SCAN_FOR_BSS_INFO)
+ result = self._check_response_error(Command.SCAN_FOR_BSS_INFO, resp)
- return self.send_command(test_cmd, {})
+ if not isinstance(result, dict):
+ raise TypeError(f'Expected "result" to be dict, got {type(result)}')
- def wlanConnectToNetwork(self, target_ssid, target_bss_desc, target_pwd=None):
+ ssid_bss_desc_map: dict[str, list[BssDescription]] = {}
+ for ssid_key, bss_list in result.items():
+ if not isinstance(bss_list, list):
+ raise TypeError(
+ f'Expected "bss_list" to be list, got {type(bss_list)}'
+ )
+
+ # Create BssDescription type out of return values
+ bss_descriptions: list[BssDescription] = []
+ for bss in bss_list:
+ bss_map = MapValidator(bss)
+ bssid = bss_map.list("bssid").all(int)
+ ies = bss_map.list("ies").all(int)
+ channel_map = MapValidator(bss_map.get(dict, "channel"))
+
+ wlan_channel = WlanChannel(
+ primary=channel_map.get(int, "primary"),
+ cbw=ChannelBandwidth(channel_map.get(str, "cbw")),
+ secondary80=channel_map.get(int, "secondary80"),
+ )
+
+ bss_block = BssDescription(
+ bssid=bssid,
+ bss_type=BssType(bss_map.get(str, "bss_type")),
+ beacon_period=bss_map.get(int, "beacon_period"),
+ capability_info=bss_map.get(int, "capability_info"),
+ ies=ies,
+ channel=wlan_channel,
+ rssi_dbm=bss_map.get(int, "rssi_dbm"),
+ snr_db=bss_map.get(int, "snr_db"),
+ )
+ bss_descriptions.append(bss_block)
+
+ ssid_bss_desc_map[ssid_key] = bss_descriptions
+
+ return ssid_bss_desc_map
+
+ def connect(
+ self, target_ssid: str, target_pwd: str | None, target_bss_desc: BssDescription
+ ) -> bool:
"""Triggers a network connection
Args:
- target_ssid: the network to attempt a connection to
- target_pwd: (optional) password for the target network
+ target_ssid: The network to connect to.
+ target_pwd: The password for the network.
+ target_bss_desc: The basic service set for target network.
Returns:
boolean indicating if the connection was successful
+
+ Raises:
+ WlanFailure: Sl4f run command failed.
"""
- test_cmd = COMMAND_CONNECT
- test_args = {
+ method_params = {
"target_ssid": target_ssid,
"target_pwd": target_pwd,
- "target_bss_desc": target_bss_desc,
+ "target_bss_desc": asdict(target_bss_desc),
}
+ if self.honeydew_fd:
+ try:
+ return self.honeydew_fd.wlan.connect(
+ target_ssid, target_pwd, target_bss_desc
+ )
+ except errors.Sl4fError as e:
+ raise WlanFailure(f"SL4F call {Command.CONNECT} failed.") from e
+ else:
+ resp = self.send_command(Command.CONNECT, method_params)
+ result = self._check_response_error(Command.CONNECT, resp)
- return self.send_command(test_cmd, test_args)
+ if not isinstance(result, bool):
+ raise TypeError(f'Expected "result" to be bool, got {type(result)}')
- def wlanDisconnect(self):
- """Disconnect any current wifi connections"""
- test_cmd = COMMAND_DISCONNECT
+ return result
- return self.send_command(test_cmd, {})
+ def disconnect(self) -> None:
+ """Disconnect any current wifi connections
- def wlanCreateIface(
+ Raises:
+ WlanFailure: Sl4f run command failed.
+ """
+ if self.honeydew_fd:
+ try:
+ self.honeydew_fd.wlan.disconnect()
+ except errors.Sl4fError as e:
+ raise WlanFailure(f"SL4F call {Command.DISCONNECT} failed.") from e
+ else:
+ resp = self.send_command(Command.DISCONNECT)
+ self._check_response_error(Command.DISCONNECT, resp)
+
+ def create_iface(
self, phy_id: int, role: WlanMacRole, sta_addr: str | None = None
- ) -> CreateIfaceResult:
+ ) -> int:
"""Create a new WLAN interface.
+
Args:
- phy_id: the interface id.
- role: the role of new interface.
+ phy_id: The interface id.
+ role: The role of new interface.
sta_addr: MAC address for softAP interface only.
Returns:
- Dictionary, service id if success, error if error.
+ Iface id of newly created interface.
Raises:
- WlanError: Device responded with an error.
+ WlanFailure: Sl4f run command failed.
"""
- test_cmd = COMMAND_CREATE_IFACE
- test_args = {
+ method_params = {
"phy_id": phy_id,
"role": role,
"sta_addr": sta_addr,
}
+ if self.honeydew_fd:
+ try:
+ return self.honeydew_fd.wlan.create_iface(phy_id, role, sta_addr)
+ except errors.Sl4fError as e:
+ raise WlanFailure(f"SL4F call {Command.CREATE_IFACE} failed.") from e
+ else:
+ resp = self.send_command(Command.CREATE_IFACE, method_params)
+ result = self._check_response_error(Command.CREATE_IFACE, resp)
- resp = MapValidator(self.send_command(test_cmd, test_args))
- err = resp.get(str, "error", None)
- if err is not None:
- raise WlanError(f'Failed to create wlan iface: "{err}"')
+ if not isinstance(result, int):
+ raise TypeError(f'Expected "result" to be int, got {type(result)}')
- iface_id = resp.get(int, "result")
- return CreateIfaceResult(
- interface_id=iface_id,
- )
+ return result
- def wlanDestroyIface(self, iface_id):
+ def destroy_iface(self, iface_id: int) -> None:
"""Destroy WLAN interface by ID.
+
Args:
- iface_id: the interface id.
+ iface_id: The interface to destroy.
+
+ Raises:
+ WlanFailure: Sl4f run command failed.
+ """
+ method_params = {"identifier": iface_id}
+ if self.honeydew_fd:
+ try:
+ self.honeydew_fd.wlan.destroy_iface(iface_id)
+ except errors.Sl4fError as e:
+ raise WlanFailure(f"SL4F call {Command.DESTROY_IFACE} failed.") from e
+ else:
+ resp = self.send_command(Command.DESTROY_IFACE, method_params)
+ self._check_response_error(Command.DESTROY_IFACE, resp)
+
+ def get_iface_id_list(self) -> list[int]:
+ """Get list of wlan iface IDs on device.
Returns:
- Dictionary, service id if success, error if error.
+ A list of wlan iface IDs that are present on the device.
+
+ Raises:
+ WlanFailure: Sl4f run command failed.
"""
- test_cmd = COMMAND_DESTROY_IFACE
- test_args = {"identifier": iface_id}
+ if self.honeydew_fd:
+ try:
+ return self.honeydew_fd.wlan.get_iface_id_list()
+ except errors.Sl4fError as e:
+ raise WlanFailure(
+ f"SL4F call {Command.GET_IFACE_ID_LIST} failed."
+ ) from e
+ else:
+ resp = self.send_command(Command.GET_IFACE_ID_LIST)
+ result = self._check_response_error(Command.GET_IFACE_ID_LIST, resp)
- return self.send_command(test_cmd, test_args)
+ if not isinstance(result, list):
+ raise TypeError(f'Expected "result" to be list, got {type(result)}')
- def wlanGetIfaceIdList(self) -> dict[int, Any]:
- """Get a list if wlan interface IDs.
+ return result
+
+ def get_phy_id_list(self) -> list[int]:
+ """Get list of phy ids on device.
Returns:
- List of uint16 phy IDs.
+ A list of phy ids that is present on the device.
+
+ Raises:
+ WlanFailure: Sl4f run command failed.
"""
- test_cmd = COMMAND_GET_IFACE_ID_LIST
+ if self.honeydew_fd:
+ try:
+ return self.honeydew_fd.wlan.get_phy_id_list()
+ except errors.Sl4fError as e:
+ raise WlanFailure(f"SL4F call {Command.GET_PHY_ID_LIST} failed.") from e
+ else:
+ resp = self.send_command(Command.GET_PHY_ID_LIST)
+ result = self._check_response_error(Command.GET_PHY_ID_LIST, resp)
- return self.send_command(test_cmd, {})
+ if not isinstance(result, list):
+ raise TypeError(f'Expected "result" to be list, got {type(result)}')
- def wlanPhyIdList(self) -> dict[str, Any]:
- """Get a list if wlan phy IDs.
+ return result
- Returns:
- List of IDs if success, error if error.
- """
- test_cmd = COMMAND_GET_PHY_ID_LIST
-
- return self.send_command(test_cmd, {})
-
- def wlanStatus(self, iface_id=None):
+ def status(self) -> ClientStatusResponse:
"""Request connection status
- Args:
- iface_id: unsigned 16-bit int, the wlan interface id
- (defaults to None)
-
Returns:
- Client state summary containing WlanClientState and
- status of various networks connections
+ ClientStatusResponse state summary and
+ status of various networks connections.
+
+ Raises:
+ WlanFailure: Sl4f run command failed.
"""
- test_cmd = COMMAND_STATUS
- test_args = {}
- if iface_id:
- test_args = {"iface_id": iface_id}
+ if self.honeydew_fd:
+ try:
+ return self.honeydew_fd.wlan.status()
+ except errors.Sl4fError as e:
+ raise WlanFailure(f"SL4F call {Command.STATUS} failed.") from e
+ else:
+ resp = self.send_command(Command.STATUS)
+ result = self._check_response_error(Command.STATUS, resp)
- return self.send_command(test_cmd, test_args)
+ if not isinstance(result, dict):
+ raise TypeError(f'Expected "result" to be dict, got {type(result)}')
- def wlanGetCountry(self, phy_id):
+ result_map = MapValidator(result)
+ # Only one of these keys in result should be present.
+ if STATUS_IDLE_KEY in result:
+ return ClientStatusIdle()
+ elif STATUS_CONNECTING_KEY in result:
+ ssid = result.get("Connecting")
+ if not isinstance(ssid, list):
+ raise TypeError(
+ f'Expected "connecting" to be list, got "{type(ssid)}"'
+ )
+ return ClientStatusConnecting(ssid=ssid)
+ else:
+ connected_map = MapValidator(result_map.get(dict, "Connected"))
+ channel_map = MapValidator(connected_map.get(dict, "channel"))
+ bssid = connected_map.list("bssid").all(int)
+ ssid = connected_map.list("ssid").all(int)
+ protection = connected_map.get(str, "protection")
+
+ channel = WlanChannel(
+ primary=channel_map.get(int, "primary"),
+ cbw=ChannelBandwidth(channel_map.get(str, "cbw")),
+ secondary80=channel_map.get(int, "secondary80"),
+ )
+
+ return ClientStatusConnected(
+ bssid=bssid,
+ ssid=ssid,
+ rssi_dbm=connected_map.get(int, "rssi_dbm"),
+ snr_db=connected_map.get(int, "snr_db"),
+ channel=channel,
+ protection=Protection(string_to_int_enum_map.get(protection, 0)),
+ )
+
+ def get_country(self, phy_id: int) -> CountryCode:
"""Reads the currently configured country for `phy_id`.
Args:
phy_id: unsigned 16-bit integer.
Returns:
- Dictionary, String if success, error if error.
+ The currently configured country code from phy_id.
+
+ Raises:
+ WlanFailure: Sl4f run command failed.
"""
- test_cmd = COMMAND_GET_COUNTRY
- test_args = {"phy_id": phy_id}
+ method_params = {"phy_id": phy_id}
+ if self.honeydew_fd:
+ try:
+ return self.honeydew_fd.wlan.get_country(phy_id)
+ except errors.Sl4fError as e:
+ raise WlanFailure(f"SL4F call {Command.GET_COUNTRY} failed.") from e
+ else:
+ resp = self.send_command(Command.GET_COUNTRY, method_params)
+ result = self._check_response_error(Command.GET_COUNTRY, resp)
- return self.send_command(test_cmd, test_args)
+ if not isinstance(result, list):
+ raise TypeError(f'Expected "result" to be list, got {type(result)}')
- def wlanGetDevPath(self, phy_id):
- """Queries the device path for `phy_id`.
+ set_code = "".join([chr(ascii_char) for ascii_char in result])
- Args:
- phy_id: unsigned 16-bit integer.
+ return CountryCode(set_code)
- Returns:
- Dictionary, String if success, error if error.
- """
- test_cmd = COMMAND_GET_DEV_PATH
- test_args = {"phy_id": phy_id}
-
- return self.send_command(test_cmd, test_args)
-
- def wlanQueryInterface(self, iface_id: int) -> QueryIfaceResult:
+ def query_iface(self, iface_id: int) -> QueryIfaceResponse:
"""Retrieves interface info for given wlan iface id.
Args:
- iface_id: unsigned 16-bit int, the wlan interface id.
+ iface_id: The iface_id to query
Returns:
QueryIfaceResults from the SL4F server
Raises:
- WlanError: Device responded with an error
+ WlanFailure: Sl4f run command failed.
"""
- test_cmd = COMMAND_QUERY_IFACE
- test_args = {"iface_id": iface_id}
+ method_params = {"iface_id": iface_id}
+ if self.honeydew_fd:
+ try:
+ return self.honeydew_fd.wlan.query_iface(iface_id)
+ except errors.Sl4fError as e:
+ raise WlanFailure(f"SL4F call {Command.QUERY_IFACE} failed.") from e
+ else:
+ resp = self.send_command(Command.QUERY_IFACE, method_params)
+ result = self._check_response_error(Command.QUERY_IFACE, resp)
- resp = MapValidator(self.send_command(test_cmd, test_args))
- err = resp.get(str, "error", None)
- if err is not None:
- raise WlanError(f'Failed to query wlan iface: "{err}"')
+ if not isinstance(result, dict):
+ raise TypeError(f'Expected "network" to be dict, got {type(result)}')
- result = MapValidator(resp.get(dict, "result"))
- sta_addr = result.list("sta_addr", optional=True)
- if sta_addr is None:
- # Fallback to older field name to maintain backwards compatibility
- # with older versions of SL4F's QueryIfaceResponse.
- # See https://fxrev.dev/562146
- sta_addr = result.list("mac")
- return QueryIfaceResult(
- role=WlanMacRole(result.get(str, "role")),
- id=result.get(int, "id"),
- phy_id=result.get(int, "phy_id"),
- phy_assigned_id=result.get(int, "phy_assigned_id"),
- sta_addr=sta_addr.all(int),
- )
+ iface_results = MapValidator(result)
+ sta_addr = iface_results.list("sta_addr")
+
+ return QueryIfaceResponse(
+ role=WlanMacRole(iface_results.get(str, "role")),
+ id=iface_results.get(int, "id"),
+ phy_id=iface_results.get(int, "phy_id"),
+ phy_assigned_id=iface_results.get(int, "phy_assigned_id"),
+ sta_addr=sta_addr.all(int),
+ )
+
+ def set_region(self, region_code: CountryCode) -> None:
+ """Set regulatory region.
+
+ Args:
+ region_code: CountryCode which is a 2-byte ASCII string.
+
+ Raises:
+ WlanFailure: Sl4f run command failed.
+ """
+ method_params = {"region": region_code.value}
+ if self.honeydew_fd:
+ try:
+ self.honeydew_fd.wlan.set_region(region_code)
+ except errors.Sl4fError as e:
+ raise WlanFailure(f"SL4F call {Command.SET_REGION} failed.") from e
+ else:
+ resp = self.send_command(Command.SET_REGION, method_params)
+ self._check_response_error(Command.SET_REGION, resp)
diff --git a/packages/antlion/controllers/fuchsia_lib/wlan_policy_lib.py b/packages/antlion/controllers/fuchsia_lib/wlan_policy_lib.py
index 12ab736..ff16637 100644
--- a/packages/antlion/controllers/fuchsia_lib/wlan_policy_lib.py
+++ b/packages/antlion/controllers/fuchsia_lib/wlan_policy_lib.py
@@ -46,7 +46,7 @@
"""Exception for SL4F commands executed by WLAN Policy."""
-class Sl4fMethods(StrEnum):
+class Command(StrEnum):
"""Sl4f Server Commands."""
CONNECT = "wlan_policy.connect"
@@ -68,7 +68,7 @@
self.honeydew_fd = honeydew_fd
def _check_response_error(
- self, cmd: Sl4fMethods, response_json: dict[str, object]
+ self, cmd: Command, response_json: dict[str, object]
) -> object | None:
"""Helper method to process errors from SL4F calls.
@@ -124,8 +124,8 @@
except errors.Sl4fError as e:
raise WlanPolicyError from e
else:
- resp = self.send_command(Sl4fMethods.CONNECT, method_params)
- result = self._check_response_error(Sl4fMethods.CONNECT, resp)
+ resp = self.send_command(Command.CONNECT, method_params)
+ result = self._check_response_error(Command.CONNECT, resp)
if not isinstance(result, str):
raise TypeError(f'Expected "result" to be str, got {type(result)}')
@@ -140,12 +140,12 @@
"""
if self.honeydew_fd:
try:
- return self.honeydew_fd.wlan_policy.create_client_controller()
+ self.honeydew_fd.wlan_policy.create_client_controller()
except errors.Sl4fError as e:
raise WlanPolicyError from e
else:
- resp = self.send_command(Sl4fMethods.CREATE_CLIENT_CONTROLLER)
- self._check_response_error(Sl4fMethods.CREATE_CLIENT_CONTROLLER, resp)
+ resp = self.send_command(Command.CREATE_CLIENT_CONTROLLER)
+ self._check_response_error(Command.CREATE_CLIENT_CONTROLLER, resp)
def get_saved_networks(self) -> list[NetworkConfig]:
"""Gets networks saved on device.
@@ -163,8 +163,8 @@
except errors.Sl4fError as e:
raise WlanPolicyError from e
else:
- resp = self.send_command(Sl4fMethods.GET_SAVED_NETWORKS)
- result = self._check_response_error(Sl4fMethods.GET_SAVED_NETWORKS, resp)
+ resp = self.send_command(Command.GET_SAVED_NETWORKS)
+ result = self._check_response_error(Command.GET_SAVED_NETWORKS, resp)
if not isinstance(result, list):
raise TypeError(f'Expected "result" to be list, got {type(result)}')
@@ -216,8 +216,8 @@
except errors.Sl4fError as e:
raise WlanPolicyError from e
else:
- resp = self.send_command(Sl4fMethods.GET_UPDATE, response_timeout=timeout)
- result_raw = self._check_response_error(Sl4fMethods.GET_UPDATE, resp)
+ resp = self.send_command(Command.GET_UPDATE, response_timeout=timeout)
+ result_raw = self._check_response_error(Command.GET_UPDATE, resp)
if not isinstance(result_raw, dict):
raise TypeError(f'Expected "result" to be dict, got {type(result_raw)}')
@@ -264,8 +264,8 @@
except errors.Sl4fError as e:
raise WlanPolicyError from e
else:
- resp = self.send_command(Sl4fMethods.REMOVE_ALL_NETWORKS)
- self._check_response_error(Sl4fMethods.REMOVE_ALL_NETWORKS, resp)
+ resp = self.send_command(Command.REMOVE_ALL_NETWORKS)
+ self._check_response_error(Command.REMOVE_ALL_NETWORKS, resp)
def remove_network(
self,
@@ -294,14 +294,14 @@
}
if self.honeydew_fd:
try:
- return self.honeydew_fd.wlan_policy.remove_network(
+ self.honeydew_fd.wlan_policy.remove_network(
target_ssid, security_type, target_pwd
)
except errors.Sl4fError as e:
raise WlanPolicyError from e
else:
- resp = self.send_command(Sl4fMethods.REMOVE_NETWORK, method_params)
- self._check_response_error(Sl4fMethods.REMOVE_NETWORK, resp)
+ resp = self.send_command(Command.REMOVE_NETWORK, method_params)
+ self._check_response_error(Command.REMOVE_NETWORK, resp)
def save_network(
self,
@@ -333,14 +333,14 @@
}
if self.honeydew_fd:
try:
- return self.honeydew_fd.wlan_policy.save_network(
+ self.honeydew_fd.wlan_policy.save_network(
target_ssid, hd_security_type, target_pwd
)
except errors.Sl4fError as e:
raise WlanPolicyError from e
else:
- resp = self.send_command(Sl4fMethods.SAVE_NETWORK, method_params)
- self._check_response_error(Sl4fMethods.SAVE_NETWORK, resp)
+ resp = self.send_command(Command.SAVE_NETWORK, method_params)
+ self._check_response_error(Command.SAVE_NETWORK, resp)
def scan_for_networks(self) -> list[str]:
"""Scans for networks.
@@ -358,8 +358,8 @@
except errors.Sl4fError as e:
raise WlanPolicyError from e
else:
- resp = self.send_command(Sl4fMethods.SCAN_FOR_NETWORKS)
- result = self._check_response_error(Sl4fMethods.SCAN_FOR_NETWORKS, resp)
+ resp = self.send_command(Command.SCAN_FOR_NETWORKS)
+ result = self._check_response_error(Command.SCAN_FOR_NETWORKS, resp)
if not isinstance(result, list):
raise TypeError(f'Expected "result" to be list, got {type(result)}')
@@ -381,8 +381,8 @@
except errors.Sl4fError as e:
raise WlanPolicyError from e
else:
- resp = self.send_command(Sl4fMethods.SET_NEW_UPDATE_LISTENER)
- self._check_response_error(Sl4fMethods.SET_NEW_UPDATE_LISTENER, resp)
+ resp = self.send_command(Command.SET_NEW_UPDATE_LISTENER)
+ self._check_response_error(Command.SET_NEW_UPDATE_LISTENER, resp)
def start_client_connections(self) -> None:
"""Enables device to initiate connections to networks.
@@ -392,12 +392,12 @@
"""
if self.honeydew_fd:
try:
- return self.honeydew_fd.wlan_policy.start_client_connections()
+ self.honeydew_fd.wlan_policy.start_client_connections()
except errors.Sl4fError as e:
raise WlanPolicyError from e
else:
- resp = self.send_command(Sl4fMethods.START_CLIENT_CONNECTIONS)
- self._check_response_error(Sl4fMethods.START_CLIENT_CONNECTIONS, resp)
+ resp = self.send_command(Command.START_CLIENT_CONNECTIONS)
+ self._check_response_error(Command.START_CLIENT_CONNECTIONS, resp)
def stop_client_connections(self) -> None:
"""Disables device for initiating connections to networks.
@@ -407,9 +407,9 @@
"""
if self.honeydew_fd:
try:
- return self.honeydew_fd.wlan_policy.stop_client_connections()
+ self.honeydew_fd.wlan_policy.stop_client_connections()
except errors.Sl4fError as e:
raise WlanPolicyError from e
else:
- resp = self.send_command(Sl4fMethods.STOP_CLIENT_CONNECTIONS)
- self._check_response_error(Sl4fMethods.STOP_CLIENT_CONNECTIONS, resp)
+ resp = self.send_command(Command.STOP_CLIENT_CONNECTIONS)
+ self._check_response_error(Command.STOP_CLIENT_CONNECTIONS, resp)
diff --git a/packages/antlion/test_utils/abstract_devices/wlan_device.py b/packages/antlion/test_utils/abstract_devices/wlan_device.py
index 7be00c7..3e0d608 100644
--- a/packages/antlion/test_utils/abstract_devices/wlan_device.py
+++ b/packages/antlion/test_utils/abstract_devices/wlan_device.py
@@ -19,6 +19,11 @@
import enum
from typing import Protocol, runtime_checkable
+from honeydew.typing.wlan import (
+ ClientStatusConnected,
+ ClientStatusConnecting,
+ ClientStatusIdle,
+)
from mobly.records import TestResultRecord
from antlion.controllers import iperf_client
@@ -32,7 +37,6 @@
from antlion.controllers.pdu import PduDevice
from antlion.test_utils.wifi import wifi_test_utils as awutils
from antlion.utils import PingResult, adb_shell_ping
-from antlion.validation import MapValidator
FUCHSIA_VALID_SECURITY_TYPES = {"none", "wep", "wpa", "wpa2", "wpa3"}
@@ -96,7 +100,8 @@
ssid: If specific, check if device is connect to a specific network.
Returns:
- True if connected to requested network; otherwise, False.
+ True if connected to requested network or if ssid not specified connected to
+ any network; otherwise, False.
"""
...
@@ -136,7 +141,7 @@
"""
...
- def get_wlan_interface_id_list(self) -> list[str]:
+ def get_wlan_interface_id_list(self) -> list[int]:
"""List available WLAN interfaces.
Returns:
@@ -144,14 +149,11 @@
"""
...
- def destroy_wlan_interface(self, iface_id: str) -> bool:
+ def destroy_wlan_interface(self, iface_id: int) -> None:
"""Destroy the specified WLAN interface.
Args:
iface_id: ID of the interface to destroy.
-
- Returns:
- True if successfully destroyed wlan interface, False if not.
"""
...
@@ -276,13 +278,13 @@
def disconnect(self) -> None:
awutils.turn_location_off_and_scan_toggle_off(self.device)
- def get_wlan_interface_id_list(self) -> list[str]:
+ def get_wlan_interface_id_list(self) -> list[int]:
raise NotImplementedError("get_wlan_interface_id_list is not implemented")
def get_default_wlan_test_interface(self) -> str:
return "wlan0"
- def destroy_wlan_interface(self, iface_id: str) -> bool:
+ def destroy_wlan_interface(self, iface_id: int) -> None:
raise NotImplementedError("destroy_wlan_interface is not implemented")
def is_connected(self, ssid: str | None = None) -> bool:
@@ -372,25 +374,19 @@
) -> bool:
match self.association_mode:
case AssociationMode.DRIVER:
- bss_scan_response = self.device.sl4f.wlan_lib.wlanScanForBSSInfo()
- if bss_scan_response.get("error"):
- self.device.log.error(
- f"Scan for BSS info failed. Err: {bss_scan_response['error']}"
- )
- return False
+ ssid_bss_desc_map = self.device.sl4f.wlan_lib.scan_for_bss_info()
- bss_descs_for_ssid = bss_scan_response["result"].get(target_ssid, None)
+ bss_descs_for_ssid = ssid_bss_desc_map.get(target_ssid, None)
if not bss_descs_for_ssid or len(bss_descs_for_ssid) < 1:
self.device.log.error(
- "Scan failed to find a BSS description for target_ssid %s"
- % target_ssid
+ "Scan failed to find a BSS description for target_ssid "
+ f"{target_ssid}"
)
return False
- connection_response = self.device.sl4f.wlan_lib.wlanConnectToNetwork(
- target_ssid, bss_descs_for_ssid[0], target_pwd=target_pwd
+ return self.device.sl4f.wlan_lib.connect(
+ target_ssid, target_pwd, bss_descs_for_ssid[0]
)
- return self.device.check_connect_response(connection_response)
case AssociationMode.POLICY:
try:
self.device.wlan_policy_controller.save_and_connect(
@@ -412,8 +408,7 @@
"""
match self.association_mode:
case AssociationMode.DRIVER:
- disconnect_response = self.device.sl4f.wlan_lib.wlanDisconnect()
- self.device.check_disconnect_response(disconnect_response)
+ self.device.sl4f.wlan_lib.disconnect()
case AssociationMode.POLICY:
self.device.sl4f.wlan_policy_lib.remove_all_networks()
self.device.wlan_policy_controller.wait_for_no_connections()
@@ -455,52 +450,34 @@
)
def get_wlan_interface_id_list(self) -> list[int]:
- response = self.device.sl4f.wlan_lib.wlanGetIfaceIdList()
- if response.get("error"):
- raise ConnectionError(
- f'Failed to get wlan interface IDs: {response["error"]}'
- )
- return MapValidator(response).list("result").all(int)
+ return self.device.sl4f.wlan_lib.get_iface_id_list()
def get_default_wlan_test_interface(self) -> str:
if self.device.wlan_client_test_interface_name is None:
raise TypeError("Expected wlan_client_test_interface_name to be str")
return self.device.wlan_client_test_interface_name
- def destroy_wlan_interface(self, iface_id: str) -> bool:
- result = self.device.sl4f.wlan_lib.wlanDestroyIface(iface_id)
- if result.get("error") is None:
- return True
- else:
- self.device.log.error(
- f"Failed to destroy interface with: {result.get('error')}"
- )
- return False
+ def destroy_wlan_interface(self, iface_id: int) -> None:
+ self.device.sl4f.wlan_lib.destroy_iface(iface_id)
def is_connected(self, ssid: str | None = None) -> bool:
- response = self.device.sl4f.wlan_lib.wlanStatus()
- if response.get("error"):
- raise ConnectionError("Failed to get client network connection status")
- result = response.get("result")
- if isinstance(result, dict):
- connected_to = result.get("Connected")
- # TODO(https://fxbug.dev/42166892): Remove backwards compatibility once
- # ACTS is versioned with Fuchsia.
- if not connected_to:
- connected_to = result.get("connected_to")
- if not connected_to:
+ result = self.device.sl4f.wlan_lib.status()
+ match result:
+ case ClientStatusIdle():
+ self.device.log.info("Client status idle")
return False
-
- if ssid:
- # Replace encoding errors instead of raising an exception.
- # Since `ssid` is a string, this will not affect the test
- # for equality.
- connected_ssid = bytearray(connected_to["ssid"]).decode(
- encoding="utf-8", errors="replace"
+ case ClientStatusConnecting():
+ ssid_bytes = bytearray(result.ssid).decode(encoding="utf-8", errors="replace")
+ self.device.log.info(f"Client status connecting to ssid: {ssid_bytes}")
+ return False
+ case ClientStatusConnected():
+ ssid_bytes = bytearray(result.ssid).decode(encoding="utf-8", errors="replace")
+ self.device.log.info(f"Client connected to ssid: {ssid_bytes}")
+ return ssid == ssid_bytes
+ case _:
+ raise ValueError(
+ "Status did not return a valid status response: " f"{result}"
)
- return ssid == connected_ssid
- return True
- return False
def hard_power_cycle(self, pdus: list[PduDevice]) -> None:
self.device.reboot(reboot_type="hard", testbed_pdus=pdus)
diff --git a/tests/netstack/ToggleWlanInterfaceStressTest.py b/tests/netstack/ToggleWlanInterfaceStressTest.py
index ca26475..f48f899 100644
--- a/tests/netstack/ToggleWlanInterfaceStressTest.py
+++ b/tests/netstack/ToggleWlanInterfaceStressTest.py
@@ -66,17 +66,21 @@
for i in range(1000):
wlan_interfaces = self.dut.get_wlan_interface_id_list()
- print(wlan_interfaces)
+ self.log.info(wlan_interfaces)
+
if len(wlan_interfaces) < 1:
raise signals.TestFailure("Not enough wlan interfaces for test")
- if not self.dut.destroy_wlan_interface(wlan_interfaces[0]):
- raise signals.TestFailure("Failed to destroy WLAN interface")
+
+ self.dut.destroy_wlan_interface(wlan_interfaces[0])
# Really make sure it is dead
self.fuchsia_devices[0].ssh.run(f"wlan iface del {wlan_interfaces[0]}")
+
# Grace period
time.sleep(2)
+
self.fuchsia_devices[0].ssh.run("wlan iface new --phy 0 --role Client")
end_time = time.time() + 300
+
while time.time() < end_time:
time.sleep(1)
if self.dut.is_connected():
diff --git a/tests/wlan/compliance/RegulatoryComplianceTest.py b/tests/wlan/compliance/RegulatoryComplianceTest.py
index 3133990..6577a36 100644
--- a/tests/wlan/compliance/RegulatoryComplianceTest.py
+++ b/tests/wlan/compliance/RegulatoryComplianceTest.py
@@ -17,6 +17,7 @@
import logging
from typing import NamedTuple
+from honeydew.typing.wlan import CountryCode
from mobly import asserts, test_runner
from antlion import utils
@@ -172,7 +173,7 @@
specific channel and channel bandwidth. Run with generated test cases
in the verify_regulatory_compliance parent test.
"""
- self.fuchsia_device.wlan_controller.set_country_code(country_code)
+ self.fuchsia_device.wlan_controller.set_country_code(CountryCode(country_code))
ssid = self.setup_ap(channel, channel_bandwidth)
diff --git a/tests/wlan/facade/WlanDeprecatedConfigurationTest.py b/tests/wlan/facade/WlanDeprecatedConfigurationTest.py
index 96a3b18..d7e0d8f 100644
--- a/tests/wlan/facade/WlanDeprecatedConfigurationTest.py
+++ b/tests/wlan/facade/WlanDeprecatedConfigurationTest.py
@@ -65,9 +65,7 @@
AttributeError, if no interface has role 'Ap'
"""
for wlan_iface in self.dut.get_wlan_interface_id_list():
- result = self.fuchsia_device.sl4f.wlan_lib.wlanQueryInterface(
- int(wlan_iface)
- )
+ result = self.fuchsia_device.sl4f.wlan_lib.query_iface(wlan_iface)
if result.role is WlanMacRole.AP:
return utils.mac_address_list_to_str(bytes(result.sta_addr))
raise AttributeError(
diff --git a/tests/wlan/facade/WlanFacadeTest.py b/tests/wlan/facade/WlanFacadeTest.py
index 94cc8a4..42dca63 100644
--- a/tests/wlan/facade/WlanFacadeTest.py
+++ b/tests/wlan/facade/WlanFacadeTest.py
@@ -20,7 +20,7 @@
import array
import logging
-from mobly import asserts, signals, test_runner
+from mobly import signals, test_runner
from antlion.test_utils.wifi import base_test
@@ -36,46 +36,17 @@
)
self.fuchsia_device = self.fuchsia_devices[0]
- def test_get_phy_id_list(self):
- result = self.fuchsia_device.sl4f.wlan_lib.wlanPhyIdList()
- error = result["error"]
- asserts.assert_true(error is None, error)
+ def test_get_phy_id_list(self) -> None:
+ result = self.fuchsia_device.sl4f.wlan_lib.get_phy_id_list()
+ self.log.info(f"Got Phy IDs {result}")
- self.log.info(f"Got Phy IDs {result['result']}")
- return True
-
- def test_get_country(self):
+ def test_get_country(self) -> None:
wlan_lib = self.fuchsia_device.sl4f.wlan_lib
+ phy_id_list = wlan_lib.get_phy_id_list()
+ country_bytes = wlan_lib.get_country(phy_id_list[0])
- result = wlan_lib.wlanPhyIdList()
- error = result["error"]
- asserts.assert_true(error is None, error)
- phy_id = result["result"][0]
-
- result = wlan_lib.wlanGetCountry(phy_id)
- error = result["error"]
- asserts.assert_true(error is None, error)
-
- country_bytes = result["result"]
country_string = str(array.array("b", country_bytes), encoding="us-ascii")
self.log.info(f"Got country {country_string} ({country_bytes})")
- return True
-
- def test_get_dev_path(self):
- wlan_lib = self.fuchsia_device.sl4f.wlan_lib
-
- result = wlan_lib.wlanPhyIdList()
- error = result["error"]
- asserts.assert_true(error is None, error)
- phy_id = result["result"][0]
-
- result = wlan_lib.wlanGetDevPath(phy_id)
- error = result["error"]
- asserts.assert_true(error is None, error)
-
- dev_path = result["result"]
- self.log.info("Got device path: %s", dev_path)
- return True
if __name__ == "__main__":
diff --git a/tests/wlan/facade/WlanStatusTest.py b/tests/wlan/facade/WlanStatusTest.py
index 8d2e447..c312c1b 100644
--- a/tests/wlan/facade/WlanStatusTest.py
+++ b/tests/wlan/facade/WlanStatusTest.py
@@ -22,6 +22,7 @@
from mobly import signals, test_runner
from mobly.records import TestResultRecord
+from antlion.controllers.fuchsia_lib.wlan_lib import WlanFailure
from antlion.test_utils.wifi import base_test
@@ -55,13 +56,13 @@
"""
for fd in self.fuchsia_devices:
fd.deconfigure_wlan()
-
- status = fd.sl4f.wlan_lib.wlanStatus()
- self.log.debug(status)
- if not status["error"] or status["result"]:
- raise signals.TestFailure("DUT's WLAN client status should be empty")
-
- raise signals.TestPass("Success")
+ try:
+ _ = fd.sl4f.wlan_lib.status()
+ raise signals.TestFailure(
+ "Calling WLAN status with no WLAN interfaces should throw WlanError"
+ )
+ except WlanFailure:
+ raise signals.TestPass("Success")
def test_wlan_started_client_status(self):
"""Queries WLAN status on DUTs with WLAN ifaces.
@@ -74,15 +75,7 @@
fd.configure_wlan(
association_mechanism="policy", preserve_saved_networks=True
)
-
- status = fd.sl4f.wlan_lib.wlanStatus()
- self.log.debug(status)
- if status["error"] or not status["result"]:
- raise signals.TestFailure(
- "DUT's WLAN client status should be populated"
- )
-
- raise signals.TestPass("Success")
+ _ = fd.sl4f.wlan_lib.status()
if __name__ == "__main__":
diff --git a/tests/wlan/functional/ChannelSwitchTest.py b/tests/wlan/functional/ChannelSwitchTest.py
index b6d9f21..df89444 100644
--- a/tests/wlan/functional/ChannelSwitchTest.py
+++ b/tests/wlan/functional/ChannelSwitchTest.py
@@ -22,6 +22,7 @@
import time
from typing import Sequence
+from honeydew.typing.wlan import ClientStatusConnected
from mobly import asserts, signals, test_runner
from antlion.controllers.access_point import setup_ap
@@ -31,7 +32,7 @@
ConnectivityMode,
OperatingBand,
)
-from antlion.controllers.fuchsia_lib.wlan_lib import WlanError, WlanMacRole
+from antlion.controllers.fuchsia_lib.wlan_lib import WlanFailure, WlanMacRole
from antlion.test_utils.abstract_devices.wlan_device import (
AssociationMode,
create_wlan_device,
@@ -184,20 +185,22 @@
self.dut.is_connected(),
"Failed to stay connected after channel switch.",
)
- client_channel = self._client_channel()
- asserts.assert_equal(
- client_channel,
- channel_num,
- f"Client interface on wrong channel ({client_channel})",
- )
- if test_with_soft_ap:
- soft_ap_channel = self._soft_ap_channel()
+ status = self.fuchsia_device.sl4f.wlan_lib.status()
+ if isinstance(status, ClientStatusConnected):
+ client_channel = status.channel.primary
asserts.assert_equal(
- soft_ap_channel,
+ client_channel,
channel_num,
- f"SoftAP interface on wrong channel ({soft_ap_channel})",
+ f"Client interface on wrong channel ({client_channel})",
)
- time.sleep(1)
+ if test_with_soft_ap:
+ soft_ap_channel = self._soft_ap_channel()
+ asserts.assert_equal(
+ soft_ap_channel,
+ channel_num,
+ f"SoftAP interface on wrong channel ({soft_ap_channel})",
+ )
+ time.sleep(1)
def test_channel_switch_2g(self) -> None:
"""Channel switch through all (US only) channels in the 2 GHz band."""
@@ -379,34 +382,6 @@
f"SL4F: Failed to stop all SoftAPs. Err: {response['error']}"
)
- def _client_channel(self) -> int:
- """Determine the channel of the DUT client interface.
-
- If the interface is not connected, the method will assert a test
- failure.
-
- Returns: channel number
-
- Raises:
- EnvironmentError: if client interface channel cannot be
- determined
- """
- status = self.fuchsia_device.sl4f.wlan_lib.wlanStatus()
- if status["error"]:
- raise EnvironmentError("Could not determine client channel")
-
- result = status["result"]
- if isinstance(result, dict):
- if result.get("Connected"):
- primary = result["Connected"]["channel"]["primary"]
- if not isinstance(primary, int):
- raise TypeError(
- f'Expected "Connected.channel.primary" to be int, got {type(primary)}'
- )
- return primary
- asserts.fail("Client interface not connected")
- raise EnvironmentError("Could not determine client channel")
-
def _soft_ap_channel(self) -> int:
"""Determine the channel of the DUT SoftAP interface.
@@ -421,26 +396,15 @@
iface_ids = self.dut.get_wlan_interface_id_list()
for iface_id in iface_ids:
try:
- result = self.fuchsia_device.sl4f.wlan_lib.wlanQueryInterface(
- int(iface_id)
- )
- except WlanError as e:
+ result = self.fuchsia_device.sl4f.wlan_lib.query_iface(iface_id)
+ except WlanFailure as e:
self.log.warn(f"Query iface {iface_id} failed: {e}")
continue
if result.role is WlanMacRole.AP:
- status = self.fuchsia_device.sl4f.wlan_lib.wlanStatus(iface_id)
- if status["error"]:
- continue
- status_result = status["result"]
- if isinstance(status_result, dict):
- if status_result.get("Connected"):
- primary = status_result["Connected"]["channel"]["primary"]
- if not isinstance(primary, int):
- raise TypeError(
- f'Expected "Connected.channel.primary" to be int, got {type(primary)}'
- )
- return primary
- asserts.fail("SoftAP interface not connected")
+ status = self.fuchsia_device.sl4f.wlan_lib.status()
+ if not isinstance(status, ClientStatusConnected):
+ raise EnvironmentError("Client not connected")
+ return status.channel.primary
raise EnvironmentError("Could not determine SoftAP channel")
diff --git a/tests/wlan/functional/SoftApTest.py b/tests/wlan/functional/SoftApTest.py
index f51f031..cad06d2 100644
--- a/tests/wlan/functional/SoftApTest.py
+++ b/tests/wlan/functional/SoftApTest.py
@@ -550,7 +550,7 @@
"""
if isinstance(device, FuchsiaDevice):
- device.wlan_controller.update_wlan_interfaces()
+ device.update_wlan_interfaces()
if role is DeviceRole.CLIENT:
if device.wlan_client_test_interface_name is None:
raise TypeError(
diff --git a/tests/wlan/functional/WlanDriverRestartTest.py b/tests/wlan/functional/WlanDriverRestartTest.py
index bb84e77..d7fa34a 100644
--- a/tests/wlan/functional/WlanDriverRestartTest.py
+++ b/tests/wlan/functional/WlanDriverRestartTest.py
@@ -21,7 +21,6 @@
from mobly.config_parser import TestRunConfig
from antlion.test_utils.wifi import base_test
-from antlion.validation import MapValidator
# Time to wait until an interface is recreated after the softmac WLAN driver
# restarts.
@@ -50,17 +49,10 @@
def test_driver_restart_recreates_interface(self) -> None:
"""Verify the WLAN interface gets recreated after its driver restarts."""
- res = MapValidator(self.fd.sl4f.wlan_lib.wlanPhyIdList())
- error = res.get(str, "error", None)
- asserts.assert_is_none(error, f"Failed to get phy id list: {error}")
- phys = res.list("result").all(int)
+ # Store existing phy and interface identifiers.
+ phys = self.fd.sl4f.wlan_lib.get_phy_id_list()
asserts.assert_equal(len(phys), 1, "Expected one phy_id")
-
- # Store existing interface identifiers.
- res = MapValidator(self.fd.sl4f.wlan_lib.wlanGetIfaceIdList())
- error = res.get(str, "error", None)
- asserts.assert_is_none(error, f"Failed to get iface list: {error}")
- old_interfaces = res.list("result").all(int)
+ old_interfaces = self.fd.sl4f.wlan_lib.get_iface_id_list()
asserts.assert_not_equal(old_interfaces, [], "Iface not found.")
# Restarting should replace the old interface with a new one.
@@ -75,10 +67,7 @@
# Check for new phy and interface identifiers.
timeout = time.time() + DELAY_FOR_DRIVER_RESTART_SEC
while time.time() < timeout:
- res = MapValidator(self.fd.sl4f.wlan_lib.wlanGetIfaceIdList())
- error = res.get(str, "error", None)
- asserts.assert_is_none(error, f"Failed to get iface list: {error}")
- new_interfaces = res.list("result").all(int)
+ new_interfaces = self.fd.sl4f.wlan_lib.get_iface_id_list()
if new_interfaces == old_interfaces:
# Interface has not been deleted yet. Keep waiting.
@@ -102,10 +91,7 @@
f"New interface not created within {DELAY_FOR_DRIVER_RESTART_SEC}s"
)
- res = MapValidator(self.fd.sl4f.wlan_lib.wlanPhyIdList())
- error = res.get(str, "error", None)
- asserts.assert_is_none(error, f"Failed to get phy id list: {error}")
- phys = res.list("result").all(int)
+ phys = self.fd.sl4f.wlan_lib.get_phy_id_list()
asserts.assert_equal(len(phys), 1, "Expected one phy_id")
diff --git a/tests/wlan/functional/WlanScanTest.py b/tests/wlan/functional/WlanScanTest.py
index 85d629f..fca98b6 100644
--- a/tests/wlan/functional/WlanScanTest.py
+++ b/tests/wlan/functional/WlanScanTest.py
@@ -152,7 +152,7 @@
def teardown_test(self):
for fd in self.fuchsia_devices:
- fd.sl4f.wlan_lib.wlanDisconnect()
+ fd.sl4f.wlan_lib.disconnect()
def teardown_class(self):
if self.start_access_point:
@@ -202,7 +202,7 @@
if "password" in wlan_network_params:
target_pwd = wlan_network_params["password"]
- bss_scan_response = fd.sl4f.wlan_lib.wlanScanForBSSInfo().get("result")
+ bss_scan_response = fd.sl4f.wlan_lib.scan_for_bss_info().get("result")
connection_response = fd.sl4f.wlan_lib.wlanConnectToNetwork(
target_ssid, bss_scan_response[target_ssid][0], target_pwd=target_pwd
)
@@ -216,7 +216,7 @@
"""
start_time = datetime.now()
- scan_response = fd.sl4f.wlan_lib.wlanStartScan()
+ scan_response = fd.sl4f.wlan_lib.scan()
# first check if we received an error
if scan_response.get("error") is None:
diff --git a/tests/wlan/functional/WlanWirelessNetworkManagementTest.py b/tests/wlan/functional/WlanWirelessNetworkManagementTest.py
index 505d48e..39c702a 100644
--- a/tests/wlan/functional/WlanWirelessNetworkManagementTest.py
+++ b/tests/wlan/functional/WlanWirelessNetworkManagementTest.py
@@ -161,7 +161,7 @@
WlanError if the DUT interface query fails.
"""
for wlan_iface in self.dut.get_wlan_interface_id_list():
- result = self.fuchsia_device.sl4f.wlan_lib.wlanQueryInterface(wlan_iface)
+ result = self.fuchsia_device.sl4f.wlan_lib.query_iface(wlan_iface)
if result.role is WlanMacRole.CLIENT:
return utils.mac_address_list_to_str(bytes(result.sta_addr))
raise ValueError(
diff --git a/tests/wlan/misc/WlanInterfaceTest.py b/tests/wlan/misc/WlanInterfaceTest.py
index 09bd39a..864d577 100644
--- a/tests/wlan/misc/WlanInterfaceTest.py
+++ b/tests/wlan/misc/WlanInterfaceTest.py
@@ -14,7 +14,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from mobly import signals, test_runner
+from mobly import test_runner
from antlion.test_utils.abstract_devices.wlan_device import (
AssociationMode,
@@ -60,11 +60,7 @@
Priority: 1
"""
wlan_interfaces = self.dut.get_wlan_interface_id_list()
- if len(wlan_interfaces) < 1:
- raise signals.TestFailure("Not enough wlan interfaces for test")
- if not self.dut.destroy_wlan_interface(wlan_interfaces[0]):
- raise signals.TestFailure("Failed to destroy WLAN interface")
- raise signals.TestPass("Success")
+ self.dut.destroy_wlan_interface(wlan_interfaces[0])
if __name__ == "__main__":
diff --git a/tests/wlan/performance/ChannelSweepTest.py b/tests/wlan/performance/ChannelSweepTest.py
index 23b402a..0d9af36 100644
--- a/tests/wlan/performance/ChannelSweepTest.py
+++ b/tests/wlan/performance/ChannelSweepTest.py
@@ -21,6 +21,7 @@
from pathlib import Path
from statistics import pstdev
+from honeydew.typing.wlan import CountryCode
from mobly import asserts, test_runner
from mobly.config_parser import TestRunConfig
@@ -574,7 +575,9 @@
TestFailure, if throughput (either direction) is less than
the directions given minimum throughput threshold.
"""
- self.fuchsia_device.wlan_controller.set_country_code(test.country_code)
+ self.fuchsia_device.wlan_controller.set_country_code(
+ CountryCode(test.country_code)
+ )
if test.security_mode is not SecurityMode.OPEN:
if test.security_mode is SecurityMode.WEP:
diff --git a/tests/wlan_policy/RegulatoryRecoveryTest.py b/tests/wlan_policy/RegulatoryRecoveryTest.py
index 8a39196..260e5c3 100644
--- a/tests/wlan_policy/RegulatoryRecoveryTest.py
+++ b/tests/wlan_policy/RegulatoryRecoveryTest.py
@@ -14,7 +14,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from honeydew.typing.wlan import WlanClientState
+from honeydew.typing.wlan import CountryCode, WlanClientState
from mobly import signals, test_runner
from antlion.controllers.ap_lib.hostapd_security import FuchsiaSecurityType
@@ -68,7 +68,7 @@
def setup_test(self) -> None:
"""Set PHYs to world-wide mode and disable AP and client connections."""
for fd in self.fuchsia_devices:
- fd.wlan_controller.set_country_code("WW")
+ fd.wlan_controller.set_country_code(CountryCode.WORLDWIDE)
fd.sl4f.wlan_ap_policy_lib.wlanStopAllAccessPoint()
def _set_country_code_check(self, fd: FuchsiaDevice) -> None:
