diff --git a/BlocksScreen/lib/network/manager.py b/BlocksScreen/lib/network/manager.py index 08b2af58..4789306c 100644 --- a/BlocksScreen/lib/network/manager.py +++ b/BlocksScreen/lib/network/manager.py @@ -43,6 +43,7 @@ class NetworkManager(QObject): error_occurred = pyqtSignal(str, str) reconnect_complete = pyqtSignal() hotspot_config_updated = pyqtSignal(str, str, str) + network_password_loaded = pyqtSignal(str, str) def __init__(self, parent: QObject | None = None) -> None: """Create the worker, wire all signals""" @@ -57,7 +58,7 @@ def __init__(self, parent: QObject | None = None) -> None: self._shutting_down: bool = False self._worker_ready: bool = False - self._pending_futures: set["asyncio.Future"] = set() + self._pending_futures: set[asyncio.Future] = set() self._worker = NetworkManagerWorker() @@ -72,6 +73,7 @@ def __init__(self, parent: QObject | None = None) -> None: self._worker.error_occurred.connect(self.error_occurred) self._worker.hotspot_info_ready.connect(self._on_hotspot_info_ready) self._worker.reconnect_complete.connect(self.reconnect_complete) + self._worker.network_password_loaded.connect(self.network_password_loaded) self._worker.initialized.connect(self._on_worker_initialized) # Keepalive timer: safety net for any missed D-Bus signals. @@ -170,9 +172,11 @@ def _on_networks_scanned(self, networks: list) -> None: @pyqtSlot(list) def _on_saved_networks_loaded(self, networks: list) -> None: - """Cache saved profiles, rebuild lowercase lookup map, and re-emit.""" + """Cache saved profiles, rebuild lowercase lookup map, and re-emit if changed.""" if self._shutting_down: return + if networks == self._cached_saved: + return self._cached_saved = networks self._saved_network_map = {n.ssid.lower(): n for n in networks} self.saved_networks_loaded.emit(networks) @@ -249,6 +253,10 @@ def update_network( # nosec B107 """Update the password and/or autoconnect priority for a saved profile.""" self._schedule(self._worker._async_update_network(ssid, password, priority)) + def get_network_password(self, ssid: str) -> None: + """Ask NM for a saved profile's psk; answered by network_password_loaded.""" + self._schedule(self._worker._async_get_network_password(ssid)) + def set_wifi_enabled(self, enabled: bool) -> None: """Enable or disable the Wi-Fi radio.""" self._schedule(self._worker._async_set_wifi_enabled(enabled)) diff --git a/BlocksScreen/lib/network/models.py b/BlocksScreen/lib/network/models.py index 5bae85ae..2c88aabe 100644 --- a/BlocksScreen/lib/network/models.py +++ b/BlocksScreen/lib/network/models.py @@ -237,7 +237,7 @@ class SavedNetwork: signal_strength: int = 0 timestamp: int = 0 # Unix time of last successful activation is_dhcp: bool = True # True = auto (DHCP), False = manual (static IP) - ip_address: str = "" + ip_address: str = "" # static IPv4 config, all empty while on DHCP netmask: str = "" gateway: str = "" dns_servers: tuple[str, ...] = () diff --git a/BlocksScreen/lib/network/worker.py b/BlocksScreen/lib/network/worker.py index a0c1632f..2d07bc5a 100644 --- a/BlocksScreen/lib/network/worker.py +++ b/BlocksScreen/lib/network/worker.py @@ -6,6 +6,7 @@ import logging import os import socket as _socket +import string import struct import threading from collections.abc import Awaitable, Callable @@ -15,6 +16,7 @@ from configfile import get_configparser from PyQt6.QtCore import QObject, pyqtSignal from sdbus_async import networkmanager as dbus_nm +from sdbus_async.networkmanager import exceptions as nm_exc from .models import ( ConnectionPriority, @@ -49,6 +51,14 @@ _SHUTDOWN_DRAIN_TIMEOUT: float = 2.0 # Timeout for _wait_for_connection: must cover 802.11 handshake + DHCP. _WIFI_CONNECT_TIMEOUT: float = 20.0 +# WPA-PSK passphrase bounds NM enforces; 64 chars is the raw hex key. +PSK_MIN_LENGTH: int = 8 +PSK_MAX_LENGTH: int = 63 +PSK_HEX_LENGTH: int = 64 +# Only these active-connection types may supply the IP shown to the user. +_PHYSICAL_CONNECTION_TYPES: frozenset[str] = frozenset( + {"802-11-wireless", "802-3-ethernet", "vlan", "bridge", "bond"} +) class NetworkManagerWorker(QObject): @@ -67,6 +77,7 @@ class NetworkManagerWorker(QObject): connectivity_changed = pyqtSignal(ConnectivityState, name="connectivityChanged") error_occurred = pyqtSignal(str, str, name="errorOccurred") hotspot_info_ready = pyqtSignal(str, str, str, name="hotspotInfoReady") + network_password_loaded = pyqtSignal(str, str, name="networkPasswordLoaded") reconnect_complete = pyqtSignal(name="reconnectComplete") _MAX_DBUS_ERRORS_BEFORE_RECONNECT: int = 3 @@ -88,7 +99,7 @@ def __init__(self) -> None: # Set once no interface was found, so rediscovery does not re-alarm the UI. self._no_iface_reported: bool = False - # Path strings only - read-proxies are always created fresh. + # Path strings only: read-proxies are always created fresh. self._primary_wifi_path: str = "" self._primary_wifi_iface: str = "" self._primary_wired_path: str = "" @@ -190,9 +201,7 @@ async def _async_shutdown(self) -> None: pending = [ task for task in {*self._listener_tasks, *self._background_tasks} - if task is not current - and isinstance(task, asyncio.Task) - and not task.done() + if task is not current and not task.done() ] for task in pending: task.cancel() @@ -333,8 +342,11 @@ async def _async_initialize(self) -> None: Detects network interfaces, activates any saved VLANs if ethernet is present, triggers an initial Wi-Fi scan, and starts all D-Bus signal - listeners. Emits ``initialized`` when done (even on failure, so the + listeners. Emits ``initialized`` when done (even on failure, so the manager can unblock its caller). + + Wired autoconnect is deliberately not re-armed here: NM's latch is what + persists the user's "ethernet off" choice across reboots. """ try: if not self._system_bus: @@ -353,14 +365,15 @@ async def _async_initialize(self) -> None: self._hotspot_config.security, ) + # Listeners first: AccessPointAdded fired during the initial scan is lost otherwise. + await self._start_signal_listeners() + if self._primary_wifi_path: try: await self._wifi().request_scan({}) except Exception as exc: logger.debug("Initial Wi-Fi scan request ignored: %s", exc) - await self._start_signal_listeners() - logger.info( "NetworkManagerWorker initialised on thread '%s' " "(sdbus_async, signal-reactive)", @@ -378,7 +391,7 @@ async def _detect_interfaces(self) -> None: Iterates all NetworkManager devices, maps interface names to D-Bus object paths, and stores the first WIFI and ETHERNET device found as the primary interfaces used for all subsequent operations. Emits - ``error_occurred`` once if no interfaces at all are found. + ``error_occurred`` if no interfaces at all are found. """ try: devices = await self._nm().get_devices() @@ -403,6 +416,14 @@ async def _detect_interfaces(self) -> None: ): self._primary_wired_path = device_path self._primary_wired_iface = iface_name + logger.info( + "detect_interfaces: wifi=%s(%s) wired=%s(%s) of %d device(s)", + self._primary_wifi_iface, + self._primary_wifi_path, + self._primary_wired_iface, + self._primary_wired_path, + len(devices), + ) except Exception as exc: logger.error("Failed to detect interfaces: %s", exc) @@ -439,7 +460,7 @@ async def _ensure_wired_autoconnect(self) -> None: """Re-arm wired autoconnect on both the device and the saved profiles. Called only when the user asks for ethernet, so autoconnect staying off - keeps meaning "user turned it off". Best-effort: never propagates. + keeps meaning "user turned it off". Best-effort: never propagates. """ if not self._primary_wired_path: return @@ -448,8 +469,9 @@ async def _ensure_wired_autoconnect(self) -> None: wired = self._generic(self._primary_wired_path) if not await wired.autoconnect: await wired.autoconnect.set_async(True) + logger.info("Re-armed wired device autoconnect") except Exception as exc: - logger.debug("Device autoconnect re-arm ignored: %s", exc) + logger.warning("Wired autoconnect re-arm failed (non-fatal): %s", exc) async def _start_signal_listeners(self) -> None: """Create persistent proxies and spawn all D-Bus signal listeners. @@ -583,16 +605,15 @@ async def _listen_nm_state_changed(self) -> None: async def _listen_ap_added(self) -> None: """React to new access points appearing in scan results. - Triggers a debounced scan rebuild (not a full rescan — NM has + Triggers a debounced scan rebuild (not a full rescan: NM has already updated its internal AP list). """ if not self._signal_wifi: return logger.debug("AP Added listener started on %s", self._primary_wifi_path) - async for ap_path in self._signal_wifi.access_point_added: + async for _ in self._signal_wifi.access_point_added: if not self._running: return - logger.debug("AP added: %s", ap_path) self._schedule_debounced_scan() async def _listen_ap_removed(self) -> None: @@ -600,10 +621,9 @@ async def _listen_ap_removed(self) -> None: if not self._signal_wifi: return logger.debug("AP Removed listener started on %s", self._primary_wifi_path) - async for ap_path in self._signal_wifi.access_point_removed: + async for _ in self._signal_wifi.access_point_removed: if not self._running: return - logger.debug("AP removed: %s", ap_path) self._schedule_debounced_scan() async def _listen_wired_state_changed(self) -> None: @@ -630,7 +650,7 @@ async def _listen_wifi_state_changed(self) -> None: """React to Wi-Fi device state transitions. Detects enabled/disabled, connecting, disconnected transitions - instantly — complements the NM global ``state_changed`` signal + instantly: complements the NM global ``state_changed`` signal which may not fire for all device-level transitions. """ if not self._signal_wifi: @@ -695,7 +715,7 @@ def _schedule_debounced_state_rebuild(self) -> None: ) def _fire_state_rebuild(self) -> None: - """Debounce callback — spawns the actual async state rebuild.""" + """Debounce callback: spawns the actual async state rebuild.""" self._state_debounce_handle = None if self._running: self._track_task( @@ -719,7 +739,7 @@ def _schedule_debounced_scan(self) -> None: ) def _fire_scan_rebuild(self) -> None: - """Debounce callback — spawns the async scan rebuild.""" + """Debounce callback: spawns the async scan rebuild.""" self._scan_debounce_handle = None if self._running: self._track_task( @@ -756,6 +776,8 @@ async def _ensure_dbus_connection(self) -> bool: try: _ = await self._nm().version self._consecutive_dbus_errors = 0 + # The bus survives an NM restart but device paths do not. + await self._recover_signal_sources() return True except Exception as exc: self._consecutive_dbus_errors += 1 @@ -783,8 +805,7 @@ async def _ensure_dbus_connection(self) -> bool: self._signal_wired = None self._signal_settings = None self._ensure_signal_proxies() - # Cancel stale listener tasks bound to old proxies - # and restart them on the new bus connection. + # Listener tasks hold old proxies; restart them on the new bus. for task in self._listener_tasks: if not task.done(): task.cancel() @@ -811,6 +832,30 @@ async def _is_ethernet_connected(self) -> bool: logger.debug("Error checking ethernet state: %s", exc) return False + async def _wifi_device_state(self) -> int: + """Return the primary Wi-Fi device's NM state, or -1 when unreadable.""" + if not self._primary_wifi_path: + return -1 + try: + return await self._generic(self._primary_wifi_path).state + except Exception as exc: + logger.debug("Error checking Wi-Fi device state: %s", exc) + return -1 + + async def _wifi_activation_failed(self) -> bool: + """Return True if the Wi-Fi device is in NM's terminal FAILED state (120).""" + return await self._wifi_device_state() == 120 + + async def _is_wifi_ap_mode(self) -> bool: + """True when the radio sits in NM's AP mode (3), whoever started the hotspot.""" + if not self._primary_wifi_path: + return False + try: + return int(await self._wifi(self._primary_wifi_path).mode) == 3 + except Exception as exc: + logger.debug("Error reading Wi-Fi device mode: %s", exc) + return False + async def _has_ethernet_carrier(self) -> bool: """Return True if the primary wired device has a physical link (state >= 30). @@ -823,7 +868,7 @@ async def _has_ethernet_carrier(self) -> bool: try: return await self._generic(self._primary_wired_path).state >= 30 except Exception: - # D-Bus read failed; carrier state unknown — treat as no carrier. + # D-Bus read failed; carrier state unknown: treat as no carrier. return False async def _wait_for_wifi_radio(self, desired: bool, timeout: float = 3.0) -> bool: @@ -897,31 +942,12 @@ async def _wait_for_wifi_device_ready(self, timeout: float = 8.0) -> bool: return False async def _async_get_current_state(self) -> None: - """Rebuild and emit the full NetworkState, enforcing runtime mutual exclusion.""" + """Rebuild and emit the full NetworkState. Read-only: never mutates NM.""" try: if not await self._ensure_dbus_connection(): self.state_changed.emit(NetworkState()) return - state = await self._build_current_state() - if ( - state.ethernet_connected - and state.wifi_enabled - and not state.hotspot_enabled - and not self._is_hotspot_active - ): - logger.info( - "Runtime mutual exclusion: ethernet active + " - "Wi-Fi — disabling Wi-Fi" - ) - if self._primary_wifi_path: - try: - await self._wifi().disconnect() - except Exception as exc: - logger.debug("Disconnect before Wi-Fi disable ignored: %s", exc) - await self._nm().wireless_enabled.set_async(False) - await asyncio.sleep(0.5) - state = await self._build_current_state() - self.state_changed.emit(state) + self.state_changed.emit(await self._build_current_state()) except Exception as exc: logger.error("Failed to get current state: %s", exc) self.error_occurred.emit("get_current_state", str(exc)) @@ -930,7 +956,7 @@ async def _async_get_current_state(self) -> None: def _get_ip_os_fallback(iface: str) -> str: """Return the IPv4 address for *iface* via a raw ioctl SIOCGIFADDR call. - Used as a fallback when the NM D-Bus IPv4Config path returns nothing — + Fallback for when the NM IPv4Config path returns nothing, which is common immediately after DHCP on slower hardware. """ if not iface: @@ -950,67 +976,53 @@ async def _build_current_state(self) -> NetworkState: if not self._system_bus: return NetworkState() try: - connectivity_value = await self._nm().check_connectivity() - connectivity = self._map_connectivity(connectivity_value) - wifi_enabled = bool(await self._nm().wireless_enabled) - current_ssid = await self._get_current_ssid() + wifi_iface = self._get_wifi_iface_name() + # Independent property reads: one round-trip batch instead of four. + ( + connectivity, + wifi_raw, + current_ssid, + eth_connected, + ap_mode, + ) = await asyncio.gather( + self._read_connectivity(), + self._nm().wireless_enabled, + self._get_current_ssid(), + self._is_ethernet_connected(), + self._is_wifi_ap_mode(), + ) + wifi_enabled = bool(wifi_raw) - eth_connected = await self._is_ethernet_connected() - if eth_connected: - current_ip = await self._get_ip_by_interface( - self._primary_wired_iface or "eth0" - ) - if not current_ip: - current_ip = self._get_ip_os_fallback( - self._primary_wired_iface or "eth0" - ) - current_ssid = "" - elif current_ssid: - current_ip = await self._get_ip_by_interface("wlan0") - if not current_ip: - current_ip = await self._get_current_ip() - else: - current_ip = "" + current_ip, eth_connected = await self._resolve_current_ip( + wifi_iface, current_ssid, bool(eth_connected), connectivity + ) + signal, sec_type = await self._wifi_signal_and_security(current_ssid) - if not current_ip and connectivity in ( - ConnectivityState.FULL, - ConnectivityState.LIMITED, - ): - for _iface in ( - self._primary_wired_iface or "eth0", - "wlan0", - ): - _fallback = self._get_ip_os_fallback(_iface) - if _fallback: - current_ip = _fallback - if _iface != "wlan0": - eth_connected = True - logger.debug("OS fallback IP for '%s': %s", _iface, _fallback) - break - - signal = 0 - sec_type = "" - if current_ssid: - signal_map = await self._build_signal_map() - signal = signal_map.get(current_ssid.lower(), 0) - saved = await self._get_saved_network_cached(current_ssid) - sec_type = saved.security_type if saved else "" - - hotspot_enabled = current_ssid == self._hotspot_config.ssid + # Device mode is authoritative; the SSID match only covers our own AP. + hotspot_enabled = bool(ap_mode) or current_ssid == self._hotspot_config.ssid if not hotspot_enabled and self._is_hotspot_active and not current_ssid: hotspot_enabled = True current_ssid = self._hotspot_config.ssid - logger.debug( - "Hotspot SSID not found via D-Bus, using config: '%s'", - current_ssid, - ) if hotspot_enabled: sec_type = self._hotspot_config.security if not current_ip: - current_ip = await self._get_ip_by_interface("wlan0") + current_ip = await self._get_ip_by_interface(wifi_iface) + carrier, vlans = await asyncio.gather( + self._has_ethernet_carrier(), self._get_active_vlans() + ) + logger.info( + "state: conn=%s ssid='%s' ip=%s wifi=%s eth=%s hotspot=%s sig=%d", + connectivity.name, + current_ssid, + current_ip or "-", + wifi_enabled, + eth_connected, + hotspot_enabled, + signal, + ) return NetworkState( connectivity=connectivity, current_ssid=current_ssid, @@ -1020,13 +1032,57 @@ async def _build_current_state(self) -> NetworkState: signal_strength=signal, security_type=sec_type, ethernet_connected=eth_connected, - ethernet_carrier=await self._has_ethernet_carrier(), - active_vlans=await self._get_active_vlans(), + ethernet_carrier=carrier, + active_vlans=vlans, ) except Exception as exc: logger.error("Error building current state: %s", exc) return NetworkState() + async def _resolve_current_ip( + self, + wifi_iface: str, + ssid: str, + eth_connected: bool, + connectivity: ConnectivityState, + ) -> tuple[str, bool]: + """Resolve the active IPv4 address, returning it plus the ethernet flag.""" + wired = self._primary_wired_iface or "eth0" + if eth_connected: + # Wi-Fi may stay up alongside the cable; keep the SSID for signal. + current_ip = await self._get_ip_by_interface(wired) + if not current_ip: + current_ip = self._get_ip_os_fallback(wired) + elif ssid: + current_ip = await self._get_ip_by_interface(wifi_iface) + if not current_ip: + current_ip = await self._get_current_ip() + else: + current_ip = "" + + if current_ip or connectivity not in ( + ConnectivityState.FULL, + ConnectivityState.LIMITED, + ): + return current_ip, eth_connected + + for iface in (wired, wifi_iface): + fallback = self._get_ip_os_fallback(iface) + if fallback: + return fallback, eth_connected or iface != wifi_iface + return current_ip, eth_connected + + async def _wifi_signal_and_security(self, ssid: str) -> tuple[int, str]: + """Return the live signal strength and saved security type for ``ssid``.""" + if not ssid: + return 0, "" + # The associated AP has live strength; the scan cache may be empty. + signal = await self._active_ap_signal() + if not signal: + signal = (await self._build_signal_map()).get(ssid.lower(), 0) + saved = await self._get_saved_network_cached(ssid) + return signal, saved.security_type if saved else "" + @staticmethod def _map_connectivity(value: int) -> ConnectivityState: """Map a raw NM connectivity integer to a ConnectivityState enum member.""" @@ -1035,15 +1091,21 @@ def _map_connectivity(value: int) -> ConnectivityState: except ValueError: return ConnectivityState.UNKNOWN + async def _read_connectivity(self) -> ConnectivityState: + """Cached connectivity property, active probe only if UNKNOWN.""" + nm = self._nm() + state = self._map_connectivity(await nm.connectivity) + if state is ConnectivityState.UNKNOWN: + state = self._map_connectivity(await nm.check_connectivity()) + return state + async def _async_check_connectivity(self) -> None: """Query NM connectivity and emit connectivity_changed.""" try: if not self._system_bus: self.connectivity_changed.emit(ConnectivityState.UNKNOWN) return - self.connectivity_changed.emit( - self._map_connectivity(await self._nm().check_connectivity()) - ) + self.connectivity_changed.emit(await self._read_connectivity()) except Exception as exc: logger.error("Failed to check connectivity: %s", exc) self.connectivity_changed.emit(ConnectivityState.UNKNOWN) @@ -1097,6 +1159,10 @@ async def _get_current_ip(self) -> str: primary_con = await self._nm().primary_connection if primary_con == "/": return "" + # NM can promote a VPN to primary; its IP is not reachable on the LAN. + conn_type = await self._active_conn(primary_con).connection_type + if conn_type not in _PHYSICAL_CONNECTION_TYPES: + return "" ip4_path = await self._active_conn(primary_con).ip4_config if ip4_path == "/": return "" @@ -1257,27 +1323,67 @@ async def _gather_ap_properties( ) return list(zip(ap_paths, props)) + async def _gather_settings(self, paths: list[str]) -> list[tuple[str, dict]]: + """Read every profile's settings in one concurrent batch, dropping failures.""" + + async def one(path: str) -> tuple[str, dict | None]: + """Fetch one profile's settings, returning None if the read fails.""" + try: + return path, await self._conn_settings(path).get_settings() + except Exception as exc: + logger.debug("GetSettings failed for %s: %s", path, exc) + return path, None + + results = await asyncio.gather(*(one(p) for p in paths)) + return [(p, s) for p, s in results if s is not None] + + async def _active_ap_signal(self) -> int: + """Return the associated AP's live signal strength, or 0 when unavailable.""" + if not self._primary_wifi_path: + return 0 + try: + ap_path = await self._wifi().active_access_point + if not ap_path or ap_path == "/": + return 0 + return int(await self._ap(ap_path).strength) + except Exception as exc: + logger.debug("active_ap_signal failed: %s", exc) + return 0 + async def _build_signal_map(self) -> dict[str, int]: - """Return a mapping of lowercase SSID to best-seen signal strength (0-100).""" - signal_map: dict[str, int] = {} + """Return a mapping of lowercase SSID to best-seen signal strength (0-100). + + Retries once after re-detecting interfaces so a stale Wi-Fi path cannot + leave every saved network stuck showing no signal. + """ if not self._primary_wifi_path: - return signal_map + return {} try: - ap_paths = await self._wifi().access_points - for ap_path in ap_paths: - try: - props = await self._get_all_ap_properties(ap_path) - ssid = self._decode_ssid(props.get("ssid", b"")) - if ssid: - strength = int(props.get("strength", 0)) - key = ssid.lower() - if strength > signal_map.get(key, 0): - signal_map[key] = strength - except Exception as exc: - logger.debug("Skipping AP in signal map: %s", exc) - continue + return await self._signal_map_once() except Exception as exc: - logger.debug("Error building signal map: %s", exc) + logger.warning("Signal map failed (%s), re-detecting interfaces", exc) + try: + await self._recover_signal_sources() + return await self._signal_map_once() + except Exception as retry_exc: + logger.debug("Signal map unavailable: %s", retry_exc) + return {} + + async def _signal_map_once(self) -> dict[str, int]: + """Single signal-map attempt; raises so the caller can recover and retry.""" + signal_map: dict[str, int] = {} + ap_paths = await self._wifi().access_points + for _, props in await self._gather_ap_properties(ap_paths): + try: + ssid = self._decode_ssid(props.get("ssid", b"")) + if ssid: + strength = int(props.get("strength", 0)) + key = ssid.lower() + if strength > signal_map.get(key, 0): + signal_map[key] = strength + except Exception as exc: + logger.debug("Skipping AP in signal map: %s", exc) + continue return signal_map async def _parse_ap( @@ -1322,6 +1428,40 @@ async def _parse_ap( security_type=security, ) + @staticmethod + def _classify_settings_error(exc: Exception) -> ConnectionResult | None: + """Map a typed NM settings error to a user-facing result, or None if unrecognised.""" + if isinstance(exc, nm_exc.NmSettingsPermissionDeniedError): + return ConnectionResult( + False, "Not authorised to change networks", "insufficient_privileges" + ) + if isinstance(exc, nm_exc.NmConnectionInvalidPropertyError): + return ConnectionResult( + False, "Wrong password, try again.", "invalid_password" + ) + # Fallback for NM builds that return an untyped error. + err = str(exc).lower() + if "psk" in err and ("invalid" in err or "property" in err): + return ConnectionResult( + False, "Wrong password, try again.", "invalid_password" + ) + return None + + @staticmethod + def _validate_psk(password: str) -> ConnectionResult | None: + """Reject a WPA passphrase NM would refuse, so we fail before touching the profile.""" + if PSK_MIN_LENGTH <= len(password) <= PSK_MAX_LENGTH: + return None + if len(password) == PSK_HEX_LENGTH and all( + c in string.hexdigits for c in password + ): + return None + return ConnectionResult( + False, + f"Password must be {PSK_MIN_LENGTH}-{PSK_MAX_LENGTH} characters.", + "invalid_password_length", + ) + @staticmethod def _decode_ssid(raw: object) -> str: """Decode a raw SSID byte string to a UTF-8 str, replacing invalid bytes.""" @@ -1382,20 +1522,6 @@ async def _async_load_saved_networks(self) -> None: self.error_occurred.emit("load_saved_networks", str(exc)) self.saved_networks_loaded.emit([]) - async def _gather_settings(self, paths: list[str]) -> list[tuple[str, dict]]: - """Read every profile's settings in one concurrent batch, dropping failures.""" - - async def one(path: str) -> tuple[str, dict | None]: - """Fetch one profile's settings, returning None if the read fails.""" - try: - return path, await self._conn_settings(path).get_settings() - except Exception as exc: - logger.debug("GetSettings failed for %s: %s", path, exc) - return path, None - - results = await asyncio.gather(*(one(p) for p in paths)) - return [(p, s) for p, s in results if s is not None] - async def _get_saved_networks_impl(self) -> list[SavedNetwork]: """Enumerate NM connection profiles and return infrastructure Wi-Fi ones.""" if not self._system_bus: @@ -1486,19 +1612,26 @@ async def _add_network_impl( ) -> ConnectionResult: """Scan for the SSID, build a connection profile, add it to NM, and activate it. - Deletes any pre-existing profile for the same SSID before adding. + Any pre-existing profile for the same SSID is backed up before being + replaced, and restored if the new credentials fail to activate. Returns a failed ConnectionResult if the SSID is not visible, the security type is unsupported, or the 20-second activation wait times out. """ + logger.info("add_network: ssid='%s' priority=%d", ssid, priority) if not self._primary_wifi_path or not self._system_bus: + logger.warning( + "add_network: no wifi interface (path=%s)", self._primary_wifi_path + ) return ConnectionResult(False, "No Wi-Fi interface", "no_interface") + # Guard before the backup/delete below, so a bad psk cannot cost the profile. + if password and (bad := self._validate_psk(password)): + logger.info("add_network: '%s' rejected, psk len=%d", ssid, len(password)) + return bad + backup = await self._backup_and_drop_existing(ssid) - try: - await self._wifi().request_scan({}) - except Exception as exc: - logger.debug("Pre-connect scan request ignored: %s", exc) + await self._request_scan_if_allowed() target_ap_props = await self._find_ap_props(ssid) if target_ap_props is None: @@ -1519,14 +1652,9 @@ async def _add_network_impl( nm_settings = self._nm_settings() conn_path = await nm_settings.add_connection(conn_props) except Exception as exc: - err_str = str(exc).lower() - if "psk" in err_str and ("invalid" in err_str or "property" in err_str): - return ConnectionResult( - False, - "Wrong password, try again.", - "invalid_password", - ) - return ConnectionResult(False, str(exc), "add_failed") + return self._classify_settings_error(exc) or ConnectionResult( + False, str(exc), "add_failed" + ) await self._reload_connections() @@ -1534,6 +1662,7 @@ async def _add_network_impl( await self._nm().activate_connection(conn_path) if not await self._wait_for_connection(ssid, timeout=_WIFI_CONNECT_TIMEOUT): return await self._rollback_failed_add(ssid, backup) + logger.info("add_network: '%s' activated", ssid) return ConnectionResult(True, f"Network '{ssid}' added and connecting") except Exception as act_err: logger.warning("Activate after add failed: %s", act_err) @@ -1548,32 +1677,23 @@ async def _reload_connections(self) -> None: except Exception as reload_err: logger.debug("reload_connections non-fatal: %s", reload_err) - async def _backup_profile(self, ssid: str) -> dict | None: - """Snapshot a saved profile's settings plus secrets so it can be re-added.""" - conn_path = await self._get_connection_path(ssid) - if not conn_path: - return None - try: - cs = self._conn_settings(conn_path) - settings = dict(await cs.get_settings()) - await self._merge_wifi_secrets(cs, settings) - settings.get("connection", {}).pop("timestamp", None) - logger.debug("backup_profile: '%s' sections=%s", ssid, sorted(settings)) - return settings - except Exception as exc: - logger.warning("backup_profile: could not snapshot '%s': %s", ssid, exc) + async def _backup_and_drop_existing(self, ssid: str) -> dict | None: + """Back up and delete a saved profile for *ssid* so it can be re-added cleanly.""" + if not await self._is_known(ssid): return None + backup = await self._backup_profile(ssid) + logger.info("add_network: replacing saved '%s' (backup=%s)", ssid, bool(backup)) + await self._delete_network_impl(ssid) + self._invalidate_saved_cache() + return backup - async def _restore_profile(self, ssid: str, settings: dict) -> bool: - """Re-add a backed-up profile after a failed replacement; True when restored.""" - try: - await self._nm_settings().add_connection(settings) - self._invalidate_saved_cache() - logger.info("restore_profile: '%s' restored after failed add", ssid) - return True - except Exception as exc: - logger.error("restore_profile: could not restore '%s': %s", ssid, exc) - return False + async def _find_ap_props(self, ssid: str) -> dict[str, object] | None: + """Return the scanned AP properties for *ssid*, or None if it is not visible.""" + ap_paths = await self._wifi().get_all_access_points() + for _ap_path, props in await self._gather_ap_properties(ap_paths): + if self._decode_ssid(props.get("ssid", b"")) == ssid: + return props + return None async def _rollback_failed_add( self, ssid: str, backup: dict | None @@ -1598,15 +1718,99 @@ async def _rollback_failed_add( "auth_failed", ) - async def _backup_and_drop_existing(self, ssid: str) -> dict | None: - """Back up and delete a saved profile for *ssid* so it can be re-added cleanly.""" - if not await self._is_known(ssid): + def _build_connection_properties( + self, + ssid: str, + password: str, + interface: str, + priority: int, + ap_props: dict[str, object], + ) -> dict[str, object] | None: + """Build NM connection property dict for *ssid* from its AP capability flags. + + Returns None if the security type is unsupported (e.g. WPA-EAP). + Handles OPEN, WPA-PSK, WPA2-PSK, and WPA3-SAE (including SAE-transition). + """ + flags = int(ap_props.get("flags", 0)) + wpa_flags = int(ap_props.get("wpa_flags", 0)) + rsn_flags = int(ap_props.get("rsn_flags", 0)) + + props: dict[str, object] = { + "connection": { + "id": ("s", ssid), + "uuid": ("s", str(uuid4())), + "type": ("s", "802-11-wireless"), + "interface-name": ("s", interface), + "autoconnect": ("b", True), + "autoconnect-priority": ("i", priority), + }, + "802-11-wireless": { + "mode": ("s", "infrastructure"), + "ssid": ("ay", ssid.encode("utf-8")), + }, + "ipv4": { + "method": ("s", "auto"), + "route-metric": ("i", 200), + }, + "ipv6": {"method": ("s", "auto")}, + } + + if (flags & 1) == 0: + return props + + props["802-11-wireless"]["security"] = ( + "s", + "802-11-wireless-security", + ) + security = self._determine_security_type(flags, wpa_flags, rsn_flags) + + if not is_connectable_security(security): + logger.warning( + "Rejecting connection to '%s': unsupported security %s", + ssid, + security.value, + ) return None - backup = await self._backup_profile(ssid) - logger.info("add_network: replacing saved '%s' (backup=%s)", ssid, bool(backup)) - await self._delete_network_impl(ssid) - self._invalidate_saved_cache() - return backup + + if security == SecurityType.WPA3_SAE: + has_psk = bool((rsn_flags & 0x100) or wpa_flags) + if has_psk: + logger.debug( + "SAE transition for '%s': using wpa-psk + PMF optional", + ssid, + ) + props["802-11-wireless-security"] = { + "key-mgmt": ("s", "wpa-psk"), + "auth-alg": ("s", "open"), + "psk": ("s", password), + "pmf": ("u", 2), # OPTIONAL: required for SAE-transition APs + } + else: + logger.debug("Pure SAE detected for '%s'", ssid) + props["802-11-wireless-security"] = { + "key-mgmt": ("s", "sae"), + "auth-alg": ("s", "open"), + "psk": ("s", password), + "pmf": ("u", 3), # REQUIRED: mandatory for pure WPA3-SAE + } + elif security in ( + SecurityType.WPA2_PSK, + SecurityType.WPA_PSK, + ): + props["802-11-wireless-security"] = { + "key-mgmt": ("s", "wpa-psk"), + "auth-alg": ("s", "open"), + "psk": ("s", password), + } + else: + logger.warning( + "Unsupported security type '%s' for '%s'", + security.value, + ssid, + ) + return None + + return props async def _async_connect_network(self, ssid: str) -> None: """Activate an existing saved Wi-Fi profile and emit connection_result.""" @@ -1631,27 +1835,54 @@ async def _wait_for_connection( """Poll until *ssid* is active and has an IP, or until *timeout* expires. Starts with a 1.5 s initial delay to let NM begin the association. - Returns False early if the SSID disappears for 3 consecutive polls. + Gives up early only once the Wi-Fi device reports a terminal failure. """ loop = asyncio.get_running_loop() - deadline = loop.time() + timeout + started = loop.time() + deadline = started + timeout + last_state = -1 + logger.info("wait_for_connection: '%s' timeout=%.1fs", ssid, timeout) await asyncio.sleep(1.5) - consecutive_empty = 0 while loop.time() < deadline: try: + # Device state transitions are the only trace of a failed join. + state = await self._wifi_device_state() + if state != last_state: + logger.info( + "wait_for_connection: '%s' dev state %d at %.1fs", + ssid, + state, + loop.time() - started, + ) + last_state = state current = await self._get_current_ssid() if current and current.lower() == ssid.lower(): ip = await self._get_current_ip() if ip: + logger.info( + "wait_for_connection: '%s' up with ip=%s after %.1fs", + ssid, + ip, + loop.time() - started, + ) return True - consecutive_empty = 0 - else: - consecutive_empty += 1 - if consecutive_empty >= 3: - return False + elif await self._wifi_activation_failed(): + logger.warning( + "wait_for_connection: '%s' failed (state %d) after %.1fs", + ssid, + state, + loop.time() - started, + ) + return False except Exception as exc: logger.debug("Connection wait poll failed: %s", exc) await asyncio.sleep(0.5) + logger.warning( + "wait_for_connection: '%s' timed out after %.1fs, last state %d", + ssid, + timeout, + last_state, + ) return False async def _connect_network_impl(self, ssid: str) -> ConnectionResult: @@ -1692,9 +1923,8 @@ async def _find_connection_path_direct(self, ssid: str) -> str | None: """Search NM settings for an infrastructure profile matching *ssid* directly.""" try: connections = await self._nm_settings().list_connections() - for conn_path in connections: + for conn_path, settings in await self._gather_settings(connections): try: - settings = await self._conn_settings(conn_path).get_settings() if settings["connection"]["type"][1] != "802-11-wireless": continue conn_ssid = settings["802-11-wireless"]["ssid"][1].decode() @@ -1764,6 +1994,34 @@ async def _delete_network_impl(self, ssid: str) -> ConnectionResult: except Exception as exc: return ConnectionResult(False, str(exc), "delete_failed") + async def _backup_profile(self, ssid: str) -> dict | None: + """Snapshot a saved profile's settings plus secrets so it can be re-added.""" + conn_path = await self._get_connection_path(ssid) + if not conn_path: + return None + try: + cs = self._conn_settings(conn_path) + settings = dict(await cs.get_settings()) + await self._merge_wifi_secrets(cs, settings) + # NM rejects a timestamp it did not write itself. + settings.get("connection", {}).pop("timestamp", None) + logger.debug("backup_profile: '%s' sections=%s", ssid, sorted(settings)) + return settings + except Exception as exc: + logger.warning("backup_profile: could not snapshot '%s': %s", ssid, exc) + return None + + async def _restore_profile(self, ssid: str, settings: dict) -> bool: + """Re-add a backed-up profile after a failed replacement; True when restored.""" + try: + await self._nm_settings().add_connection(settings) + self._invalidate_saved_cache() + logger.info("restore_profile: '%s' restored after failed add", ssid) + return True + except Exception as exc: + logger.error("restore_profile: could not restore '%s': %s", ssid, exc) + return False + async def _async_update_network( self, ssid: str, @@ -1796,6 +2054,11 @@ async def _update_network_impl( priority: int | None, ) -> ConnectionResult: """Merge updated password/priority into the existing NM connection settings.""" + if password and (bad := self._validate_psk(password)): + logger.info( + "update_network: '%s' rejected, psk len=%d", ssid, len(password) + ) + return bad conn_path = await self._get_connection_path(ssid) if not conn_path: return ConnectionResult(False, f"Network '{ssid}' not found", "not_found") @@ -1804,10 +2067,18 @@ async def _update_network_impl( props = await cs.get_settings() await self._merge_wifi_secrets(cs, props) - if password and "802-11-wireless-security" in props: - props["802-11-wireless-security"]["psk"] = ( - "s", - password, + if password: + # NM omits the security section for open profiles; recreate it. + had_sec = "802-11-wireless-security" in props + sec = props.setdefault("802-11-wireless-security", {}) + sec.setdefault("key-mgmt", ("s", "wpa-psk")) + sec["psk"] = ("s", password) + logger.info( + "update_network: '%s' psk len=%d key-mgmt=%s sec_existed=%s", + ssid, + len(password), + sec["key-mgmt"][1], + had_sec, ) if priority is not None: @@ -1818,18 +2089,27 @@ async def _update_network_impl( logger.debug("Setting priority for '%s' to %d", ssid, priority) await cs.update(props) - logger.debug("Network '%s' update() succeeded", ssid) + logger.info("update_network: '%s' update() succeeded", ssid) return ConnectionResult(True, f"Network '{ssid}' updated") except Exception as exc: logger.error("Update failed for '%s': %s", ssid, exc) - err_str = str(exc).lower() - if "psk" in err_str and ("invalid" in err_str or "property" in err_str): - return ConnectionResult( - False, - "Wrong password, try again.", - "invalid_password", - ) - return ConnectionResult(False, str(exc), "update_failed") + return self._classify_settings_error(exc) or ConnectionResult( + False, str(exc), "update_failed" + ) + + async def _async_get_network_password(self, ssid: str) -> None: + """Fetch a saved profile's psk on demand and emit network_password_loaded.""" + password = "" # nosec B105 - empty default, not a credential + try: + conn_path = await self._get_connection_path(ssid) + if conn_path: + cs = self._conn_settings(conn_path) + secrets = await cs.get_secrets("802-11-wireless-security") + sec = secrets.get("802-11-wireless-security", {}) + password = str(self._unwrap(sec.get("psk", "")) or "") + except Exception as exc: + logger.debug("get_network_password: '%s' unavailable: %s", ssid, exc) + self.network_password_loaded.emit(ssid, password) async def _async_set_wifi_enabled(self, enabled: bool) -> None: """Enable or disable the Wi-Fi radio. Ethernet is left untouched.""" @@ -1943,7 +2223,6 @@ async def _async_connect_ethernet(self) -> None: await self._async_toggle_hotspot(False) await self._ensure_wired_autoconnect() - await self._nm().activate_connection("/", self._primary_wired_path, "/") await asyncio.sleep(1.5) @@ -1976,18 +2255,9 @@ async def _async_create_vlan( if self._is_hotspot_active: await self._async_toggle_hotspot(False) - if self._primary_wifi_path: - try: - await self._wifi().disconnect() - except Exception as exc: - logger.debug("Pre-VLAN disconnect ignored: %s", exc) - await asyncio.sleep(0.5) - - if await self._nm().wireless_enabled: - await self._nm().wireless_enabled.set_async(False) - await self._wait_for_wifi_radio(False, timeout=8.0) - + # A VLAN rides on eth0; Wi-Fi is orthogonal and stays up. if not await self._is_ethernet_connected(): + await self._ensure_wired_autoconnect() await self._nm().activate_connection("/", self._primary_wired_path, "/") await asyncio.sleep(1.5) @@ -2013,39 +2283,13 @@ async def _async_create_vlan( await self._delete_all_connections_by_id(vlan_conn_id) await asyncio.sleep(0.5) - prefix = self._mask_to_prefix(subnet_mask) - ip_uint = self._ip_to_nm_uint32(ip_address) - gw_uint = self._ip_to_nm_uint32(gateway) if gateway else 0 - dns_list: list[int] = [] - if dns1: - dns_list.append(self._ip_to_nm_uint32(dns1)) - if dns2: - dns_list.append(self._ip_to_nm_uint32(dns2)) - - conn_props: dict[str, object] = { - "connection": { - "id": ("s", vlan_conn_id), - "uuid": ("s", str(uuid4())), - "type": ("s", "vlan"), - "autoconnect": ("b", False), - }, - "vlan": { - "id": ("u", vlan_id), - "parent": ("s", iface), - }, - "ipv4": { - "method": ("s", "manual"), - "addresses": ( - "aau", - [[ip_uint, prefix, gw_uint]], - ), - "gateway": ("s", gateway or ""), - "dns": ("au", dns_list), - "route-metric": ("i", 500), - }, - "ipv6": {"method": ("s", "ignore")}, - } - + conn_props = self._build_vlan_properties( + vlan_conn_id, + vlan_id, + iface, + (ip_address, subnet_mask, gateway), + (dns1, dns2), + ) conn_path = await self._nm_settings().add_connection(conn_props) await self._nm().activate_connection(conn_path, "/", "/") self.state_changed.emit(await self._build_current_state()) @@ -2079,6 +2323,43 @@ async def _vlan_profile_exists(self, vlan_id: int, iface: str) -> bool: logger.debug("Duplicate VLAN check failed (non-fatal): %s", dup_err) return False + def _build_vlan_properties( + self, + conn_id: str, + vlan_id: int, + iface: str, + ipv4: tuple[str, str, str], + dns: tuple[str, str], + ) -> dict[str, object]: + """Build the NM property dict for a static-IP VLAN profile on *iface*.""" + ip_address, subnet_mask, gateway = ipv4 + dns_list = [self._ip_to_nm_uint32(d) for d in dns if d] + addr = [ + self._ip_to_nm_uint32(ip_address), + self._mask_to_prefix(subnet_mask), + self._ip_to_nm_uint32(gateway) if gateway else 0, + ] + return { + "connection": { + "id": ("s", conn_id), + "uuid": ("s", str(uuid4())), + "type": ("s", "vlan"), + "autoconnect": ("b", False), + }, + "vlan": { + "id": ("u", vlan_id), + "parent": ("s", iface), + }, + "ipv4": { + "method": ("s", "manual"), + "addresses": ("aau", [addr]), + "gateway": ("s", gateway or ""), + "dns": ("au", dns_list), + "route-metric": ("i", 500), + }, + "ipv6": {"method": ("s", "ignore")}, + } + async def _async_delete_vlan(self, vlan_id: int) -> None: """Delete all NM connection profiles for *vlan_id* and emit connection_result.""" try: @@ -2248,41 +2529,40 @@ async def _reconnect_wifi_profile(self, ssid: str) -> None: ) return + found_ip = await self._wait_for_profile_ip(ssid, timeout=10.0) + if not found_ip: + logger.warning("Reconnect for '%s': IP not assigned within 10 s", ssid) + return + + logger.info("Reconnect complete for '%s': IP=%s", ssid, found_ip) + try: + self._invalidate_saved_cache() + self.saved_networks_loaded.emit(await self._get_saved_networks_impl()) + except Exception as cache_err: + logger.debug("Cache refresh after reconnect failed: %s", cache_err) + + async def _wait_for_profile_ip(self, ssid: str, timeout: float) -> str: + """Poll for an IPv4 address on *ssid*, returning it or "" once *timeout* elapses.""" loop = asyncio.get_running_loop() - deadline = loop.time() + 10.0 + deadline = loop.time() + timeout while loop.time() < deadline: await asyncio.sleep(1.0) - found_ip: str = "" try: current = await self._get_current_ssid() - if current and current.lower() == ssid.lower(): - found_ip = await self._get_current_ip() or "" - if not found_ip: - found_ip = self._get_ip_os_fallback("wlan0") or "" + if not current or current.lower() != ssid.lower(): + continue + found_ip = await self._get_current_ip() or "" + if not found_ip: + found_ip = ( + self._get_ip_os_fallback(self._get_wifi_iface_name()) or "" + ) + if found_ip: + return found_ip except Exception as exc: logger.debug( "IP address lookup during connection wait ignored: %s", exc ) - - if found_ip: - logger.info( - "Reconnect complete for '%s': IP=%s", - ssid, - found_ip, - ) - try: - self._invalidate_saved_cache() - self.saved_networks_loaded.emit( - await self._get_saved_networks_impl() - ) - except Exception as cache_err: - logger.debug( - "Cache refresh after reconnect failed: %s", - cache_err, - ) - return - - logger.warning("Reconnect for '%s': IP not assigned within 10 s", ssid) + return "" async def _async_update_wifi_static_ip( self, @@ -2294,8 +2574,18 @@ async def _async_update_wifi_static_ip( dns2: str, ) -> None: """Apply a static IPv4 configuration to a saved Wi-Fi profile and reconnect.""" + logger.info( + "static_ip: '%s' ip=%s mask=%s gw=%s dns=%s,%s", + ssid, + ip_address, + subnet_mask, + gateway, + dns1, + dns2, + ) conn_path = await self._get_connection_path(ssid) if not conn_path: + logger.warning("static_ip: '%s' has no saved profile", ssid) self.error_occurred.emit("wifi_static_ip", f"'{ssid}' not found") return try: @@ -2345,14 +2635,21 @@ async def _async_update_wifi_static_ip( async def _async_reset_wifi_to_dhcp(self, ssid: str) -> None: """Reset a saved Wi-Fi profile's IPv4 settings to DHCP and reconnect.""" + logger.info("reset_dhcp: '%s'", ssid) conn_path = await self._get_connection_path(ssid) if not conn_path: + logger.warning("reset_dhcp: '%s' has no saved profile", ssid) self.error_occurred.emit("wifi_dhcp", f"'{ssid}' not found") return try: cs = self._conn_settings(conn_path) props = await cs.get_settings() await self._merge_wifi_secrets(cs, props) + logger.info( + "reset_dhcp: '%s' was method=%s", + ssid, + self._setting(props, "ipv4", "method"), + ) props["ipv4"] = {"method": ("s", "auto")} await cs.update(props) self._invalidate_saved_cache() @@ -2439,15 +2736,7 @@ async def _async_create_and_activate_hotspot( "proceeding with hotspot activation anyway" ) - ethernet_was_active = await self._is_ethernet_connected() - if ethernet_was_active: - try: - await self._async_disconnect_ethernet() - except Exception as exc: - logger.debug("Pre-hotspot ethernet disconnect ignored: %s", exc) - # Brief pause to let eth0 finish deactivating before NM - # processes the hotspot activation request. - await asyncio.sleep(1.0) + # eth0 does not conflict with an AP; dropping it stranded the machine. if self._primary_wifi_path: try: await self._wifi().disconnect() @@ -2455,11 +2744,7 @@ async def _async_create_and_activate_hotspot( logger.debug("Pre-hotspot Wi-Fi disconnect ignored: %s", exc) await self._delete_all_ap_mode_connections() - # Also delete by connection id in case a non-AP profile shares the - # hotspot name (e.g. a leftover infrastructure profile named the - # same as the SSID). _delete_all_ap_mode_connections already caught - # all AP-mode profiles, so this second list_connections call is a - # narrow safety net. + # Safety net: a non-AP profile may still hold the hotspot name. await self._delete_connections_by_id(config_ssid) conn_props: dict[str, object] = { @@ -2492,8 +2777,7 @@ async def _async_create_and_activate_hotspot( "psk": ("s", config_pwd), "pmf": ("u", 0), } - # AP mode is always WPA2-PSK; WPA3-SAE in AP mode requires driver - # support not guaranteed on the target hardware. + # AP mode is always WPA2-PSK; SAE needs driver support we cannot assume. config_sec = HotspotSecurity.WPA2_PSK.value self._hotspot_config.security = config_sec @@ -2591,6 +2875,12 @@ async def _async_update_hotspot_config( async def _async_toggle_hotspot(self, enable: bool) -> None: """Enable or disable the hotspot, cleaning up profiles and Wi-Fi radio state.""" + logger.info( + "toggle_hotspot(%s): active=%s ssid='%s'", + enable, + self._is_hotspot_active, + self._hotspot_config.ssid, + ) try: if enable: await self._async_create_and_activate_hotspot( @@ -2698,108 +2988,6 @@ async def _delete_connections_where( logger.error("Cleanup for '%s' failed: %s", label, exc) return deleted - async def _find_ap_props(self, ssid: str) -> dict[str, object] | None: - """Return the scanned AP properties for *ssid*, or None if it is not visible.""" - ap_paths = await self._wifi().get_all_access_points() - for _ap_path, props in await self._gather_ap_properties(ap_paths): - if self._decode_ssid(props.get("ssid", b"")) == ssid: - return props - return None - - def _build_connection_properties( - self, - ssid: str, - password: str, - interface: str, - priority: int, - ap_props: dict[str, object], - ) -> dict[str, object] | None: - """Build NM connection property dict for *ssid* from its AP capability flags. - - Returns None if the security type is unsupported (e.g. WPA-EAP). - Handles OPEN, WPA-PSK, WPA2-PSK, and WPA3-SAE (including SAE-transition). - """ - flags = int(ap_props.get("flags", 0)) - wpa_flags = int(ap_props.get("wpa_flags", 0)) - rsn_flags = int(ap_props.get("rsn_flags", 0)) - - props: dict[str, object] = { - "connection": { - "id": ("s", ssid), - "uuid": ("s", str(uuid4())), - "type": ("s", "802-11-wireless"), - "interface-name": ("s", interface), - "autoconnect": ("b", True), - "autoconnect-priority": ("i", priority), - }, - "802-11-wireless": { - "mode": ("s", "infrastructure"), - "ssid": ("ay", ssid.encode("utf-8")), - }, - "ipv4": { - "method": ("s", "auto"), - "route-metric": ("i", 200), - }, - "ipv6": {"method": ("s", "auto")}, - } - - if (flags & 1) == 0: - return props - - props["802-11-wireless"]["security"] = ( - "s", - "802-11-wireless-security", - ) - security = self._determine_security_type(flags, wpa_flags, rsn_flags) - - if not is_connectable_security(security): - logger.warning( - "Rejecting connection to '%s': unsupported security %s", - ssid, - security.value, - ) - return None - - if security == SecurityType.WPA3_SAE: - has_psk = bool((rsn_flags & 0x100) or wpa_flags) - if has_psk: - logger.debug( - "SAE transition for '%s': using wpa-psk + PMF optional", - ssid, - ) - props["802-11-wireless-security"] = { - "key-mgmt": ("s", "wpa-psk"), - "auth-alg": ("s", "open"), - "psk": ("s", password), - "pmf": ("u", 2), # OPTIONAL: required for SAE-transition APs - } - else: - logger.debug("Pure SAE detected for '%s'", ssid) - props["802-11-wireless-security"] = { - "key-mgmt": ("s", "sae"), - "auth-alg": ("s", "open"), - "psk": ("s", password), - "pmf": ("u", 3), # REQUIRED: mandatory for pure WPA3-SAE - } - elif security in ( - SecurityType.WPA2_PSK, - SecurityType.WPA_PSK, - ): - props["802-11-wireless-security"] = { - "key-mgmt": ("s", "wpa-psk"), - "auth-alg": ("s", "open"), - "psk": ("s", password), - } - else: - logger.warning( - "Unsupported security type '%s' for '%s'", - security.value, - ssid, - ) - return None - - return props - async def _delete_all_connections_by_id(self, conn_id: str) -> int: """Delete every NM connection profile whose id exactly matches *conn_id*.""" return await self._delete_connections_where( @@ -2815,15 +3003,17 @@ async def _delete_all_ap_mode_connections(self) -> int: auto-activate them on the next boot. """ - def is_ap_mode(s: dict) -> bool: - """Check if this is a Wi-Fi connection in AP mode.""" - conn_type = self._setting(s, "connection", "type") - if conn_type != "802-11-wireless": - return False - mode = self._setting(s, "802-11-wireless", "mode") - return mode == "ap" + def is_ap_profile(s: dict) -> bool: + """True when the settings dict describes a Wi-Fi profile in AP mode.""" + return ( + self._setting(s, "connection", "type") == "802-11-wireless" + and self._setting(s, "802-11-wireless", "mode") == "ap" + ) - return await self._delete_connections_where(is_ap_mode, "ap-mode connections") + deleted = await self._delete_connections_where(is_ap_profile, "stale AP mode") + if deleted: + self._invalidate_saved_cache() + return deleted async def _delete_connections_by_id(self, ssid: str) -> int: """Delete every NM connection profile whose id matches *ssid* (case-insensitive).""" @@ -2897,5 +3087,11 @@ def _mask_to_prefix(mask_str: str) -> int: prefix = int(stripped) if 0 <= prefix <= 32: return prefix + logger.warning("mask_to_prefix: CIDR prefix out of range: %s", stripped) raise ValueError(f"CIDR prefix out of range: {prefix}") - return bin(int(ipaddress.IPv4Address(stripped))).count("1") + # IPv4Network rejects non-contiguous masks such as 255.0.255.0. + try: + return ipaddress.IPv4Network(f"0.0.0.0/{stripped}").prefixlen + except ValueError as exc: + logger.warning("mask_to_prefix: rejected mask '%s': %s", stripped, exc) + raise ValueError(f"Invalid subnet mask: {mask_str}") from exc diff --git a/BlocksScreen/lib/panels/networkWindow.py b/BlocksScreen/lib/panels/networkWindow.py index e22d331d..2bcd1592 100644 --- a/BlocksScreen/lib/panels/networkWindow.py +++ b/BlocksScreen/lib/panels/networkWindow.py @@ -37,7 +37,7 @@ from lib.utils.icon_button import IconButton from lib.utils.list_model import EntryDelegate, EntryListModel, ListItem from PyQt6 import QtCore, QtGui, QtWidgets -from PyQt6.QtCore import QTimer +from PyQt6.QtCore import QTimer, pyqtSlot logger = logging.getLogger(__name__) @@ -99,7 +99,7 @@ def __init__( self, parent: QtWidgets.QWidget | None = None, *, - placeholder: str = "0.0.0.0", # nosec B104 — UI placeholder text, not a socket bind + placeholder: str = "0.0.0.0", # nosec B104: UI placeholder text, not a socket bind ) -> None: """Initialise the IP-address input field with regex validation and optional placeholder.""" super().__init__(parent) @@ -120,10 +120,7 @@ def is_valid_mask(self) -> bool: """Return ``True`` when the current text is a valid subnet mask or CIDR prefix.""" txt = self.text().strip() if txt.isdigit(): - n = int(txt) - if 0 <= n <= 32: - return True - return False + return 0 <= int(txt) <= 32 try: _ipaddress.IPv4Network(f"0.0.0.0/{txt}", strict=False) @@ -131,6 +128,7 @@ def is_valid_mask(self) -> bool: except ValueError: return False + @pyqtSlot(str) def _on_text_changed(self, text: str) -> None: """Update the field border colour in real-time as the user types.""" if not text: @@ -144,6 +142,45 @@ def _on_text_changed(self, text: str) -> None: self.update() +# No SSID and no hotspot is enough; the radio itself may still be settling. +def _link_down(_win: "NetworkControlWindow", state: NetworkState) -> bool: + return not state.current_ssid and not state.hotspot_enabled + + +def _joined_target(win: "NetworkControlWindow", state: NetworkState) -> bool: + return bool( + win._target_ssid + and state.current_ssid == win._target_ssid + and state.current_ip + and state.connectivity in (ConnectivityState.FULL, ConnectivityState.LIMITED) + ) + + +# Success messages the state stream already reflects; showing a popup would be noise. +_SILENT_SUCCESS = ("wi-fi disabled", "disconnected", "wi-fi enabled") +# NM errors seen while the wired link tears down; a retry usually succeeds. +_TRANSIENT_MISMATCH = ( + "not compatible with device", + "mismatching interface", + "not available because profile", +) + + +# Pending operation -> (outcome once settled, predicate telling it has settled). +_PENDING_OP_RULES = { + PendingOperation.WIFI_OFF: (False, _link_down), + PendingOperation.HOTSPOT_OFF: (False, _link_down), + PendingOperation.HOTSPOT_ON: ( + True, + lambda w, s: bool(s.hotspot_enabled and s.current_ssid and s.current_ip), + ), + PendingOperation.WIFI_ON: (True, _joined_target), + PendingOperation.CONNECT: (True, _joined_target), + PendingOperation.ETHERNET_ON: (True, lambda w, s: bool(s.ethernet_connected)), + PendingOperation.ETHERNET_OFF: (False, lambda w, s: not s.ethernet_connected), +} + + class NetworkControlWindow(QtWidgets.QStackedWidget): """Stacked-widget UI for all network control pages (Wi-Fi, Ethernet, VLAN, Hotspot). @@ -183,14 +220,16 @@ def _init_instance_variables(self) -> None: self._current_network_is_hidden = False self._is_connecting = False self._target_ssid: str | None = None - self._was_ethernet_connected: bool = False self._initial_priority: ConnectionPriority = ConnectionPriority.MEDIUM + self._initial_password: str = "" + self._password_ssid: str = "" # guards the async psk reply against page changes self._pending_operation: PendingOperation = PendingOperation.NONE self._pending_expected_ip: str = ( "" # IP to wait for before clearing WIFI_STATIC_IP loading ) self._cached_scan_networks: list[NetworkInfo] = [] self._last_active_signal_bars: int = -1 + self._last_state_summary: tuple | None = None self._active_signal: int = 0 # Key = SSID, value = (signal_bars, status_label, ListItem). self._item_cache: dict[str, tuple[int, str, ListItem]] = {} @@ -232,12 +271,14 @@ def _init_network_manager(self) -> None: self._nm.hotspot_config_updated.connect(self._on_hotspot_config_updated) + self._nm.network_password_loaded.connect(self._on_network_password_loaded) + self._prefill_ip_from_os() def _prefill_ip_from_os(self) -> None: """Read the current IP via SIOCGIFADDR ioctl and show it immediately. - Bypasses NetworkManager D-Bus entirely — runs on the main thread, + Bypasses NetworkManager D-Bus entirely: runs on the main thread, costs a single syscall, and completes in microseconds. Called once during init so the user never sees "IP: --" if a connection was already active before the UI launched. @@ -257,10 +298,9 @@ def _prefill_ip_from_os(self) -> None: except OSError: continue - @QtCore.pyqtSlot() + @pyqtSlot() def _on_reconnect_complete(self) -> None: """Navigate back to the main panel after a static-IP or DHCP-reset operation.""" - logger.debug("reconnect_complete received — navigating to main_network_page") self.setCurrentIndex(self.indexOf(self.main_network_page)) def _init_timers(self) -> None: @@ -279,16 +319,19 @@ def _init_model_view(self) -> None: self._entry_delegate.item_selected.connect(self._on_ssid_item_clicked) self._configure_list_view_palette() - @QtCore.pyqtSlot(NetworkState) + @pyqtSlot(NetworkState) def _on_network_state_changed(self, state: NetworkState) -> None: """React to a NetworkState update: sync toggles, populate header and connection info.""" - logger.debug( - "Network state: %s, SSID: %s, IP: %s, eth: %s", + # The poll re-emits unchanged state; log transitions only to keep logs usable. + summary = ( state.connectivity.name, state.current_ssid, state.current_ip, state.ethernet_connected, ) + if summary != self._last_state_summary: + self._last_state_summary = summary + logger.info("Network state: %s, SSID: %s, IP: %s, eth: %s", *summary) if ( state.current_ssid @@ -303,126 +346,28 @@ def _on_network_state_changed(self, state: NetworkState) -> None: self._handle_first_run(state) self._emit_status_icon(state) self._is_first_run = False - self._was_ethernet_connected = state.ethernet_connected - return - - # Cable just plugged in while Wi-Fi is active -> disable Wi-Fi - if ( - state.ethernet_connected - and not self._was_ethernet_connected - and state.wifi_enabled - and not self._is_connecting - ): - logger.info("Ethernet connected — turning off Wi-Fi") - self._was_ethernet_connected = True - wifi_btn = self.wifi_button.toggle_button - hotspot_btn = self.hotspot_button.toggle_button - with QtCore.QSignalBlocker(wifi_btn): - wifi_btn.state = wifi_btn.State.OFF - with QtCore.QSignalBlocker(hotspot_btn): - hotspot_btn.state = hotspot_btn.State.OFF - self._nm.set_wifi_enabled(False) - self._sync_ethernet_panel(state) - self._emit_status_icon(state) return - self._was_ethernet_connected = state.ethernet_connected + # Exclusivity applies to user toggles only; a cable never kills the radio. - # Ethernet panel visibility is pure hardware state (carrier + - # connection) and must update even while a loading operation is - # in-flight. + # Pure hardware state, so it updates even mid-operation. self._sync_ethernet_panel(state) - - # Sync toggle states (skipped when _is_connecting) self._sync_toggle_states(state) if self._is_connecting: - # OFF operations: complete when radio off + no connection - if self._pending_operation in ( - PendingOperation.WIFI_OFF, - PendingOperation.HOTSPOT_OFF, - ): - if ( - not state.wifi_enabled - and not state.hotspot_enabled - and not state.current_ssid - ): - self._clear_loading() - self._display_disconnected_state() - self._emit_status_icon(state) - return - # Also catch partial-off (wifi still disabling, no ssid) - if not state.current_ssid and not state.hotspot_enabled: - self._clear_loading() - self._display_disconnected_state() - self._emit_status_icon(state) - return - # Still transitioning — keep loading visible - return - - # Hotspot ON: complete when hotspot_enabled + SSID + IP - if self._pending_operation == PendingOperation.HOTSPOT_ON: - if state.hotspot_enabled and state.current_ssid and state.current_ip: - self._clear_loading() - self._display_connected_state(state) - self._emit_status_icon(state) - return - # Still waiting for hotspot to fully come up - return - - if self._pending_operation in ( - PendingOperation.WIFI_ON, - PendingOperation.CONNECT, - ): - if self._target_ssid and state.current_ssid == self._target_ssid: - if state.current_ip and state.connectivity in ( - ConnectivityState.FULL, - ConnectivityState.LIMITED, - ): - self._clear_loading() - self._display_connected_state(state) - self._emit_status_icon(state) - return - return - - if self._pending_operation == PendingOperation.ETHERNET_ON: - if state.ethernet_connected: - self._clear_loading() - self._sync_ethernet_panel(state) - self._display_connected_state(state) - self._emit_status_icon(state) - return - return - - if self._pending_operation == PendingOperation.ETHERNET_OFF: - if not state.ethernet_connected: - self._clear_loading() - self._sync_ethernet_panel(state) - self._display_disconnected_state() - self._emit_status_icon(state) - return - return - - # Wi-Fi static IP / DHCP reset: complete when we have the right IP. - if self._pending_operation == PendingOperation.WIFI_STATIC_IP: - ip = state.current_ip or "" - expected = self._pending_expected_ip - ip_matches = ip and (not expected or ip == expected) - if ip_matches: - self._pending_expected_ip = "" - self._clear_loading() - self._display_connected_state(state) - self._emit_status_icon(state) - return - # IP not yet correct — keep loading visible - return - + outcome = self._pending_op_outcome(state) + if outcome is None: + return # Still transitioning: keep the loading screen up. + self._clear_loading() + if outcome: + self._display_connected_state(state) + else: + self._display_disconnected_state() + self._emit_status_icon(state) return # Normal (not connecting) display updates. - if state.ethernet_connected: - self._display_connected_state(state) - elif ( + if state.ethernet_connected or ( state.current_ssid and state.current_ip and state.connectivity @@ -440,7 +385,22 @@ def _on_network_state_changed(self, state: NetworkState) -> None: self._emit_status_icon(state) self._sync_active_network_list_icon(state) - @QtCore.pyqtSlot(list) + def _pending_op_outcome(self, state: NetworkState) -> bool | None: + """Return the settled connected-ness of the pending operation, None while in flight.""" + op = self._pending_operation + + if op == PendingOperation.WIFI_STATIC_IP: + # Settles once an IP arrives that matches the one we asked for. + ip = state.current_ip or "" + if not ip or self._pending_expected_ip not in ("", ip): + return None + self._pending_expected_ip = "" + return True + + outcome, settled = _PENDING_OP_RULES.get(op, (None, None)) + return outcome if settled and settled(self, state) else None + + @pyqtSlot(list) def _on_scan_complete(self, networks: list[NetworkInfo]) -> None: """Receive scan results, filter/sort them, and rebuild the SSID list view. @@ -456,14 +416,11 @@ def _on_scan_complete(self, networks: list[NetworkInfo]) -> None: current_ssid = self._nm.current_ssid if current_ssid: - # Stamp the connected AP as ACTIVE so the list is correct on first - # render even when the scan ran before the connection fully settled. + # Stamp the connected AP ACTIVE; the scan may predate the connection. filtered = [ - ( - replace(net, network_status=NetworkStatus.ACTIVE) - if net.ssid == current_ssid - else net - ) + replace(net, network_status=NetworkStatus.ACTIVE) + if net.ssid == current_ssid + else net for net in filtered ] active = next((n for n in filtered if n.ssid == current_ssid), None) @@ -482,107 +439,90 @@ def _on_scan_complete(self, networks: list[NetworkInfo]) -> None: state = self._nm.current_state self._emit_status_icon(state) - @QtCore.pyqtSlot(list) + @pyqtSlot(list) def _on_saved_networks_loaded(self, networks: list[SavedNetwork]) -> None: """Receive saved-network data and update the priority spinbox for the active SSID.""" logger.debug("Loaded %d saved networks", len(networks)) - @QtCore.pyqtSlot(ConnectionResult) + @pyqtSlot(ConnectionResult) def _on_operation_complete(self, result: ConnectionResult) -> None: """Handle network operation completion.""" logger.debug("Operation: success=%s, msg=%s", result.success, result.message) - if result.success: - msg_lower = result.message.lower() - if "deleted" in msg_lower: - ssid_deleted = ( - self._target_ssid - ) # capture before _clear_loading wipes it - self._show_info_popup(result.message) - self._clear_loading() - self._display_wifi_on_no_connection() - self.setCurrentIndex(self.indexOf(self.main_network_page)) - if ssid_deleted: - self._patch_cached_network_status( - ssid_deleted, NetworkStatus.DISCOVERED - ) - elif "hotspot" in msg_lower and "activated" in msg_lower: - self._show_hotspot_qr( - self._nm.hotspot_ssid, - self._nm.hotspot_password, - self._nm.hotspot_security, - ) - elif "hotspot disabled" in msg_lower: - self.qrcode_img.clearPixmap() - self.qrcode_img.setText("Hotspot not active") - elif "wi-fi disabled" in msg_lower: - pass - elif "config updated" in msg_lower: - self._show_info_popup(result.message) - elif any( - skip in msg_lower - for skip in ( - "added", - "connecting", - "disconnected", - "wi-fi enabled", - ) - ): - if ( - ("added" in msg_lower or "connecting" in msg_lower) - and self._target_ssid - and not self._current_network_is_hidden - ): - # Hidden networks are not in the scan cache; the next scan - # will surface them once NM reports them as saved/active. - self._patch_cached_network_status( - self._target_ssid, NetworkStatus.SAVED - ) - elif self._pending_operation == PendingOperation.WIFI_STATIC_IP: - # Loading cleared by state machine (IP appears) or reconnect_complete. - # No popup — the updated IP in the header is the confirmation. - pass - else: - self._show_info_popup(result.message) + self._handle_operation_success(result) else: - msg_lower = result.message.lower() - - # Duplicate VLAN: clear loading and show the reason. - if result.error_code == "duplicate_vlan": - self._clear_loading() - self._show_error_popup(result.message) + self._handle_operation_failure(result) + + def _handle_operation_success(self, result: ConnectionResult) -> None: + """Route a successful operation message to its UI reaction.""" + msg = result.message.lower() + for keys, handler in ( + (("deleted",), self._on_network_deleted), + (("hotspot", "activated"), self._on_hotspot_activated), + (("hotspot disabled",), self._on_hotspot_disabled), + (("config updated",), self._show_info_popup), + (("added",), self._on_network_saved), + (("connecting",), self._on_network_saved), + ): + if all(k in msg for k in keys): + handler(result.message) return - # When switching from ethernet to wifi, NM may report a - # device-mismatch error because the wired profile hasn't - # fully deactivated yet. Retry the connection instead of - # showing a confusing popup to the user. - is_transient_mismatch = ( - "not compatible with device" in msg_lower - or "mismatching interface" in msg_lower - or "not available because profile" in msg_lower - ) - if ( - is_transient_mismatch - and self._pending_operation + # Silent successes: the state stream is the confirmation, no popup. + if not any(k in msg for k in _SILENT_SUCCESS) and ( + self._pending_operation != PendingOperation.WIFI_STATIC_IP + ): + self._show_info_popup(result.message) + + def _on_network_deleted(self, message: str) -> None: + """Return to the main page after a saved network was removed.""" + ssid = self._target_ssid # captured before _clear_loading wipes it + self._show_info_popup(message) + self._clear_loading() + self._display_wifi_on_no_connection() + self.setCurrentIndex(self.indexOf(self.main_network_page)) + if ssid: + self._patch_cached_network_status(ssid, NetworkStatus.DISCOVERED) + + def _on_hotspot_activated(self, _message: str) -> None: + """Show the QR code for the freshly activated hotspot.""" + self._show_hotspot_qr( + self._nm.hotspot_ssid, + self._nm.hotspot_password, + self._nm.hotspot_security, + ) + + def _on_hotspot_disabled(self, _message: str) -> None: + """Clear the hotspot QR code.""" + self.qrcode_img.clearPixmap() + self.qrcode_img.setText("Hotspot not active") + + def _on_network_saved(self, _message: str) -> None: + """Mark the target SSID as saved in the list cache.""" + # Hidden networks reach the cache only on the next scan. + if self._target_ssid and not self._current_network_is_hidden: + self._patch_cached_network_status(self._target_ssid, NetworkStatus.SAVED) + + def _handle_operation_failure(self, result: ConnectionResult) -> None: + """Show the failure reason, retrying once on a transient device mismatch.""" + msg = result.message.lower() + if result.error_code != "duplicate_vlan" and self._target_ssid: + # Ethernet-to-Wi-Fi races the wired teardown; retry instead of erroring. + if any(k in msg for k in _TRANSIENT_MISMATCH) and ( + self._pending_operation in (PendingOperation.WIFI_ON, PendingOperation.CONNECT) - and self._target_ssid ): - logger.debug( - "Transient NM device-mismatch during wifi activation " - "— retrying in 2 s: %s", - result.message, - ) + logger.debug("Transient NM device-mismatch, retry in 2 s: %s", msg) ssid = self._target_ssid QTimer.singleShot( 2000, lambda _ssid=ssid: self._nm.connect_network(_ssid) ) return # Keep loading visible; state machine handles completion - self._clear_loading() - self._show_error_popup(result.message) + self._clear_loading() + self._show_error_popup(result.message) - @QtCore.pyqtSlot(str, str) + @pyqtSlot(str, str) def _on_network_error(self, operation: str, message: str) -> None: """Log network errors and surface critical failures in the info box.""" logger.error("Network error [%s]: %s", operation, message) @@ -623,7 +563,7 @@ def _emit_status_icon(self, state: NetworkState) -> None: WifiIconKey.from_signal(self._active_signal, False) ) else: - # Disconnected / no connection — 0-bar unprotected + # Disconnected / no connection: 0-bar unprotected self.update_wifi_icon.emit(WifiIconKey.from_bars(0, False)) def _sync_active_network_list_icon(self, state: NetworkState) -> None: @@ -643,10 +583,7 @@ def _sync_active_network_list_icon(self, state: NetworkState) -> None: new_bars = signal_to_bars(self._active_signal) - # Also check whether the cached status already reflects ACTIVE. - # If not, we must rebuild even when bars haven't changed (e.g. the - # scan ran before the connection was fully established and marked the - # network SAVED instead of ACTIVE). + # Rebuild on a stale SAVED status too, not only on a bar change. cached_active = next( (n for n in self._cached_scan_networks if n.ssid == state.current_ssid), None, @@ -654,23 +591,19 @@ def _sync_active_network_list_icon(self, state: NetworkState) -> None: status_needs_update = cached_active is not None and not cached_active.is_active if new_bars == self._last_active_signal_bars and not status_needs_update: - return # No visual change — skip the rebuild + return # No visual change: skip the rebuild - # Invalidate cache for the active SSID so _get_or_create_item - # creates a fresh ListItem with the updated signal icon and status. + # Drop the cached item so it is rebuilt with the new icon and status. self._item_cache.pop(state.current_ssid, None) - # Update the cached entry with the authoritative signal and status updated = [ - ( - replace( - net, - signal_strength=self._active_signal, - network_status=NetworkStatus.ACTIVE, - ) - if net.ssid == state.current_ssid - else net + replace( + net, + signal_strength=self._active_signal, + network_status=NetworkStatus.ACTIVE, ) + if net.ssid == state.current_ssid + else net for net in self._cached_scan_networks ] @@ -691,8 +624,7 @@ def _handle_first_run(self, state: NetworkState) -> None: hotspot_on = False if state.ethernet_connected: - if state.wifi_enabled: - self._nm.set_wifi_enabled(False) + # Display only: never force the radio off here, it is the recovery path. self._display_connected_state(state) elif state.connectivity == ConnectivityState.FULL and state.current_ssid: wifi_on = True @@ -735,6 +667,7 @@ def _sync_toggle_states(self, state: NetworkState) -> None: wifi_on = False hotspot_on = False + # One link at a time: the cable wins the display, then hotspot, then Wi-Fi. if state.ethernet_connected: pass elif state.hotspot_enabled: @@ -749,6 +682,15 @@ def _sync_toggle_states(self, state: NetworkState) -> None: hotspot_btn.State.ON if hotspot_on else hotspot_btn.State.OFF ) + def _claim_link(self, winner) -> None: + """Force the other two link toggles OFF; only one link may be on at a time.""" + for btn in (self.wifi_button, self.hotspot_button, self.ethernet_button): + if btn is winner: + continue + toggle = btn.toggle_button + with QtCore.QSignalBlocker(toggle): + toggle.state = toggle.State.OFF + def _sync_ethernet_panel(self, state: NetworkState) -> None: """Show/hide the ethernet panel and sync its toggle state. @@ -769,7 +711,7 @@ def _sync_ethernet_panel(self, state: NetworkState) -> None: def _display_connected_state(self, state: NetworkState) -> None: """Display connected network information. - Ethernet always takes display priority — if ``ethernet_connected`` + Ethernet always takes display priority: if ``ethernet_connected`` is True we show "Ethernet" even if a Wi-Fi SSID is still lingering (e.g. during the brief overlap before NM finishes disabling wifi). """ @@ -793,16 +735,13 @@ def _display_connected_state(self, state: NetworkState) -> None: self.netlist_vlans_combo.blockSignals(True) self.netlist_vlans_combo.clear() self.netlist_vlans_combo.addItem( - f"Ethernet — {state.current_ip or '--'}", + f"Ethernet: {state.current_ip or '--'}", state.current_ip or "", ) for v in state.active_vlans: - if v.is_dhcp: - ip_label = v.ip_address or "DHCP" - else: - ip_label = v.ip_address or "--" + ip_label = v.ip_address or ("DHCP" if v.is_dhcp else "--") self.netlist_vlans_combo.addItem( - f"VLAN {v.vlan_id} — {ip_label}", + f"VLAN {v.vlan_id}: {ip_label}", v.ip_address or "", ) self.netlist_vlans_combo.setCurrentIndex(0) @@ -833,7 +772,7 @@ def _display_connected_state(self, state: NetworkState) -> None: self.update() def _display_disconnected_state(self) -> None: - """Display disconnected state — both toggles OFF.""" + """Display disconnected state: both toggles OFF.""" self._hide_all_info_elements() self.mn_info_box.setVisible(True) @@ -924,52 +863,53 @@ def _clear_loading(self) -> None: """Hide the loading widget and re-enable the full UI.""" self._set_loading_state(False) + @pyqtSlot() def _handle_load_timeout(self) -> None: """Hide the loading widget if it is still visible after the timeout fires.""" if not self.loadingwidget.isVisible(): return state = self._nm.current_state - if ( - self._pending_operation == PendingOperation.HOTSPOT_ON - and state.hotspot_enabled - and state.current_ssid - ): - self._clear_loading() - self._display_connected_state(state) + if self._settle_late_success(state): return - if ( - self._pending_operation - in (PendingOperation.WIFI_ON, PendingOperation.CONNECT) - and self._target_ssid - ): - if state.current_ssid == self._target_ssid and state.current_ip: + self._display_timeout_failure() + + def _settle_late_success(self, state: NetworkState) -> bool: + """Accept an operation that completed just as the timer fired; True if handled.""" + op = self._pending_operation + + if op == PendingOperation.HOTSPOT_ON and state.hotspot_enabled: + if state.current_ssid: self._clear_loading() self._display_connected_state(state) - return - if ( - self._pending_operation == PendingOperation.ETHERNET_ON - and state.ethernet_connected - ): + return True + elif op in (PendingOperation.WIFI_ON, PendingOperation.CONNECT): + if ( + self._target_ssid + and state.current_ssid == self._target_ssid + and state.current_ip + ): + self._clear_loading() + self._display_connected_state(state) + return True + elif op == PendingOperation.ETHERNET_ON and state.ethernet_connected: self._clear_loading() self._sync_ethernet_panel(state) self._display_connected_state(state) - return - - # Static IP / DHCP reset — if a state with an IP has arrived, accept it. - if self._pending_operation == PendingOperation.WIFI_STATIC_IP: - if state.current_ip: - self._clear_loading() - self._display_connected_state(state) - return - # No IP yet after timeout — clear loading and show whatever state we have. + return True + elif op == PendingOperation.WIFI_STATIC_IP: + # Never a failure: show whatever state arrived, with or without an IP. self._clear_loading() - if state.current_ssid: + if state.current_ip or state.current_ssid: self._display_connected_state(state) else: self._display_disconnected_state() - return + return True + + return False + def _display_timeout_failure(self) -> None: + """Show the timeout message matching the operation that failed and re-enable the UI.""" self._clear_loading() self._hide_all_info_elements() self._configure_info_box_centered() @@ -1008,15 +948,6 @@ def _configure_info_box_centered(self) -> None: self.mn_info_box.setWordWrap(True) self.mn_info_box.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter) - def _claim_link(self, winner) -> None: - """Force the other two link toggles OFF; only one link may be on at a time.""" - for btn in (self.wifi_button, self.hotspot_button, self.ethernet_button): - if btn is winner: - continue - toggle = btn.toggle_button - with QtCore.QSignalBlocker(toggle): - toggle.state = toggle.State.OFF - @QtCore.pyqtSlot(object, name="stateChange") def _on_toggle_state(self, new_state) -> None: """Route a toggle-button state change to the correct handler (Wi-Fi or hotspot).""" @@ -1033,8 +964,7 @@ def _on_toggle_state(self, new_state) -> None: elif sender_button is eth_btn: self._handle_ethernet_toggle(is_on) - # Both OFF state is now handled by _on_network_state_changed - # when the worker emits the disconnected state. + # The both-off case arrives via _on_network_state_changed. def _handle_wifi_toggle(self, is_on: bool) -> None: """Enable or disable Wi-Fi; turning it on drops the hotspot and the cable.""" @@ -1201,6 +1131,7 @@ def _show_hotspot_qr(self, ssid: str, password: str, security: str) -> None: self.qrcode_img.clearPixmap() self.qrcode_img.setText("QR error") + @pyqtSlot() def _on_ethernet_button_clicked(self) -> None: """Navigate to the ethernet/VLAN settings page when the ethernet button is clicked.""" if ( @@ -1211,6 +1142,7 @@ def _on_ethernet_button_clicked(self) -> None: return self.setCurrentIndex(self.indexOf(self.vlan_page)) + @pyqtSlot() def _on_vlan_apply(self) -> None: """Validate VLAN fields and call ``create_vlan_connection`` on the facade.""" vlan_id = self.vlan_id_spinbox.value() @@ -1252,31 +1184,36 @@ def _on_vlan_apply(self) -> None: ) self._nm.request_state_soon(delay_ms=3000) + @pyqtSlot() def _on_vlan_delete(self) -> None: """Read the VLAN ID from the spinbox and request deletion via the facade.""" vlan_id = self.vlan_id_spinbox.value() self._nm.delete_vlan_connection(vlan_id) self._show_warning_popup(f"VLAN {vlan_id} profile removed.") + @pyqtSlot(int) def _on_interface_combo_changed(self, index: int) -> None: """Swap the displayed IP when the user selects a different interface.""" ip = self.netlist_vlans_combo.itemData(index) if ip is not None: self.netlist_ip.setText(f"IP: {ip}" if ip else "IP: --") + @pyqtSlot() def _on_wifi_static_ip_clicked(self) -> None: """Navigate from saved details page to WiFi static IP page.""" ssid = self.snd_name.text() self.wifi_sip_title.setText(ssid) - self.wifi_sip_ip_field.clear() - self.wifi_sip_mask_field.clear() - self.wifi_sip_gateway_field.clear() - self.wifi_sip_dns1_field.clear() - self.wifi_sip_dns2_field.clear() - - # Enable "Reset to DHCP" only when the profile is currently using a - # static IP — if it is already DHCP there is nothing to reset. + saved = self._nm.get_saved_network(ssid) + dns = saved.dns_servers if saved else () + # Prefill with the profile's current static config so the mask is visible. + self.wifi_sip_ip_field.setText(saved.ip_address if saved else "") + self.wifi_sip_mask_field.setText(saved.netmask if saved else "") + self.wifi_sip_gateway_field.setText(saved.gateway if saved else "") + self.wifi_sip_dns1_field.setText(dns[0] if len(dns) > 0 else "") + self.wifi_sip_dns2_field.setText(dns[1] if len(dns) > 1 else "") + + # Nothing to reset when the profile is already on DHCP. is_dhcp = saved.is_dhcp if saved else True self.wifi_sip_dhcp_button.setEnabled(not is_dhcp) self.wifi_sip_dhcp_button.setToolTip( @@ -1285,12 +1222,13 @@ def _on_wifi_static_ip_clicked(self) -> None: self.setCurrentIndex(self.indexOf(self.wifi_static_ip_page)) + @pyqtSlot() def _on_wifi_static_ip_apply(self) -> None: """Validate static-IP fields and apply them to the current Wi-Fi connection. Mirrors the VLAN-creation UX: navigate to the main panel immediately, show the loading overlay, and clear it silently once ``reconnect_complete`` - fires (no popup — the updated IP appears in the panel header instead). + fires (no popup: the updated IP appears in the panel header instead). """ ssid = self.wifi_sip_title.text() ip_addr = self.wifi_sip_ip_field.text().strip() @@ -1323,10 +1261,11 @@ def _on_wifi_static_ip_apply(self) -> None: self._nm.update_wifi_static_ip(ssid, ip_addr, mask, gateway, dns1, dns2) self._nm.request_state_soon(delay_ms=3000) + @pyqtSlot() def _on_wifi_reset_dhcp(self) -> None: """Reset the current Wi-Fi connection back to DHCP via the facade. - Same loading-screen pattern as static IP — no popup on success. + Same loading-screen pattern as static IP: no popup on success. """ ssid = self.wifi_sip_title.text() self.setCurrentIndex(self.indexOf(self.main_network_page)) @@ -1343,7 +1282,7 @@ def _build_network_list_from_scan(self, networks: list[NetworkInfo]) -> None: Uses the model's built-in reconcile() with an item cache so that ListItems are only allocated for networks whose visual state actually changed (different signal bars or status label). - Unchanged items are reused from the cache — zero allocation. + Unchanged items are reused from the cache: zero allocation. """ self.listView.blockSignals(True) @@ -1398,8 +1337,8 @@ def _get_or_create_item(self, network: NetworkInfo) -> ListItem | None: unchanged, otherwise create a new one and update the cache. Visual state = (signal_bars, status_label). When both match - the cached entry, the existing ListItem is returned as-is — - no QPixmap lookup, no allocation. + the cached entry the existing ListItem is returned as-is, with + no QPixmap lookup and no allocation. """ if network.is_hidden or is_hidden_ssid(network.ssid): return None @@ -1537,8 +1476,21 @@ def _show_saved_network_page(self, network: NetworkInfo) -> None: if self.saved_connection_change_password_view.isChecked(): self.saved_connection_change_password_view.setChecked(False) + # Answered asynchronously by _on_network_password_loaded. + self._password_ssid = ssid + self._initial_password = "" # nosec B105 - empty default, not a credential + if not network.is_open: + self._nm.get_network_password(ssid) + saved = self._nm.get_saved_network(ssid) + if saved and not saved.is_dhcp and saved.ip_address: + self.saved_connection_address_info_label.setText( + f"{saved.ip_address}\n{saved.netmask or '--'}" + ) + else: + self.saved_connection_address_info_label.setText("DHCP") + if saved: self._set_priority_button(saved.priority) # Track initial values for change detection @@ -1546,9 +1498,7 @@ def _show_saved_network_page(self, network: NetworkInfo) -> None: else: self._initial_priority = ConnectionPriority.MEDIUM - # Signal strength — for the active network, use the unified - # _active_signal so the details page matches the main panel - # and header icon exactly. + # _active_signal keeps this page, the main panel and the icon in step. is_active = ssid == self._nm.current_ssid if is_active and self._active_signal > 0: signal_value = self._active_signal @@ -1595,10 +1545,6 @@ def _set_priority_button(self, priority: int | None) -> None: else: target = self.med_priority_btn - logger.debug( - "Setting priority button: priority=%r -> %s", priority, target.text() - ) - target.setChecked(True) self.high_priority_btn.update() @@ -1608,14 +1554,6 @@ def _set_priority_button(self, priority: int | None) -> None: def _get_selected_priority(self) -> ConnectionPriority: """Return the ``ConnectionPriority`` matching the currently selected radio button.""" checked = self.priority_btn_group.checkedButton() - logger.debug( - "Priority selection: checked=%s, h=%s m=%s l=%s", - checked.text() if checked else "None", - self.high_priority_btn.isChecked(), - self.med_priority_btn.isChecked(), - self.low_priority_btn.isChecked(), - ) - if checked is self.high_priority_btn: return ConnectionPriority.HIGH elif checked is self.low_priority_btn: @@ -1658,6 +1596,7 @@ def _add_network(self) -> None: self.add_network_validation_button.setEnabled(True) + @pyqtSlot() def _on_activate_network(self) -> None: """Activate the network shown on the saved-connection page.""" ssid = self.saved_connection_network_name.text() @@ -1676,12 +1615,14 @@ def _on_activate_network(self) -> None: self.setCurrentIndex(self.indexOf(self.main_network_page)) self._nm.connect_network(ssid) + @pyqtSlot() def _on_delete_network(self) -> None: """Delete the profile shown on the saved-connection page and navigate back.""" ssid = self.saved_connection_network_name.text() self._target_ssid = ssid self._nm.delete_network(ssid) + @pyqtSlot() def _on_save_network_details(self) -> None: """Save network settings changes (password / priority). @@ -1692,7 +1633,7 @@ def _on_save_network_details(self) -> None: password = self.saved_connection_change_password_field.text() priority = self._get_selected_priority() - password_changed = bool(password) + password_changed = password != self._initial_password priority_changed = priority != self._initial_priority if not password_changed and not priority_changed: @@ -1701,17 +1642,27 @@ def _on_save_network_details(self) -> None: self._nm.update_network( ssid, - password=password or "", + password=password if password_changed else "", priority=priority.value, ) self._nm.load_saved_networks() - # Update tracked baseline so a second press won't re-save + # Update tracked baselines so a second press won't re-save self._initial_priority = priority + self._initial_password = password - self.saved_connection_change_password_field.clear() + @QtCore.pyqtSlot(str, str) + def _on_network_password_loaded(self, ssid: str, password: str) -> None: + """Prefill the change-password field with the profile's stored psk.""" + if ssid != self._password_ssid: + return + self._initial_password = password + self.saved_connection_change_password_field.setText(password) + if password: + self.saved_connection_change_password_field.setPlaceholderText("") + @pyqtSlot() def _on_hidden_network_connect(self) -> None: """Connect to hidden network - non-blocking.""" ssid = self.hidden_network_ssid_field.text().strip() @@ -2592,6 +2543,39 @@ def _setup_saved_connection_page(self) -> None: status_layout.addWidget(self.sn_info) frame_inner_layout.addLayout(status_layout) + + self.line_6 = QtWidgets.QFrame(parent=self.frame) + self.line_6.setFrameShape(QtWidgets.QFrame.Shape.HLine) + self.line_6.setFrameShadow(QtWidgets.QFrame.Shadow.Sunken) + + frame_inner_layout.addWidget(self.line_6) + + address_layout = QtWidgets.QHBoxLayout() + + self.netlist_address_label = QtWidgets.QLabel(parent=self.frame) + self.netlist_address_label.setPalette(self._create_white_palette()) + font = QtGui.QFont() + font.setPointSize(15) + self.netlist_address_label.setFont(font) + self.netlist_address_label.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter) + self.netlist_address_label.setText("IP /\nMask") + + address_layout.addWidget(self.netlist_address_label) + + self.saved_connection_address_info_label = QtWidgets.QLabel(parent=self.frame) + self.saved_connection_address_info_label.setMinimumSize(QtCore.QSize(250, 0)) + font = QtGui.QFont() + font.setPointSize(11) + self.saved_connection_address_info_label.setFont(font) + self.saved_connection_address_info_label.setStyleSheet( + "color: rgb(255, 255, 255);" + ) + self.saved_connection_address_info_label.setAlignment( + QtCore.Qt.AlignmentFlag.AlignCenter + ) + address_layout.addWidget(self.saved_connection_address_info_label) + + frame_inner_layout.addLayout(address_layout) info_layout.addWidget(self.frame) main_content_layout.addLayout(info_layout) @@ -3327,7 +3311,7 @@ def _make_row(label_text, field): self.vlan_id_spinbox.setValue(1) self.vlan_id_spinbox.lineEdit().setFocusPolicy(QtCore.Qt.FocusPolicy.NoFocus) self.vlan_id_spinbox.lineEdit().setReadOnly(True) - # Prevent text selection when stepping — deselect after each value change + # Prevent text selection when stepping: deselect after each value change self.vlan_id_spinbox.valueChanged.connect( lambda: self.vlan_id_spinbox.lineEdit().deselect() ) @@ -3586,6 +3570,7 @@ def _setup_navigation_signals(self) -> None: self.wifi_sip_apply_button.clicked.connect(self._on_wifi_static_ip_apply) self.wifi_sip_dhcp_button.clicked.connect(self._on_wifi_reset_dhcp) + @pyqtSlot() def _on_wifi_button_clicked(self) -> None: """Navigate to the Wi-Fi scan page, starting or stopping scan polling as needed.""" if ( @@ -3775,12 +3760,14 @@ def _on_show_keyboard( self._qwerty.show() field.clearFocus() + @pyqtSlot() def _on_qwerty_go_back(self) -> None: """Hide the keyboard and return to the previously active panel.""" if self._previous_panel: self._qwerty.hide() self.setCurrentIndex(self.indexOf(self._previous_panel)) + @pyqtSlot(str) def _on_qwerty_value_selected(self, value: str) -> None: """Apply the keyboard-selected *value* to the previously focused input field.""" if self._previous_panel: @@ -3789,6 +3776,7 @@ def _on_qwerty_value_selected(self, value: str) -> None: if self._current_field: self._current_field.setText(value) + @pyqtSlot(int) def _handle_scrollbar_change(self, value: int) -> None: """Synchronise the custom scrollbar thumb to the list-view scroll position.""" self.verticalScrollBar.blockSignals(True) @@ -3852,6 +3840,5 @@ def show_network_panel(self) -> None: parent_size = self.parent().size() self.setGeometry(0, 0, parent_size.width(), parent_size.height()) self.updateGeometry() - self.repaint() self.show() self._nm.scan_networks() diff --git a/tests/network/conftest.py b/tests/network/conftest.py index 2825b801..bba34d30 100644 --- a/tests/network/conftest.py +++ b/tests/network/conftest.py @@ -1,16 +1,16 @@ -"""tests/network/conftest.py — shared fixtures for network tests. +"""tests/network/conftest.py: shared fixtures for network tests. Mocks the D-Bus modules (sdbus, sdbus_async) BEFORE any network package import so that tests run without NetworkManager or a system bus. -Provides ``AsyncProxyMock`` — a mock for sdbus_async D-Bus proxies where +Provides ``AsyncProxyMock``: a mock for sdbus_async D-Bus proxies where property access returns awaitables (matching the real sdbus_async protocol) and method access returns ``AsyncMock`` callables. Widget stub modules export **real Qt base classes** (not ``MagicMock``) so that class inheritance in networkWindow.py works at import time. (``class IPAddressLineEdit(BlocksCustomLinEdit)`` requires a real type -as its base — ``MagicMock`` triggers ``TypeError: metaclass conflict``.) +as its base: ``MagicMock`` triggers ``TypeError: metaclass conflict``.) """ import asyncio @@ -34,7 +34,7 @@ _mock_sdbus.sd_bus_open_system = MagicMock(return_value=MagicMock()) _mock_sdbus.set_default_bus = MagicMock() -# Shared D-Bus NM mock — used by both sdbus_block and sdbus_async paths. +# Shared D-Bus NM mock: used by both sdbus_block and sdbus_async paths. _mock_dbus_nm = MagicMock() _mock_dbus_nm.enums = MagicMock() _mock_dbus_nm.enums.DeviceType = MagicMock() @@ -72,7 +72,7 @@ class _ConnectionStateReason(enum.IntEnum): sys.modules["sdbus"] = _mock_sdbus -# sdbus_block (legacy — kept for any residual imports) +# sdbus_block (legacy: kept for any residual imports) _mock_sdbus_block = MagicMock() _mock_sdbus_block.networkmanager = _mock_dbus_nm sys.modules["sdbus_block"] = _mock_sdbus_block @@ -85,10 +85,10 @@ class _ConnectionStateReason(enum.IntEnum): sys.modules["sdbus_async.networkmanager"] = _mock_dbus_nm -# Widget stub modules — REAL Qt base classes (not MagicMock) +# Widget stub modules: REAL Qt base classes (not MagicMock) # networkWindow.py subclasses imported widgets: # class IPAddressLineEdit(BlocksCustomLinEdit): ... -# A MagicMock cannot be used as a class base — it triggers a TypeError. +# A MagicMock cannot be used as a class base: it triggers a TypeError. # We create lightweight stub modules whose exports are real Qt types. @@ -171,7 +171,7 @@ def clear(self): self.endResetModel() def reconcile(self, desired, key_fn): - """Simplified reconcile — just replace entries.""" + """Simplified reconcile: just replace entries.""" self.beginResetModel() self.entries[:] = list(desired) self.endResetModel() @@ -202,14 +202,14 @@ def setFlat(self, v: bool) -> None: class _BlocksCustomCheckButtonStub(QtWidgets.QCheckBox): - """BlocksCustomCheckButton stand-in — adds setFlat() used in _setupUI.""" + """BlocksCustomCheckButton stand-in: adds setFlat() used in _setupUI.""" def setFlat(self, v: bool) -> None: pass class _BlocksCustomLinEditStub(QtWidgets.QLineEdit): - """BlocksCustomLinEdit stand-in — adds clicked signal used in _setup_hidden_network_page.""" + """BlocksCustomLinEdit stand-in: adds clicked signal used in _setup_hidden_network_page.""" clicked = QtCore.pyqtSignal(name="clicked") @@ -221,6 +221,12 @@ class _KeyboardStub(QtWidgets.QWidget): def set_value(self, val): pass + def setPattern(self, pattern): + self.pattern = pattern + + def setNumericOnly(self, enabled): + self.numeric_only = enabled + # Register parent packages first (must be real modules, not MagicMock). _lib_parent_packages = ("lib", "lib.panels", "lib.panels.widgets", "lib.utils") @@ -281,7 +287,7 @@ def set_value(self, val): sys.modules["BlocksScreen." + _mod_name] = _stub -# Mock lib.qrcode_gen (short path only) — networkWindow.py imports it as +# Mock lib.qrcode_gen (short path only): networkWindow.py imports it as # ``from lib.qrcode_gen import generate_wifi_qrcode``. The BlocksScreen.* # path is intentionally NOT registered here so test_qrcode_gen_unit.py can # still import the real module via ``BlocksScreen.lib.qrcode_gen``. @@ -301,7 +307,7 @@ def set_value(self, val): sys.modules["configfile"] = _mock_configfile_mod # Now safe to import the actual network package -from BlocksScreen.lib.network.models import ( # noqa: E402 +from BlocksScreen.lib.network.models import ( ConnectionPriority, ConnectivityState, NetworkInfo, @@ -316,7 +322,7 @@ def set_value(self, val): sys.modules["lib.network"] = sys.modules["BlocksScreen.lib.network"] -# AsyncProxyMock — sdbus_async D-Bus proxy mock +# AsyncProxyMock: sdbus_async D-Bus proxy mock class _AwaitableProp: @@ -340,6 +346,9 @@ def __eq__(self, other): return self.value == other.value return self.value == other + # Real sdbus property proxies are hashable; asyncio.gather() requires it. + __hash__ = object.__hash__ + def __ne__(self, other): return not self.__eq__(other) @@ -350,8 +359,8 @@ def __repr__(self): class AsyncProxyMock: """Mock for sdbus_async D-Bus proxies. - * **Properties** -> ``_AwaitableProp`` — ``await proxy.prop`` returns value. - * **Methods** -> ``AsyncMock`` — ``await proxy.method()`` is configurable. + * **Properties** -> ``_AwaitableProp``: ``await proxy.prop`` returns value. + * **Methods** -> ``AsyncMock``: ``await proxy.method()`` is configurable. * Unknown attribute access auto-creates an ``AsyncMock`` (method). * Setting a plain value creates/updates an ``_AwaitableProp``. * Setting an ``AsyncMock`` registers it as a method. @@ -408,7 +417,7 @@ def __setattr__(self, name, value): def _run(coro): - """Run a single coroutine to completion — test helper for async worker methods.""" + """Run a single coroutine to completion: test helper for async worker methods.""" loop = asyncio.new_event_loop() try: return loop.run_until_complete(coro) @@ -419,7 +428,7 @@ def _run(coro): # QApplication singleton @pytest.fixture(scope="session") def qapp(): - """Session-scoped QApplication — created once for all tests.""" + """Session-scoped QApplication: created once for all tests.""" app = QtWidgets.QApplication.instance() if app is None: app = QtWidgets.QApplication([]) @@ -571,10 +580,10 @@ def _stub_init(self, *_a, **_kw): window._pending_operation = PendingOperation.NONE window._target_ssid = "" window._active_signal = 0 - window._was_ethernet_connected = False window._pending_expected_ip = "" window._last_active_signal_bars = 0 window._cached_scan_networks = [] + window._last_state_summary = None # Refactored list-cache instance variables window._item_cache = {} @@ -655,6 +664,8 @@ def _stub_init(self, *_a, **_kw): # Missing instance variables (from _init_instance_variables) window._initial_priority = ConnectionPriority.MEDIUM + window._initial_password = "" + window._password_ssid = "" window._current_network_is_open = False window._current_network_is_hidden = False window._previous_panel = None @@ -685,6 +696,7 @@ def _stub_init(self, *_a, **_kw): window.saved_connection_change_password_view = QtWidgets.QCheckBox(parent) window.saved_connection_signal_strength_info_frame = QtWidgets.QLabel(parent) window.saved_connection_security_type_info_label = QtWidgets.QLabel(parent) + window.saved_connection_address_info_label = QtWidgets.QLabel(parent) window.network_activate_btn = QtWidgets.QPushButton(parent) window.sn_info = QtWidgets.QLabel(parent) window.frame = QtWidgets.QFrame(parent) @@ -730,5 +742,5 @@ def _stub_init(self, *_a, **_kw): except Exception as exc: traceback.print_exc() pytest.skip( - f"NetworkControlWindow not importable — {exc.__class__.__name__}: {exc}" + f"NetworkControlWindow not importable ({exc.__class__.__name__}: {exc})" ) diff --git a/tests/network/test_network_ui.py b/tests/network/test_network_ui.py index 9fd4e750..69a76b2a 100644 --- a/tests/network/test_network_ui.py +++ b/tests/network/test_network_ui.py @@ -9,18 +9,18 @@ Coverage targets ---------------- -* _handle_first_run — all 5 branches (ethernet / wifi-full / hotspot / +* _handle_first_run: all 5 branches (ethernet / wifi-full / hotspot / wifi-on-no-conn / disconnected) -* _on_network_state_changed — normal display path + loading-state machine -* _display_connected_state — ethernet vs Wi-Fi vs hotspot +* _on_network_state_changed: normal display path + loading-state machine +* _display_connected_state: ethernet vs Wi-Fi vs hotspot * _display_disconnected_state / _display_wifi_on_no_connection -* _sync_ethernet_panel — carrier visibility + toggle sync +* _sync_ethernet_panel: carrier visibility + toggle sync * _set_loading_state / _clear_loading -* _handle_load_timeout — each pending-operation branch +* _handle_load_timeout: each pending-operation branch * _on_reconnect_complete -* _on_operation_complete — success/failure branches +* _on_operation_complete: success/failure branches * _handle_wifi_toggle / _handle_hotspot_toggle / _handle_ethernet_toggle -* _emit_status_icon — ethernet / hotspot / wifi / disconnected +* _emit_status_icon: ethernet / hotspot / wifi / disconnected """ from unittest.mock import MagicMock, patch @@ -58,6 +58,17 @@ def test_new_ip_signal_removed(): # ───────────────────────────────────────────────────────────────────────────── +def _all_on(w) -> None: + """Force all three link toggles ON so exclusivity is observable.""" + for btn in (w.wifi_button, w.hotspot_button, w.ethernet_button): + btn.toggle_button.state = btn.toggle_button.State.ON + + +def _off(btn) -> bool: + """True when a link toggle reads OFF.""" + return btn.toggle_button.state == btn.toggle_button.State.OFF + + def _eth_state(**kw) -> NetworkState: """Minimal ethernet-connected state.""" defaults = dict( @@ -178,11 +189,14 @@ def test_ethernet_connected_shows_connected_state(self, win): assert w.netlist_ssuid.isVisible() assert w.netlist_ssuid.text() == "Ethernet" - def test_ethernet_disables_wifi_if_enabled(self, win): + def test_ethernet_never_kills_radio_at_boot(self, win): + """Boot is display-only: the radio is the recovery path, never touched here.""" w, nm = win state = _eth_state(wifi_enabled=True) w._handle_first_run(state) - nm.set_wifi_enabled.assert_called_once_with(False) + nm.set_wifi_enabled.assert_not_called() + wifi_btn = w.wifi_button.toggle_button + assert wifi_btn.state == wifi_btn.State.OFF def test_ethernet_does_not_disable_wifi_if_already_off(self, win): w, nm = win @@ -481,7 +495,7 @@ def test_navigates_to_main_page(self, win, qapp): # ───────────────────────────────────────────────────────────────────────────── -# _on_network_state_changed — normal (not connecting) path +# _on_network_state_changed: normal (not connecting) path # ───────────────────────────────────────────────────────────────────────────── @@ -490,7 +504,6 @@ def _prime(self, w): """Mark first-run as done so the normal display path runs.""" w._is_first_run = False w._is_connecting = False - w._was_ethernet_connected = False def test_ethernet_shows_connected(self, win): w, _ = win @@ -522,18 +535,17 @@ def test_first_run_flag_cleared_after_first_call(self, win): w._on_network_state_changed(_disconnected_state()) assert w._is_first_run is False - def test_ethernet_plug_disables_wifi(self, win): - """Ethernet cable plugged in during Wi-Fi session -> Wi-Fi disabled.""" + def test_ethernet_plug_keeps_wifi_enabled(self, win): + """Wi-Fi is the recovery path; a plugged cable must never kill the radio.""" w, nm = win self._prime(w) - w._was_ethernet_connected = False state = _eth_state(wifi_enabled=True) w._on_network_state_changed(state) - nm.set_wifi_enabled.assert_called_with(False) + nm.set_wifi_enabled.assert_not_called() # ───────────────────────────────────────────────────────────────────────────── -# _on_network_state_changed — loading state machine +# _on_network_state_changed: loading state machine # ───────────────────────────────────────────────────────────────────────────── @@ -575,7 +587,7 @@ def test_wifi_connect_keeps_loading_for_wrong_ssid(self, win): self._start_loading(w, PendingOperation.CONNECT) w._target_ssid = "OtherNet" w._on_network_state_changed(_wifi_state()) - # Should still be loading — SSID doesn't match target + # Should still be loading: SSID doesn't match target assert w._is_connecting def test_ethernet_on_clears_on_connected(self, win): @@ -701,7 +713,7 @@ def test_transient_mismatch_retries(self, win, qapp): with patch("BlocksScreen.lib.panels.networkWindow.QTimer") as mock_timer: w._on_operation_complete(result) mock_timer.singleShot.assert_called_once() - # Loading should still be visible — retry is pending + # Loading should still be visible: retry is pending assert w._is_connecting @@ -756,6 +768,19 @@ def test_wifi_on_calls_set_wifi_enabled(self, win): w._handle_wifi_toggle(True) nm.set_wifi_enabled.assert_called_once_with(True) + def test_wifi_on_drops_ethernet(self, win): + w, nm = win + nm.saved_networks = [] + w._handle_wifi_toggle(True) + nm.disconnect_ethernet.assert_called_once() + + def test_wifi_on_turns_other_toggles_off(self, win): + w, nm = win + nm.saved_networks = [] + _all_on(w) + w._handle_wifi_toggle(True) + assert _off(w.hotspot_button) and _off(w.ethernet_button) + class TestHotspotToggle: def test_hotspot_off_calls_toggle_hotspot_false(self, win): @@ -783,6 +808,17 @@ def test_hotspot_on_sets_loading(self, win): w._handle_hotspot_toggle(True) assert w.loadingwidget.isVisible() + def test_hotspot_on_drops_ethernet(self, win): + w, nm = win + w._handle_hotspot_toggle(True) + nm.disconnect_ethernet.assert_called_once() + + def test_hotspot_on_turns_other_toggles_off(self, win): + w, nm = win + _all_on(w) + w._handle_hotspot_toggle(True) + assert _off(w.wifi_button) and _off(w.ethernet_button) + class TestEthernetToggle: def test_ethernet_on_calls_connect_ethernet(self, win): @@ -805,6 +841,22 @@ def test_ethernet_off_sets_pending_ethernet_off(self, win): w._handle_ethernet_toggle(False) assert w._pending_operation == PendingOperation.ETHERNET_OFF + def test_ethernet_on_drops_wifi(self, win): + w, nm = win + w._handle_ethernet_toggle(True) + nm.set_wifi_enabled.assert_called_once_with(False) + + def test_ethernet_on_turns_other_toggles_off(self, win): + w, nm = win + _all_on(w) + w._handle_ethernet_toggle(True) + assert _off(w.wifi_button) and _off(w.hotspot_button) + + def test_ethernet_off_leaves_wifi_alone(self, win): + w, nm = win + w._handle_ethernet_toggle(False) + nm.set_wifi_enabled.assert_not_called() + # ───────────────────────────────────────────────────────────────────────────── # _emit_status_icon @@ -936,13 +988,13 @@ def test_clears_loading(self, win): # ───────────────────────────────────────────────────────────────────────────── -# _setupUI smoke test — covers ~1 200 statements in _setupUI + page helpers +# _setupUI smoke test: covers ~1 200 statements in _setupUI + page helpers # ───────────────────────────────────────────────────────────────────────────── @pytest.mark.unit class TestSetupUIRunsWithoutError: - """Calling _setupUI() must not raise — covers the entire UI construction path.""" + """Calling _setupUI() must not raise: covers the entire UI construction path.""" def test_setup_ui_completes(self, qapp): from unittest.mock import patch @@ -966,7 +1018,7 @@ def _stub_init(self, *_a, **_kw): # ───────────────────────────────────────────────────────────────────────────── -# Step 4a: Helper classes — PixmapCache, WifiIconProvider, IPAddressLineEdit +# Step 4a: Helper classes PixmapCache, WifiIconProvider, IPAddressLineEdit # ───────────────────────────────────────────────────────────────────────────── @@ -1882,3 +1934,51 @@ def test_show_hotspot_qr_clears_on_error(self, win): w._show_hotspot_qr("TestAP", "testpass123", "wpa-psk") w.qrcode_img.clearPixmap.assert_called() w.qrcode_img.setText.assert_called_with("QR error") + + +class TestOnNetworkPasswordLoaded: + """Prefill of the change-password field, including the stale-ssid guard.""" + + def test_matching_ssid_fills_field(self, win): + w, _ = win + w._password_ssid = "HomeNet" + w._on_network_password_loaded("HomeNet", "secret123") + assert w.saved_connection_change_password_field.text() == "secret123" + + def test_matching_ssid_updates_baseline(self, win): + w, _ = win + w._password_ssid = "HomeNet" + w._on_network_password_loaded("HomeNet", "secret123") + assert w._initial_password == "secret123" + + def test_non_empty_password_clears_placeholder(self, win): + w, _ = win + w._password_ssid = "HomeNet" + w.saved_connection_change_password_field.setPlaceholderText("Enter password") + w._on_network_password_loaded("HomeNet", "secret123") + assert w.saved_connection_change_password_field.placeholderText() == "" + + def test_empty_password_keeps_placeholder(self, win): + w, _ = win + w._password_ssid = "HomeNet" + w.saved_connection_change_password_field.setPlaceholderText("Enter password") + w._on_network_password_loaded("HomeNet", "") + assert w.saved_connection_change_password_field.placeholderText() == ( + "Enter password" + ) + + def test_stale_ssid_ignored(self, win): + """A late reply for a previously viewed network must not leak its psk.""" + w, _ = win + w._password_ssid = "HomeNet" + w.saved_connection_change_password_field.setText("") + w._initial_password = "" + w._on_network_password_loaded("OtherNet", "othersecret") + assert w.saved_connection_change_password_field.text() == "" + assert w._initial_password == "" + + def test_empty_tracked_ssid_ignores_reply(self, win): + w, _ = win + w._password_ssid = "" + w._on_network_password_loaded("HomeNet", "secret123") + assert w.saved_connection_change_password_field.text() == "" diff --git a/tests/network/test_sdbus_integration.py b/tests/network/test_sdbus_integration.py index e6c3dc37..149e921c 100644 --- a/tests/network/test_sdbus_integration.py +++ b/tests/network/test_sdbus_integration.py @@ -27,12 +27,13 @@ import asyncio import os from contextlib import contextmanager +from pathlib import Path import pytest from PyQt6.QtCore import Qt # ───────────────────────────────────────────────────────────────────────────── -# Gate — skip entire module when opt-in flag is absent +# Gate: skip entire module when opt-in flag is absent # ───────────────────────────────────────────────────────────────────────────── _ENABLED = os.environ.get("NM_INTEGRATION_TESTS", "0") == "1" _SKIP = pytest.mark.skipif(not _ENABLED, reason="NM_INTEGRATION_TESTS not set") @@ -41,6 +42,28 @@ pytestmark = [_SKIP, pytest.mark.timeout(120)] +def _host_has_wired_nic() -> bool: + """True when sysfs shows a physical ARPHRD_ETHER NIC that is not Wi-Fi.""" + try: + entries = list(Path("/sys/class/net").iterdir()) + except OSError: + return False + for p in entries: + try: + if (p / "type").read_text().strip() != "1": + continue + except OSError: + continue + if not (p / "wireless").is_dir() and (p / "device").exists(): + return True + return False + + +_NEEDS_WIRED = pytest.mark.skipif( + not _host_has_wired_nic(), reason="host has no wired NIC" +) + + # ───────────────────────────────────────────────────────────────────────────── # Signal capture helper # ───────────────────────────────────────────────────────────────────────────── @@ -82,7 +105,6 @@ def real_worker(qapp): """ import sys import threading - from pathlib import Path # Add BlocksScreen/ to sys.path so `import configfile` resolves to # BlocksScreen/configfile.py (worker.py imports it at module level). @@ -107,7 +129,7 @@ def real_worker(qapp): try: from BlocksScreen.lib.network.worker import NetworkManagerWorker except ImportError as exc: - # Real sdbus packages not installed on this host — skip gracefully. + # Real sdbus packages not installed on this host: skip gracefully. sys.modules.update(_saved_stubs) if _path_was_added and sys.path and sys.path[0] == _bs_dir: sys.path.pop(0) @@ -195,6 +217,7 @@ class TestRealInterfaces: def test_wifi_path_detected(self, real_worker): assert real_worker._primary_wifi_path, "No Wi-Fi interface found" + @_NEEDS_WIRED def test_wired_path_detected(self, real_worker): assert real_worker._primary_wired_path, "No wired interface found" @@ -330,7 +353,7 @@ def test_os_fallback_unknown_iface_returns_empty(self, real_worker): # ───────────────────────────────────────────────────────────────────────────── -# Destructive write tests — TEST_-prefixed profiles only +# Destructive write tests: TEST_-prefixed profiles only # ───────────────────────────────────────────────────────────────────────────── diff --git a/tests/network/test_worker_unit.py b/tests/network/test_worker_unit.py index b102bcaf..9f84a215 100644 --- a/tests/network/test_worker_unit.py +++ b/tests/network/test_worker_unit.py @@ -1,10 +1,10 @@ """Unit tests for BlocksScreen.lib.network.worker.NetworkManagerWorker. -All D-Bus modules are mocked via conftest.py — these tests run +All D-Bus modules are mocked via conftest.py: these tests run without NetworkManager or a system bus. Architecture: Tests target the sdbus_async worker API. -Async coroutines are tested directly via ``pytest-asyncio`` — NO daemon +Async coroutines are tested directly via ``pytest-asyncio``: NO daemon thread, NO ``_run_sync``, NO ``run_coroutine_threadsafe``. The ``_make_worker`` helper bypasses ``__init__`` so the asyncio daemon @@ -48,11 +48,14 @@ def _make_worker(qapp, *, running=True, with_wifi=True, with_wired=False): ): w = NetworkManagerWorker() - # Core state — mirrors real __init__ + # Core state: mirrors real __init__ w._running = running w._stopping = False - w._system_bus = MagicMock(name="mock_system_bus") w._no_iface_reported = False + w._rediscover_lock = asyncio.Lock() + w._rediscover_gen = 0 + w._stale_logged_gen = -1 + w._system_bus = MagicMock(name="mock_system_bus") w._primary_wifi_path = ( "/org/freedesktop/NetworkManager/Devices/2" if with_wifi else "" ) @@ -77,9 +80,6 @@ def _make_worker(qapp, *, running=True, with_wifi=True, with_wired=False): w._state_debounce_handle = None w._scan_debounce_handle = None w._listener_tasks = [] - w._rediscover_lock = asyncio.Lock() - w._rediscover_gen = 0 - w._stale_logged_gen = -1 # Stubs for thread-related attrs (never used in async tests) w._asyncio_loop = MagicMock() @@ -89,15 +89,18 @@ def _make_worker(qapp, *, running=True, with_wifi=True, with_wired=False): def _bare_worker(qapp): - """Minimal worker for signal / property tests — no mock state.""" + """Minimal worker for signal / property tests: no mock state.""" with patch.object( NetworkManagerWorker, "__init__", lambda self: QObject.__init__(self) ): w = NetworkManagerWorker() w._running = False w._stopping = False - w._system_bus = None w._no_iface_reported = False + w._rediscover_lock = asyncio.Lock() + w._rediscover_gen = 0 + w._stale_logged_gen = -1 + w._system_bus = None w._primary_wifi_path = "" w._primary_wifi_iface = "" w._primary_wired_path = "" @@ -117,9 +120,6 @@ def _bare_worker(qapp): w._state_debounce_handle = None w._scan_debounce_handle = None w._listener_tasks = [] - w._rediscover_lock = asyncio.Lock() - w._rediscover_gen = 0 - w._stale_logged_gen = -1 w._asyncio_loop = MagicMock() w._asyncio_thread = MagicMock() return w @@ -132,8 +132,11 @@ def _make(qapp, *, running=True, wifi=True, wired=True): w = NetworkManagerWorker() w._running = running w._stopping = False - w._system_bus = MagicMock(name="mock_bus") w._no_iface_reported = False + w._rediscover_lock = asyncio.Lock() + w._rediscover_gen = 0 + w._stale_logged_gen = -1 + w._system_bus = MagicMock(name="mock_bus") w._primary_wifi_path = "/org/freedesktop/NetworkManager/Devices/2" if wifi else "" w._primary_wifi_iface = "wlan0" if wifi else "" w._primary_wired_path = "/org/freedesktop/NetworkManager/Devices/1" if wired else "" @@ -153,14 +156,25 @@ def _make(qapp, *, running=True, wifi=True, wired=True): w._state_debounce_handle = None w._scan_debounce_handle = None w._listener_tasks = [] - w._rediscover_lock = asyncio.Lock() - w._rediscover_gen = 0 - w._stale_logged_gen = -1 w._asyncio_loop = MagicMock() w._asyncio_thread = MagicMock() return w +class TestFixtureParity: + """The factories above bypass __init__, so new attributes must be mirrored there.""" + + def test_factories_cover_real_init_attrs(self, qapp): + with patch.object(NetworkManagerWorker, "_run_asyncio_loop"): + real = NetworkManagerWorker() + real._asyncio_thread.join(timeout=2.0) + real._asyncio_loop.close() + expected = set(vars(real)) + for factory in (_make_worker, _bare_worker, _make): + missing = expected - set(vars(factory(qapp))) + assert not missing, f"{factory.__name__} missing {sorted(missing)}" + + def _wire(w, *, nm=None, wifi_proxy=None, wired_proxy=None, settings=None): """Wire mock D-Bus proxy factories onto worker.""" if nm is not None: @@ -251,7 +265,7 @@ async def test_happy_path_sets_running(self, qapp): w = _make_worker(qapp, running=False) # Mock all async calls in initialize w._detect_interfaces = AsyncMock() - w._enforce_boot_mutual_exclusion = AsyncMock() + w._ensure_wired_autoconnect = AsyncMock() w._is_ethernet_connected = AsyncMock(return_value=False) w._activate_saved_vlans = AsyncMock() w._start_signal_listeners = AsyncMock() @@ -570,18 +584,33 @@ async def test_emits_unknown_when_no_bus(self, qapp): @pytest.mark.asyncio async def test_emits_correct_state(self, qapp): w = _make_worker(qapp) - nm_proxy = AsyncProxyMock(check_connectivity=AsyncMock(return_value=3)) + nm_proxy = AsyncProxyMock(connectivity=3) w._nm = _ProxyFactory(nm_proxy) received = [] w.connectivity_changed.connect(lambda c: received.append(c)) await w._async_check_connectivity() assert received == [ConnectivityState.LIMITED] + nm_proxy.check_connectivity.assert_not_called() + + @pytest.mark.asyncio + async def test_unknown_property_falls_back_to_active_probe(self, qapp): + w = _make_worker(qapp) + nm_proxy = AsyncProxyMock( + connectivity=0, check_connectivity=AsyncMock(return_value=4) + ) + w._nm = _ProxyFactory(nm_proxy) + received = [] + w.connectivity_changed.connect(lambda c: received.append(c)) + await w._async_check_connectivity() + assert received == [ConnectivityState.FULL] + nm_proxy.check_connectivity.assert_awaited_once() @pytest.mark.asyncio async def test_emits_unknown_on_error(self, qapp): w = _make_worker(qapp) nm_proxy = AsyncProxyMock( - check_connectivity=AsyncMock(side_effect=Exception("D-Bus error")) + connectivity=0, + check_connectivity=AsyncMock(side_effect=Exception("D-Bus error")), ) w._nm = _ProxyFactory(nm_proxy) received = [] @@ -696,12 +725,26 @@ async def test_happy_path_returns_ip(self, qapp): w = _make_worker(qapp) nm_proxy = AsyncProxyMock(primary_connection="/active/1") w._nm = _ProxyFactory(nm_proxy) - active_proxy = AsyncProxyMock(ip4_config="/ip4/1") + active_proxy = AsyncProxyMock( + ip4_config="/ip4/1", connection_type="802-11-wireless" + ) w._active_conn = lambda path: active_proxy ipv4_proxy = AsyncProxyMock(address_data=[{"address": ("s", "192.168.1.50")}]) w._ipv4 = lambda path: ipv4_proxy assert await w._get_current_ip() == "192.168.1.50" + @pytest.mark.asyncio + async def test_vpn_primary_is_ignored(self, qapp): + w = _make_worker(qapp) + w._nm = _ProxyFactory(AsyncProxyMock(primary_connection="/active/1")) + w._active_conn = lambda path: AsyncProxyMock( + ip4_config="/ip4/1", connection_type="tun" + ) + w._ipv4 = lambda path: AsyncProxyMock( + address_data=[{"address": ("s", "100.75.1.69")}] + ) + assert await w._get_current_ip() == "" + @pytest.mark.asyncio async def test_exception_returns_empty(self, qapp): w = _make_worker(qapp) @@ -712,17 +755,64 @@ async def test_exception_returns_empty(self, qapp): assert await w._get_current_ip() == "" +class TestActiveApSignal: + @pytest.mark.asyncio + async def test_returns_strength_of_active_ap(self, qapp): + w = _make_worker(qapp) + w._wifi = _ProxyFactory(AsyncProxyMock(active_access_point="/ap/1")) + w._ap = lambda path: AsyncProxyMock(strength=72) + assert await w._active_ap_signal() == 72 + + @pytest.mark.asyncio + async def test_no_wifi_device_returns_zero(self, qapp): + w = _make_worker(qapp, with_wifi=False) + assert await w._active_ap_signal() == 0 + + @pytest.mark.asyncio + async def test_unassociated_slash_path_returns_zero(self, qapp): + w = _make_worker(qapp) + w._wifi = _ProxyFactory(AsyncProxyMock(active_access_point="/")) + assert await w._active_ap_signal() == 0 + + @pytest.mark.asyncio + async def test_exception_returns_zero(self, qapp): + w = _make_worker(qapp) + w._wifi = _ProxyFactory( + AsyncProxyMock(active_access_point=AsyncMock(side_effect=Exception("gone"))) + ) + assert await w._active_ap_signal() == 0 + + class TestGetIpByInterface: @pytest.mark.asyncio async def test_cached_path_used(self, qapp): w = _make_worker(qapp) w._iface_to_device_path = {"wlan0": "/dev/wifi0"} - generic_proxy = AsyncProxyMock(interface="wlan0", ip4_config="/ip4/1") + generic_proxy = AsyncProxyMock(ip4_config="/ip4/1", interface="wlan0") w._generic = lambda path: generic_proxy ipv4_proxy = AsyncProxyMock(address_data=[{"address": ("s", "192.168.1.50")}]) w._ipv4 = lambda path: ipv4_proxy assert await w._get_ip_by_interface("wlan0") == "192.168.1.50" + @pytest.mark.asyncio + async def test_stale_cached_path_is_reresolved(self, qapp): + """NM reuses object paths across restarts; a reused path must not leak its IP.""" + w = _make_worker(qapp) + w._iface_to_device_path = {"wlan0": "/dev/stale"} + proxies = { + "/dev/stale": AsyncProxyMock(interface="eth0"), + "/dev/wifi1": AsyncProxyMock(interface="wlan0", ip4_config="/ip4/1"), + } + w._generic = lambda path: proxies[path] + w._nm = _ProxyFactory( + AsyncProxyMock(get_devices=AsyncMock(return_value=["/dev/wifi1"])) + ) + w._ipv4 = lambda path: AsyncProxyMock( + address_data=[{"address": ("s", "192.168.1.50")}] + ) + assert await w._get_ip_by_interface("wlan0") == "192.168.1.50" + assert w._iface_to_device_path["wlan0"] == "/dev/wifi1" + @pytest.mark.asyncio async def test_no_matching_interface_returns_empty(self, qapp): w = _make_worker(qapp) @@ -781,7 +871,7 @@ async def test_returns_default_when_no_bus(self, qapp): async def test_connected_state(self, qapp): w = _make_worker(qapp) nm_proxy = AsyncProxyMock( - check_connectivity=AsyncMock(return_value=4), + connectivity=4, wireless_enabled=True, primary_connection="/", active_connections=[], @@ -800,7 +890,7 @@ async def test_connected_state(self, qapp): async def test_connected_with_ssid_gets_signal_and_security(self, qapp): w = _make_worker(qapp) nm_proxy = AsyncProxyMock( - check_connectivity=AsyncMock(return_value=4), + connectivity=4, wireless_enabled=True, ) w._nm = _ProxyFactory(nm_proxy) @@ -832,7 +922,7 @@ async def test_hotspot_state_has_correct_security(self, qapp): w = _make_worker(qapp) w._hotspot_config.ssid = "TestHotspot" nm_proxy = AsyncProxyMock( - check_connectivity=AsyncMock(return_value=4), + connectivity=4, wireless_enabled=True, ) w._nm = _ProxyFactory(nm_proxy) @@ -848,13 +938,33 @@ async def test_hotspot_state_has_correct_security(self, qapp): assert state.hotspot_enabled is True assert state.security_type == "wpa-psk" + @pytest.mark.asyncio + async def test_ap_mode_detected_as_hotspot_without_our_flag(self, qapp): + """An AP the app did not start must not be reported as a client link.""" + w = _make_worker(qapp) + w._hotspot_config.ssid = "PrinterHotspot" + w._is_hotspot_active = False + nm_proxy = AsyncProxyMock(connectivity=4, wireless_enabled=True) + w._nm = _ProxyFactory(nm_proxy) + w._is_wifi_ap_mode = AsyncMock(return_value=True) + w._get_current_ssid = AsyncMock(return_value="FOREIGN-AP") + w._get_ip_by_interface = AsyncMock(return_value="10.42.0.1") + w._get_current_ip = AsyncMock(return_value="10.42.0.1") + w._is_ethernet_connected = AsyncMock(return_value=False) + w._has_ethernet_carrier = AsyncMock(return_value=False) + w._build_signal_map = AsyncMock(return_value={}) + w._get_active_vlans = AsyncMock(return_value=[]) + + state = await w._build_current_state() + assert state.hotspot_enabled is True + @pytest.mark.asyncio async def test_hotspot_flag_fallback_when_dbus_ssid_empty(self, qapp): w = _make_worker(qapp) w._hotspot_config.ssid = "PrinterHotspot" w._is_hotspot_active = True nm_proxy = AsyncProxyMock( - check_connectivity=AsyncMock(return_value=4), + connectivity=4, wireless_enabled=True, ) w._nm = _ProxyFactory(nm_proxy) @@ -874,7 +984,7 @@ async def test_hotspot_flag_fallback_when_dbus_ssid_empty(self, qapp): async def test_ethernet_connected_included_in_state(self, qapp): w = _make_worker(qapp, with_wired=True) nm_proxy = AsyncProxyMock( - check_connectivity=AsyncMock(return_value=4), + connectivity=4, wireless_enabled=False, ) w._nm = _ProxyFactory(nm_proxy) @@ -892,7 +1002,8 @@ async def test_ethernet_connected_included_in_state(self, qapp): async def test_exception_returns_default(self, qapp): w = _make_worker(qapp) nm_proxy = AsyncProxyMock( - check_connectivity=AsyncMock(side_effect=RuntimeError("bang")) + connectivity=0, + check_connectivity=AsyncMock(side_effect=RuntimeError("bang")), ) w._nm = _ProxyFactory(nm_proxy) state = await w._build_current_state() @@ -1112,6 +1223,30 @@ async def mock_props(path): result = await w._build_signal_map() assert result["samenet"] == 80 + @pytest.mark.asyncio + async def test_stale_path_recovers_and_retries(self, qapp): + w = _make_worker(qapp) + w._recover_signal_sources = AsyncMock() + w._signal_map_once = AsyncMock( + side_effect=[RuntimeError("Object does not exist at path"), {"net": 55}] + ) + assert await w._build_signal_map() == {"net": 55} + w._recover_signal_sources.assert_awaited_once() + + @pytest.mark.asyncio + async def test_returns_empty_when_retry_also_fails(self, qapp): + w = _make_worker(qapp) + w._recover_signal_sources = AsyncMock() + w._signal_map_once = AsyncMock(side_effect=RuntimeError("boom")) + assert await w._build_signal_map() == {} + + @pytest.mark.asyncio + async def test_no_wifi_path_skips_recovery(self, qapp): + w = _make_worker(qapp, with_wifi=False) + w._recover_signal_sources = AsyncMock() + assert await w._build_signal_map() == {} + w._recover_signal_sources.assert_not_awaited() + class TestSavedNetworkCache: def test_invalidate_marks_dirty(self, qapp): @@ -1280,7 +1415,7 @@ def test_wpa_psk(self, qapp): assert result["802-11-wireless-security"]["key-mgmt"] == ("s", "wpa-psk") def test_wep_returns_none(self, qapp): - """WEP is unsupported — returns None.""" + """WEP is unsupported: returns None.""" result = self._call(qapp, flags=1) # privacy flag but no WPA/RSN assert result is None @@ -1555,10 +1690,10 @@ async def test_updates_password(self, qapp): update=AsyncMock(), ) w._conn_settings = lambda path: conn_proxy - result = await w._update_network_impl("net", "newpass", None) + result = await w._update_network_impl("net", "newpass1", None) assert result.success is True call_args = conn_proxy.update.call_args[0][0] - assert call_args["802-11-wireless-security"]["psk"] == ("s", "newpass") + assert call_args["802-11-wireless-security"]["psk"] == ("s", "newpass1") @pytest.mark.asyncio async def test_updates_priority(self, qapp): @@ -1715,7 +1850,7 @@ async def test_ap_not_found_returns_not_found(self, qapp): interface="wlan0", ) w._wifi = _ProxyFactory(wifi_proxy) - result = await w._add_network_impl("Ghost", "pass", 0) + result = await w._add_network_impl("Ghost", "password1", 0) assert result.error_code == "not_found" @pytest.mark.asyncio @@ -1741,7 +1876,7 @@ async def test_unsupported_security_returns_error(self, qapp): "rsn_flags": 0x200, } ) - result = await w._add_network_impl("EAPNet", "pass", 0) + result = await w._add_network_impl("EAPNet", "password1", 0) assert result.error_code == "unsupported_security" @@ -1847,7 +1982,7 @@ def test_eap_connection_profile_returns_none(self, qapp): assert result is None def test_wep_connection_returns_none(self, qapp): - """WEP is unsupported — _build_connection_properties returns None.""" + """WEP is unsupported: _build_connection_properties returns None.""" w = _make_worker(qapp) ap_props = {"flags": 1, "wpa_flags": 0, "rsn_flags": 0} result = w._build_connection_properties( @@ -1881,6 +2016,10 @@ def test_invalid_raises(self): with pytest.raises(ValueError): NetworkManagerWorker._mask_to_prefix("33") + def test_non_contiguous_mask_rejected(self): + with pytest.raises(ValueError, match="Invalid subnet mask"): + NetworkManagerWorker._mask_to_prefix("255.0.255.0") + class TestAsyncShutdown: def test_sets_not_running(self, qapp): @@ -1888,15 +2027,16 @@ def test_sets_not_running(self, qapp): _run(w._async_shutdown()) assert w._running is False - @pytest.mark.asyncio - async def test_clears_listener_tasks(self, qapp): - async def dummy(): - await asyncio.sleep(10) - + def test_clears_listener_tasks(self, qapp): w = _make(qapp) - task = asyncio.create_task(dummy()) - w._listener_tasks = [task] - await w._async_shutdown() + + async def _body(): + task = asyncio.create_task(asyncio.sleep(30)) + w._listener_tasks = [task] + await w._async_shutdown() + return task + + task = _run(_body()) assert task.cancelled() assert w._listener_tasks == [] @@ -2046,6 +2186,115 @@ def test_calls_state_and_connectivity(self, qapp): w._async_load_saved_networks.assert_awaited_once() +class TestWiredProfilesAutoconnect: + """Device.Autoconnect dies on NM restart; only the profile flag persists.""" + + @staticmethod + def _wire_profiles(w, conn_type="802-3-ethernet", autoconnect=True): + nm_settings_proxy = AsyncProxyMock( + list_connections=AsyncMock(return_value=["/conn/eth"]) + ) + w._nm_settings = _ProxyFactory(nm_settings_proxy) + settings = { + "connection": { + "type": ("s", conn_type), + "autoconnect": ("b", autoconnect), + "timestamp": ("t", 123), + }, + "ipv4": {"method": ("s", "auto")}, + } + w._gather_settings = AsyncMock(return_value=[("/conn/eth", settings)]) + conn_proxy = AsyncProxyMock(update=AsyncMock()) + w._conn_settings = lambda path: conn_proxy + return conn_proxy + + @pytest.mark.asyncio + async def test_disables_wired_profile(self, qapp): + w = _make_worker(qapp) + conn = self._wire_profiles(w, autoconnect=True) + await w._set_wired_profiles_autoconnect(False) + props = conn.update.await_args[0][0] + assert props["connection"]["autoconnect"] == ("b", False) + + @pytest.mark.asyncio + async def test_strips_timestamp_nm_will_not_accept(self, qapp): + w = _make_worker(qapp) + conn = self._wire_profiles(w, autoconnect=True) + await w._set_wired_profiles_autoconnect(False) + assert "timestamp" not in conn.update.await_args[0][0]["connection"] + + @pytest.mark.asyncio + async def test_reenables_wired_profile(self, qapp): + w = _make_worker(qapp) + conn = self._wire_profiles(w, autoconnect=False) + await w._set_wired_profiles_autoconnect(True) + assert conn.update.await_args[0][0]["connection"]["autoconnect"] == ("b", True) + + @pytest.mark.asyncio + async def test_skips_when_already_correct(self, qapp): + w = _make_worker(qapp) + conn = self._wire_profiles(w, autoconnect=True) + await w._set_wired_profiles_autoconnect(True) + conn.update.assert_not_awaited() + + @pytest.mark.asyncio + async def test_ignores_non_ethernet_profiles(self, qapp): + w = _make_worker(qapp) + conn = self._wire_profiles(w, conn_type="802-11-wireless", autoconnect=True) + await w._set_wired_profiles_autoconnect(False) + conn.update.assert_not_awaited() + + @pytest.mark.asyncio + async def test_exception_is_non_fatal(self, qapp): + w = _make_worker(qapp) + w._nm_settings = MagicMock(side_effect=RuntimeError("boom")) + await w._set_wired_profiles_autoconnect(False) # must not raise + + +class TestEnsureWiredAutoconnect: + def test_no_wired_device_returns_early(self, qapp): + w = _make(qapp, wired=False) + wired = AsyncProxyMock(state=30, autoconnect=False) + _wire(w, wired_proxy=wired) + _run(w._ensure_wired_autoconnect()) + wired.autoconnect.set_async.assert_not_awaited() + + def test_autoconnect_off_is_rearmed(self, qapp): + w = _make(qapp) + wired = AsyncProxyMock(state=30, autoconnect=False) + _wire(w, wired_proxy=wired) + _run(w._ensure_wired_autoconnect()) + wired.autoconnect.set_async.assert_awaited_once_with(True) + + def test_autoconnect_on_is_left_alone(self, qapp): + w = _make(qapp) + wired = AsyncProxyMock(state=100, autoconnect=True) + _wire(w, wired_proxy=wired) + _run(w._ensure_wired_autoconnect()) + wired.autoconnect.set_async.assert_not_awaited() + + def test_profiles_are_rearmed_too(self, qapp): + w = _make(qapp) + w._set_wired_profiles_autoconnect = AsyncMock() + _wire(w, wired_proxy=AsyncProxyMock(state=30, autoconnect=False)) + _run(w._ensure_wired_autoconnect()) + w._set_wired_profiles_autoconnect.assert_awaited_once_with(True) + + def test_exception_is_non_fatal(self, qapp): + w = _make(qapp) + w._generic = MagicMock(side_effect=RuntimeError("boom")) + _run(w._ensure_wired_autoconnect()) # must not raise + + def test_wifi_radio_is_never_touched(self, qapp): + w = _make(qapp) + nm = AsyncProxyMock(wireless_enabled=True) + _wire(w, nm=nm) + wired = AsyncProxyMock(state=30, autoconnect=False) + _wire(w, wired_proxy=wired) + _run(w._ensure_wired_autoconnect()) + nm.wireless_enabled.set_async.assert_not_awaited() + + class TestWaitForWifiRadio: def test_returns_true_when_already_matching(self, qapp): w = _make(qapp) @@ -2083,15 +2332,18 @@ def test_disable_wifi_happy_path(self, qapp): assert received[0].success is True assert w._is_hotspot_active is False - def test_enable_wifi_leaves_ethernet_untouched(self, qapp): + def test_enable_wifi_leaves_ethernet_up(self, qapp): + """Wi-Fi is the recovery path; enabling it must never drop a live cable.""" w = _make(qapp) nm = AsyncProxyMock(wireless_enabled=False) _wire(w, nm=nm) w._is_ethernet_connected = AsyncMock(return_value=True) + w._async_disconnect_ethernet = AsyncMock() w._wait_for_wifi_radio = AsyncMock(return_value=True) w._build_current_state = AsyncMock(return_value=NetworkState()) _run(w._async_set_wifi_enabled(True)) + w._async_disconnect_ethernet.assert_not_awaited() nm.wireless_enabled.set_async.assert_awaited_once_with(True) def test_already_matching_skips_toggle(self, qapp): @@ -2140,6 +2392,54 @@ def test_calls_disconnect(self, qapp): wired.disconnect.assert_awaited_once() w._deactivate_all_vlans.assert_awaited_once() + def test_persists_choice_in_the_profile(self, qapp): + w = _make(qapp) + wired = AsyncProxyMock() + wired.disconnect = AsyncMock() + _wire(w, wired_proxy=wired) + w._is_ethernet_connected = AsyncMock(return_value=False) + w._deactivate_all_vlans = AsyncMock() + w._set_wired_profiles_autoconnect = AsyncMock() + _run(w._async_disconnect_ethernet()) + w._set_wired_profiles_autoconnect.assert_awaited_once_with(False) + + def test_already_inactive_is_not_an_error(self, qapp): + w = _make(qapp) + wired = AsyncProxyMock() + wired.disconnect = AsyncMock( + side_effect=RuntimeError("This device is not active") + ) + _wire(w, wired_proxy=wired) + w._is_ethernet_connected = AsyncMock(return_value=False) + w._deactivate_all_vlans = AsyncMock() + with patch.object(_worker_mod.logger, "error") as err: + _run(w._async_disconnect_ethernet()) + err.assert_not_called() + + def test_persists_choice_even_when_already_inactive(self, qapp): + w = _make(qapp) + wired = AsyncProxyMock() + wired.disconnect = AsyncMock( + side_effect=RuntimeError("This device is not active") + ) + _wire(w, wired_proxy=wired) + w._is_ethernet_connected = AsyncMock(return_value=False) + w._deactivate_all_vlans = AsyncMock() + w._set_wired_profiles_autoconnect = AsyncMock() + _run(w._async_disconnect_ethernet()) + w._set_wired_profiles_autoconnect.assert_awaited_once_with(False) + + def test_persists_choice_even_when_teardown_fails(self, qapp): + w = _make(qapp) + wired = AsyncProxyMock() + wired.disconnect = AsyncMock(side_effect=RuntimeError("boom")) + _wire(w, wired_proxy=wired) + w._is_ethernet_connected = AsyncMock(return_value=False) + w._deactivate_all_vlans = AsyncMock() + w._set_wired_profiles_autoconnect = AsyncMock() + _run(w._async_disconnect_ethernet()) + w._set_wired_profiles_autoconnect.assert_awaited_once_with(False) + class TestConnectEthernetAsync: def test_no_wired_path_emits_error(self, qapp): @@ -2154,15 +2454,22 @@ def test_happy_path(self, qapp): nm = AsyncProxyMock(wireless_enabled=True) nm.activate_connection = AsyncMock() _wire(w, nm=nm) - w._ensure_wired_autoconnect = AsyncMock() + wifi = AsyncProxyMock() + wifi.disconnect = AsyncMock() + _wire(w, wifi_proxy=wifi) + w._is_ethernet_connected = AsyncMock(return_value=False) + w._wait_for_wifi_radio = AsyncMock(return_value=True) w._build_current_state = AsyncMock(return_value=NetworkState()) w._activate_saved_vlans = AsyncMock() + w._ensure_wired_autoconnect = AsyncMock() w._is_hotspot_active = False results = [] w.connection_result.connect(results.append) _run(w._async_connect_ethernet()) + # Wi-Fi is the recovery path; connecting a cable must never kill the radio. + nm.wireless_enabled.set_async.assert_not_awaited() w._ensure_wired_autoconnect.assert_awaited_once() nm.activate_connection.assert_awaited_once() assert len(results) == 1 @@ -2170,8 +2477,9 @@ def test_happy_path(self, qapp): def test_exception_emits_error_and_state(self, qapp): w = _make(qapp) - nm = AsyncProxyMock(wireless_enabled=True) - nm.activate_connection = AsyncMock(side_effect=RuntimeError("x")) + nm = AsyncProxyMock( + activate_connection=AsyncMock(side_effect=RuntimeError("x")) + ) w._nm = _ProxyFactory(nm) w._ensure_wired_autoconnect = AsyncMock() w._build_current_state = AsyncMock(return_value=NetworkState()) @@ -2558,7 +2866,7 @@ async def _test(): w._listen_wifi_state_changed = AsyncMock() w._listen_settings_new_connection = AsyncMock() w._listen_settings_connection_removed = AsyncMock() - # _resilient_listener wraps them — mock it to just return + # _resilient_listener wraps them: mock it to just return w._resilient_listener = AsyncMock() w._track_task = MagicMock() await w._start_signal_listeners() @@ -2574,6 +2882,7 @@ class TestAsyncInitializeFull: def test_happy_path_full_init(self, qapp): w = _make(qapp, running=False) w._detect_interfaces = AsyncMock() + w._ensure_wired_autoconnect = AsyncMock() w._is_ethernet_connected = AsyncMock(return_value=False) w._activate_saved_vlans = AsyncMock() w._start_signal_listeners = AsyncMock() @@ -2590,6 +2899,8 @@ def test_happy_path_full_init(self, qapp): assert w._running is True w._detect_interfaces.assert_awaited_once() + # Boot must not re-arm: NM's latch is how "ethernet off" survives a reboot. + w._ensure_wired_autoconnect.assert_not_awaited() w._start_signal_listeners.assert_awaited_once() assert len(init_signals) == 1 assert len(hotspot_info) == 1 @@ -2775,3 +3086,123 @@ def test_malformed_entry_skipped_returns_valid_entries(self, qapp): result = _run(w._get_saved_networks_impl()) assert len(result) == 1 assert result[0].ssid == "GoodNet" + + +class TestValidatePsk: + """WPA passphrase length rules NM enforces, checked before we touch a profile.""" + + def test_seven_chars_rejected(self): + result = NetworkManagerWorker._validate_psk("1234567") + assert result is not None + assert result.success is False + assert result.error_code == "invalid_password_length" + + def test_eight_chars_accepted(self): + assert NetworkManagerWorker._validate_psk("12345678") is None + + def test_sixty_three_chars_accepted(self): + assert NetworkManagerWorker._validate_psk("a" * 63) is None + + def test_sixty_four_hex_accepted(self): + assert NetworkManagerWorker._validate_psk("a" * 64) is None + + def test_sixty_four_uppercase_hex_accepted(self): + assert NetworkManagerWorker._validate_psk("ABCDEF01" * 8) is None + + def test_sixty_four_non_hex_rejected(self): + assert NetworkManagerWorker._validate_psk("z" * 64) is not None + + def test_sixty_five_chars_rejected(self): + assert NetworkManagerWorker._validate_psk("a" * 65) is not None + + def test_empty_rejected(self): + assert NetworkManagerWorker._validate_psk("") is not None + + +class TestPrefixToMask: + """Prefix length to dotted-decimal mask, with out-of-range guarded.""" + + def test_prefix_24(self): + assert NetworkManagerWorker._prefix_to_mask(24) == "255.255.255.0" + + def test_prefix_16(self): + assert NetworkManagerWorker._prefix_to_mask(16) == "255.255.0.0" + + def test_prefix_8(self): + assert NetworkManagerWorker._prefix_to_mask(8) == "255.0.0.0" + + def test_prefix_0(self): + assert NetworkManagerWorker._prefix_to_mask(0) == "0.0.0.0" + + def test_prefix_32(self): + assert NetworkManagerWorker._prefix_to_mask(32) == "255.255.255.255" + + def test_negative_returns_empty(self): + assert NetworkManagerWorker._prefix_to_mask(-1) == "" + + def test_above_32_returns_empty(self): + assert NetworkManagerWorker._prefix_to_mask(33) == "" + + +class TestParseIpv4Settings: + """NM ipv4 settings dicts arrive in two shapes; both must parse.""" + + @staticmethod + def _uint(ip: str) -> int: + return NetworkManagerWorker._ip_to_nm_uint32(ip) + + def test_legacy_addresses_shape(self): + ipv4 = { + "addresses": ("aau", [[self._uint("192.168.1.50"), 24, 0]]), + "gateway": ("s", "192.168.1.1"), + "dns": ("au", [self._uint("8.8.8.8")]), + } + addr, mask, gw, dns = NetworkManagerWorker._parse_ipv4_settings(ipv4) + assert addr == "192.168.1.50" + assert mask == "255.255.255.0" + assert gw == "192.168.1.1" + assert dns == ("8.8.8.8",) + + def test_address_data_shape(self): + ipv4 = { + "addresses": ("aau", []), + "address-data": ( + "aa{sv}", + [{"address": ("s", "10.0.0.7"), "prefix": ("u", 16)}], + ), + "gateway": ("s", "10.0.0.1"), + "dns-data": ("as", ["1.1.1.1", "9.9.9.9"]), + } + addr, mask, gw, dns = NetworkManagerWorker._parse_ipv4_settings(ipv4) + assert addr == "10.0.0.7" + assert mask == "255.255.0.0" + assert gw == "10.0.0.1" + assert dns == ("1.1.1.1", "9.9.9.9") + + def test_dns_data_wins_over_legacy_dns(self): + ipv4 = { + "dns-data": ("as", ["1.1.1.1"]), + "dns": ("au", [self._uint("8.8.8.8")]), + } + _, _, _, dns = NetworkManagerWorker._parse_ipv4_settings(ipv4) + assert dns == ("1.1.1.1",) + + def test_empty_dict_yields_blanks(self): + assert NetworkManagerWorker._parse_ipv4_settings({}) == ("", "", "", ()) + + def test_zero_prefix_yields_blank_mask(self): + ipv4 = {"addresses": ("aau", [[self._uint("192.168.1.50"), 0, 0]])} + addr, mask, _, _ = NetworkManagerWorker._parse_ipv4_settings(ipv4) + assert addr == "192.168.1.50" + assert mask == "" + + def test_malformed_addresses_do_not_raise(self): + ipv4 = {"addresses": ("aau", [["not-an-int"]]), "gateway": ("s", "192.168.1.1")} + addr, mask, gw, _ = NetworkManagerWorker._parse_ipv4_settings(ipv4) + assert (addr, mask) == ("", "") + assert gw == "192.168.1.1" + + def test_malformed_dns_does_not_raise(self): + ipv4 = {"dns": ("au", ["not-an-int"])} + _, _, _, dns = NetworkManagerWorker._parse_ipv4_settings(ipv4) + assert dns == ()