Add typing to incomplete defs Add missing typing information to several functions. Fix the resulting typing errors caught that would have otherwise caused runtime errors. Change-Id: I54f839bdbb5cd2089f1bfcd67b4659e2d3f2e01b Reviewed-on: https://fuchsia-review.googlesource.com/c/antlion/+/920892 Fuchsia-Auto-Submit: Sam Balana <sbalana@google.com> Commit-Queue: Auto-Submit <auto-submit@fuchsia-infra.iam.gserviceaccount.com> Reviewed-by: Jonathan Chang <jnchang@google.com>
diff --git a/packages/antlion/controllers/fuchsia_lib/wlan_ap_policy_lib.py b/packages/antlion/controllers/fuchsia_lib/wlan_ap_policy_lib.py index 0caf345..f801046 100644 --- a/packages/antlion/controllers/fuchsia_lib/wlan_ap_policy_lib.py +++ b/packages/antlion/controllers/fuchsia_lib/wlan_ap_policy_lib.py
@@ -99,14 +99,17 @@ return self.send_command(test_cmd, test_args) def wlanStopAccessPoint( - self, target_ssid: str, security_type: FuchsiaSecurityType, target_pwd: str = "" + self, + target_ssid: str, + security_type: FuchsiaSecurityType, + target_pwd: str | None = None, ): """Stops an active Access Point. Args: target_ssid: the network to attempt a connection to security_type: the security protocol of the network - target_pwd: (optional) credential being saved with the network. No password - is equivalent to empty string. + target_pwd: credential being saved with the network. No password + is equivalent to empty string. Returns: boolean indicating if the action was successful @@ -117,7 +120,7 @@ test_args = { "target_ssid": target_ssid, "security_type": str(security_type), - "target_pwd": target_pwd, + "target_pwd": "" if target_pwd is None else target_pwd, } return self.send_command(test_cmd, test_args)
diff --git a/packages/antlion/controllers/iperf_server.py b/packages/antlion/controllers/iperf_server.py index 5aadec2..c691192 100755 --- a/packages/antlion/controllers/iperf_server.py +++ b/packages/antlion/controllers/iperf_server.py
@@ -262,7 +262,7 @@ __log_file_lock = threading.Lock() - def __init__(self, port): + def __init__(self, port: int): self._port = port # TODO(markdr): We shouldn't be storing the log files in an array like # this. Nobody should be reading this property either. Instead, the @@ -271,25 +271,24 @@ self.log_files = [] @property - def port(self): + def port(self) -> int: raise NotImplementedError("port must be specified.") @property - def started(self): + def started(self) -> bool: raise NotImplementedError("started must be specified.") - def start(self, extra_args="", tag=""): + def start(self, extra_args: int = "", tag: str = "") -> None: """Starts an iperf3 server. Args: - extra_args: A string representing extra arguments to start iperf - server with. + extra_args: Extra arguments to start iperf server with. tag: Appended to log file name to identify logs from different iperf runs. """ raise NotImplementedError("start() must be specified.") - def stop(self): + def stop(self) -> str: """Stops the iperf server. Returns: @@ -297,7 +296,7 @@ """ raise NotImplementedError("stop() must be specified.") - def _get_full_file_path(self, tag=None): + def _get_full_file_path(self, tag: str | None = None) -> str: """Returns the full file path for the IPerfServer log file. Note: If the directory for the file path does not exist, it will be @@ -320,7 +319,7 @@ return file_path @property - def log_path(self): + def log_path(self) -> str: current_context = context.get_current_context() full_out_dir = os.path.join( current_context.get_full_output_path(), f"IPerfServer{self.port}" @@ -427,7 +426,13 @@ class IPerfServerOverSsh(IPerfServerBase): """Class that handles iperf3 operations on remote machines.""" - def __init__(self, ssh_settings, port, test_interface=None, use_killall=False): + def __init__( + self, + ssh_settings: settings.SshSettings, + port: int, + test_interface: str | None = None, + use_killall: bool = False, + ): super().__init__(port) self.ssh_settings = ssh_settings self.log = acts_logger.create_tagged_trace_logger( @@ -442,29 +447,28 @@ self._use_killall = str(use_killall).lower() == "true" try: # A test interface can only be found if an ip address is specified. - # A fully qualified hostname will return None for the - # test_interface. - self.test_interface = ( - test_interface - if test_interface - else utils.get_interface_based_on_ip(self._ssh_session, self.hostname) - ) + # A fully qualified hostname will return None for the test_interface. + self.test_interface = test_interface + if self.test_interface is None: + self.test_interface = utils.get_interface_based_on_ip( + self._ssh_session, self.hostname + ) except Exception as e: self.log.warning(e) self.test_interface = None @property - def port(self): + def port(self) -> int: return self._port @property - def started(self): + def started(self) -> bool: return self._iperf_pid is not None - def _get_remote_log_path(self): + def _get_remote_log_path(self) -> str: return f"/tmp/iperf_server_port{self.port}.log" - def get_interface_ip_addresses(self, interface): + def get_interface_ip_addresses(self, interface: str) -> dict[str, list[str]]: """Gets all of the ip addresses, ipv4 and ipv6, associated with a particular interface name. @@ -480,7 +484,7 @@ return utils.get_interface_ip_addresses(self._ssh_session, interface) - def renew_test_interface_ip_address(self): + def renew_test_interface_ip_address(self) -> None: """Renews the test interface's IPv4 address. Necessary for changing DHCP scopes during a test. @@ -489,7 +493,9 @@ self.start_ssh() utils.renew_linux_ip_address(self._ssh_session, self.test_interface) - def get_addr(self, addr_type="ipv4_private", timeout_sec=None): + def get_addr( + self, addr_type: str = "ipv4_private", timeout_sec: int | None = None + ) -> str: """Wait until a type of IP address on the test interface is available then return it. """ @@ -499,7 +505,7 @@ self._ssh_session, self.test_interface, addr_type, timeout_sec ) - def _cleanup_iperf_port(self): + def _cleanup_iperf_port(self) -> None: """Checks and kills zombie iperf servers occupying intended port.""" iperf_check_cmd = ( "netstat -tulpn | grep LISTEN | grep iperf3" " | grep :{}" @@ -511,12 +517,13 @@ iperf_pid = iperf_check.split(" ")[-1].split("/")[0] self._ssh_session.run(f"kill -9 {str(iperf_pid)}") - def start(self, extra_args="", tag="", iperf_binary=None): + def start( + self, extra_args: str = "", tag: str = "", iperf_binary: str | None = None + ) -> None: """Starts iperf server on specified machine and port. Args: - extra_args: A string representing extra arguments to start iperf - server with. + extra_args: Extra arguments to start iperf server with. tag: Appended to log file name to identify logs from different iperf runs. iperf_binary: Location of iperf3 binary. If none, it is assumed the @@ -543,7 +550,7 @@ self._iperf_pid = job_result.stdout self._current_tag = tag - def stop(self): + def stop(self) -> str: """Stops the iperf server. Returns: @@ -567,12 +574,12 @@ self._iperf_pid = None return log_file - def start_ssh(self): + def start_ssh(self) -> None: """Starts an ssh session to the iperf server.""" if not self._ssh_session: self._ssh_session = connection.SshConnection(self.ssh_settings) - def close_ssh(self): + def close_ssh(self) -> None: """Closes the ssh session to the iperf server, if one exists, preventing connection reset errors when rebooting server device. """
diff --git a/packages/antlion/decorators.py b/packages/antlion/decorators.py index c41150c..58eb3d0 100644 --- a/packages/antlion/decorators.py +++ b/packages/antlion/decorators.py
@@ -16,9 +16,10 @@ from __future__ import annotations +import typing from threading import RLock from types import GenericAlias -from typing import Any, Callable, TypeVar +from typing import Callable, Generic, TypeVar S = TypeVar("S") T = TypeVar("T") @@ -28,7 +29,7 @@ _NOT_FOUND = object() -class cached_property: +class cached_property(Generic[T]): """A property whose value is computed then cached; deleter can be overridden. Similar to functools.cached_property(), with the addition of deleter function that @@ -61,7 +62,7 @@ self.__doc__ = func.__doc__ self.lock = RLock() - def __set_name__(self, owner: O, name: str): + def __set_name__(self, owner: O, name: str) -> None: if self.name is None: self.name = name elif name != self.name: @@ -70,7 +71,7 @@ f"({self.name!r} and {name!r})." ) - def _cache(self, instance: S) -> dict[str, Any]: + def _cache(self, instance: S) -> dict[str, object]: if self.name is None: raise TypeError( "Cannot use cached_property instance without calling __set_name__ on it." @@ -86,7 +87,7 @@ ) raise TypeError(msg) from None - def __get__(self, instance: S, owner: O | None = None): + def __get__(self, instance: S, owner: O | None = None) -> T: cache = self._cache(instance) assert self.name is not None val = cache.get(self.name, _NOT_FOUND) @@ -104,17 +105,20 @@ f"does not support item assignment for caching {self.name!r} property." ) raise TypeError(msg) from None - return val + return val + return typing.cast(T, val) def __delete__(self, instance: S) -> None: cache = self._cache(instance) assert self.name is not None with self.lock: val = cache.pop(self.name, _NOT_FOUND) - if self._deleter and val is not _NOT_FOUND: - self._deleter(instance, val) + if val is _NOT_FOUND: + return + if self._deleter: + self._deleter(instance, typing.cast(T, val)) - def deleter(self, deleter: Callable[[S, T], None]): + def deleter(self, deleter: Callable[[S, T], None]) -> cached_property: self._deleter = deleter prop = type(self)(self.func, deleter) prop.name = self.name
diff --git a/packages/antlion/utils.py b/packages/antlion/utils.py index 1b9cecb..b4e8aa0 100755 --- a/packages/antlion/utils.py +++ b/packages/antlion/utils.py
@@ -1584,7 +1584,12 @@ pass -def get_addr(comm_channel, interface, addr_type="ipv4_private", timeout_sec=None): +def get_addr( + comm_channel: AndroidDevice | SshConnection | FuchsiaDevice, + interface: str, + addr_type: str = "ipv4_private", + timeout_sec: int | None = None, +) -> str: """Get the requested type of IP address for an interface; if an address is not available, retry until the timeout has been reached. @@ -1628,7 +1633,7 @@ raise AddressTimeout(f'No available "{addr_type}" address after {timeout_sec}s') -def get_interface_based_on_ip(comm_channel, desired_ip_address): +def get_interface_based_on_ip(comm_channel: Any, desired_ip_address: str) -> str: """Gets the interface for a particular IP Args: @@ -1650,7 +1655,7 @@ return None -def renew_linux_ip_address(comm_channel, interface): +def renew_linux_ip_address(comm_channel: Any, interface: str): comm_channel.run(f"sudo ip link set {interface} down") comm_channel.run(f"sudo ip link set {interface} up") comm_channel.run(f"sudo dhclient -r {interface}") @@ -1885,8 +1890,8 @@ return [int(octet, 16) for octet in mac_addr_str.split(":")] -def mac_address_list_to_str(mac_addr_list): - """Converts list of decimal octets represeting mac address to string. +def mac_address_list_to_str(mac_addr_list: bytes) -> str: + """Converts list of decimal octets representing mac address to string. Args: mac_addr_list: list, representing mac address octets in decimal
diff --git a/tests/wlan/facade/WlanDeprecatedConfigurationTest.py b/tests/wlan/facade/WlanDeprecatedConfigurationTest.py index 4e00d7b..326dbdd 100644 --- a/tests/wlan/facade/WlanDeprecatedConfigurationTest.py +++ b/tests/wlan/facade/WlanDeprecatedConfigurationTest.py
@@ -19,15 +19,16 @@ from mobly import asserts, test_runner from antlion import utils +from antlion.controllers.ap_lib.hostapd_security import FuchsiaSecurityType +from antlion.controllers.fuchsia_lib.wlan_ap_policy_lib import ( + ConnectivityMode, + OperatingBand, +) from antlion.test_utils.abstract_devices.wlan_device import create_wlan_device from antlion.test_utils.wifi import base_test AP_ROLE = "Ap" DEFAULT_SSID = "testssid" -DEFAULT_SECURITY = "none" -DEFAULT_PASSWORD = "" -DEFAULT_CONNECTIVITY_MODE = "local_only" -DEFAULT_OPERATING_BAND = "any" TEST_MAC_ADDR = "12:34:56:78:9a:bc" TEST_MAC_ADDR_SECONDARY = "bc:9a:78:56:34:12" @@ -91,10 +92,10 @@ self.log.info(f"Starting SoftAP on device {self.dut.identifier}") response = self.fuchsia_device.sl4f.wlan_ap_policy_lib.wlanStartAccessPoint( DEFAULT_SSID, - DEFAULT_SECURITY, - DEFAULT_PASSWORD, - DEFAULT_CONNECTIVITY_MODE, - DEFAULT_OPERATING_BAND, + FuchsiaSecurityType.NONE, + None, + ConnectivityMode.LOCAL_ONLY, + OperatingBand.ANY, ) if response.get("error"): raise ConnectionError(f"Failed to setup SoftAP: {response['error']}")
diff --git a/tests/wlan/functional/ChannelSwitchTest.py b/tests/wlan/functional/ChannelSwitchTest.py index 2b20493..085bf03 100644 --- a/tests/wlan/functional/ChannelSwitchTest.py +++ b/tests/wlan/functional/ChannelSwitchTest.py
@@ -26,6 +26,11 @@ from antlion.controllers.access_point import setup_ap from antlion.controllers.ap_lib import hostapd_constants +from antlion.controllers.ap_lib.hostapd_security import FuchsiaSecurityType +from antlion.controllers.fuchsia_lib.wlan_ap_policy_lib import ( + ConnectivityMode, + OperatingBand, +) from antlion.test_utils.abstract_devices.wlan_device import create_wlan_device from antlion.test_utils.wifi import base_test from antlion.utils import rand_ascii_str @@ -341,15 +346,14 @@ EnvironmentError: if the SoftAP does not start """ ssid = rand_ascii_str(10) - security_type = "none" - password = "" - connectivity_mode = "local_only" - operating_band = "any" - - self.log.info("Starting SoftAP on DUT") + self.log.info(f'Starting SoftAP on DUT with ssid "{ssid}"') response = self.fuchsia_device.sl4f.wlan_ap_policy_lib.wlanStartAccessPoint( - ssid, security_type, password, connectivity_mode, operating_band + ssid, + FuchsiaSecurityType.NONE, + None, + ConnectivityMode.LOCAL_ONLY, + OperatingBand.ANY, ) if response.get("error"): raise EnvironmentError(
diff --git a/tests/wlan/performance/ChannelSweepTest.py b/tests/wlan/performance/ChannelSweepTest.py index 5f307ae..0919b99 100644 --- a/tests/wlan/performance/ChannelSweepTest.py +++ b/tests/wlan/performance/ChannelSweepTest.py
@@ -29,8 +29,9 @@ from antlion.controllers.ap_lib import hostapd_constants from antlion.controllers.ap_lib.hostapd_security import Security, SecurityMode from antlion.controllers.ap_lib.regulatory_channels import COUNTRY_CHANNELS +from antlion.controllers.fuchsia_device import FuchsiaDevice from antlion.controllers.iperf_client import IPerfClientOverAdb, IPerfClientOverSsh -from antlion.controllers.iperf_server import IPerfResult +from antlion.controllers.iperf_server import IPerfResult, IPerfServerOverSsh from antlion.test_utils.abstract_devices.wlan_device import create_wlan_device from antlion.test_utils.wifi import base_test @@ -274,18 +275,20 @@ f"channel bandwidth: {channel_bandwidth} MHz. " ) from err - def get_and_verify_iperf_address(self, channel, device, interface=None): + def get_and_verify_iperf_address( + self, channel: int, device: FuchsiaDevice | IPerfServerOverSsh, interface: str + ) -> str: """Get ip address from a devices interface and verify it belongs to expected subnet based on APs DHCP config. Args: - channel: int, channel network is running on, to determine subnet + channel: channel network is running on, to determine subnet device: device to get ip address for - interface (default: None): interface on device to get ip address. - If None, uses device.test_interface. + interface: interface on device to get ip address. If None, uses + device.test_interface. Returns: - String, ip address of device on given interface (or test_interface) + IP address of device on given interface (or test_interface) Raises: ConnectionError, if device does not have a valid ip address after @@ -297,13 +300,7 @@ subnet = self.access_point._AP_5G_SUBNET_STR end_time = time.time() + self.time_to_wait_for_ip_addr while time.time() < end_time: - if interface: - device_addresses = device.get_interface_ip_addresses(interface) - else: - device_addresses = device.get_interface_ip_addresses( - device.test_interface - ) - + device_addresses = device.get_interface_ip_addresses(interface) if device_addresses["ipv4_private"]: for ip_addr in device_addresses["ipv4_private"]: if utils.ip_in_subnet(ip_addr, subnet): @@ -598,21 +595,35 @@ self.log.info(f"DUT ({self.dut.identifier}) connected to network {ssid}.") if self.iperf_server: self.iperf_server.renew_test_interface_ip_address() + if not isinstance(self.iperf_server.test_interface, str): + raise TypeError( + "For this test, iperf_server is required to specify the " + "test_interface configuration option" + ) + self.log.info( "Getting ip address for iperf server. Will retry for " f"{self.time_to_wait_for_ip_addr} seconds." ) iperf_server_address = self.get_and_verify_iperf_address( - test.channel, self.iperf_server + test.channel, self.iperf_server, self.iperf_server.test_interface ) self.log.info( "Getting ip address for DUT. Will retry for " f"{self.time_to_wait_for_ip_addr} seconds." ) - assert isinstance( + if not isinstance( self.iperf_client, (IPerfClientOverSsh, IPerfClientOverAdb) - ) + ): + raise TypeError( + f'Unknown iperf_client type "{type(self.iperf_client)}"' + ) + if not isinstance(self.iperf_client.test_interface, str): + raise TypeError( + "For this test, iperf_client is required to specify the " + "test_interface configuration option" + ) iperf_client_address = self.get_and_verify_iperf_address( test.channel, self.fuchsia_device, self.iperf_client.test_interface )
diff --git a/tests/wlan_policy/RegulatoryRecoveryTest.py b/tests/wlan_policy/RegulatoryRecoveryTest.py index 5cbfbf5..3301894 100644 --- a/tests/wlan_policy/RegulatoryRecoveryTest.py +++ b/tests/wlan_policy/RegulatoryRecoveryTest.py
@@ -16,6 +16,11 @@ from mobly import signals, test_runner +from antlion.controllers.ap_lib.hostapd_security import FuchsiaSecurityType +from antlion.controllers.fuchsia_lib.wlan_ap_policy_lib import ( + ConnectivityMode, + OperatingBand, +) from antlion.test_utils.wifi import base_test @@ -132,13 +137,17 @@ interfaces are recreated. """ test_ssid = "test_ssid" - test_security_type = "none" + security_type = FuchsiaSecurityType.NONE for fd in self.fuchsia_devices: # Start client connections and start an AP before setting the # country code. fd.wlan_policy_controller.start_client_connections() fd.sl4f.wlan_ap_policy_lib.wlanStartAccessPoint( - test_ssid, test_security_type, "", "local_only", "any" + test_ssid, + security_type, + None, + ConnectivityMode.LOCAL_ONLY, + OperatingBand.ANY, ) # Set the country code. @@ -174,7 +183,7 @@ else: if ( ap_updates[0]["id"]["ssid"] != test_ssid - or ap_updates[0]["id"]["type_"].lower() != test_security_type + or ap_updates[0]["id"]["type_"].lower() != security_type ): raise signals.TestFailure( f"AP in unexpected state: {ap_updates[0]}"