diff --git a/BlocksScreen/lib/network/manager.py b/BlocksScreen/lib/network/manager.py index 22270d85..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""" @@ -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 f01c9030..aa995a69 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 = "" @@ -120,6 +131,8 @@ def __init__(self) -> None: self._rediscover_lock = asyncio.Lock() self._rediscover_gen = 0 self._stale_logged_gen = -1 + # Wi-Fi FAILED count, latched by the StateChanged listener. + self._wifi_failures = 0 self._asyncio_loop: asyncio.AbstractEventLoop = asyncio.new_event_loop() self._asyncio_thread = threading.Thread( @@ -194,9 +207,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() @@ -360,14 +371,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)", @@ -410,6 +422,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) @@ -446,7 +466,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 @@ -455,8 +475,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. @@ -610,7 +631,7 @@ 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: @@ -655,7 +676,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: @@ -670,6 +691,9 @@ async def _listen_wifi_state_changed(self) -> None: new_state, reason, ) + if new_state == 120: + self._wifi_failures += 1 + logger.info("Wi-Fi activation failed (reason %d)", reason) self._schedule_debounced_state_rebuild() async def _listen_settings_new_connection(self) -> None: @@ -720,7 +744,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( @@ -744,7 +768,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( @@ -828,6 +852,16 @@ async def _is_ethernet_connected(self) -> bool: logger.debug("Error checking ethernet state: %s", exc) return False + 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). @@ -840,7 +874,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: @@ -928,7 +962,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: @@ -948,20 +982,30 @@ 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_ssid = "" current_ip, eth_connected = await self._resolve_current_ip( - "wlan0", current_ssid, eth_connected, connectivity + wifi_iface, current_ssid, bool(eth_connected), connectivity ) signal, sec_type = await self._wifi_signal_and_security(current_ssid) - 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 @@ -970,8 +1014,21 @@ async def _build_current_state(self) -> NetworkState: 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, @@ -981,8 +1038,8 @@ 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) @@ -998,6 +1055,7 @@ async def _resolve_current_ip( """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) @@ -1021,10 +1079,13 @@ async def _resolve_current_ip( return current_ip, eth_connected async def _wifi_signal_and_security(self, ssid: str) -> tuple[int, str]: - """Return the scanned signal strength and saved security type for ``ssid``.""" + """Return the live signal strength and saved security type for ``ssid``.""" if not ssid: return 0, "" - signal = (await self._build_signal_map()).get(ssid.lower(), 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 "" @@ -1036,15 +1097,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) @@ -1098,6 +1165,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 "" @@ -1250,27 +1321,53 @@ async def _gather_ap_properties( ) return list(zip(ap_paths, props)) + 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( @@ -1315,6 +1412,42 @@ 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, key_mgmt: str) -> ConnectionResult | None: + """Reject a psk NM would refuse, so we fail before touching the profile.""" + # Mirrors NM verify_secrets: bytes are counted and SAE has no length rule. + if key_mgmt == "sae": + return None + size = len(password.encode()) + if PSK_MIN_LENGTH <= size <= PSK_MAX_LENGTH: + return None + if size == 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.""" @@ -1484,13 +1617,14 @@ async def _add_network_impl( 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") - 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: @@ -1507,6 +1641,14 @@ async def _add_network_impl( "unsupported_security", ) + # Guard before the backup/delete below, so a bad psk cannot cost the profile. + sec = conn_props.get("802-11-wireless-security") + if isinstance(sec, dict) and ( + bad := self._validate_psk(password, sec["key-mgmt"][1]) + ): + logger.info("add_network: '%s' rejected, psk len=%d", ssid, len(password)) + return bad + # Drop the old profile only once the new one is buildable, so early exits keep it. backup = await self._backup_and_drop_existing(ssid) try: @@ -1515,14 +1657,9 @@ async def _add_network_impl( except Exception as exc: if backup: await self._restore_profile(ssid, backup) - 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() @@ -1530,6 +1667,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) @@ -1553,6 +1691,7 @@ async def _backup_profile(self, ssid: str) -> dict | None: 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 @@ -1627,27 +1766,40 @@ 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 once the Wi-Fi device reports FAILED, as nmcli does. """ loop = asyncio.get_running_loop() - deadline = loop.time() + timeout + started = loop.time() + deadline = started + timeout + failures = self._wifi_failures + logger.info("wait_for_connection: '%s' timeout=%.1fs", ssid, timeout) await asyncio.sleep(1.5) - consecutive_empty = 0 while loop.time() < deadline: + # NM leaves FAILED on its next idle tick, so polling the state misses it. + if self._wifi_failures != failures: + logger.warning( + "wait_for_connection: '%s' failed after %.1fs", + ssid, + loop.time() - started, + ) + return False try: current = await self._get_current_ssid() if current and current.lower() == ssid.lower(): - ip = await self._get_current_ip() + # Wi-Fi's own IP: the primary may be ethernet or a VPN. + ip = await self._get_ip_by_interface(self._get_wifi_iface_name()) 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 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", ssid, timeout) return False async def _connect_network_impl(self, ssid: str) -> ConnectionResult: @@ -1688,9 +1840,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() @@ -1800,10 +1951,23 @@ 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")) + if bad := self._validate_psk(password, sec["key-mgmt"][1]): + logger.info( + "update_network: '%s' rejected, psk len=%d", ssid, len(password) + ) + return bad + 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: @@ -1814,18 +1978,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.""" @@ -1938,7 +2111,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) @@ -2268,9 +2440,12 @@ async def _wait_for_profile_ip(self, ssid: str, timeout: float) -> str: current = await self._get_current_ssid() 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("wlan0") or "" + iface = self._get_wifi_iface_name() + found_ip = ( + await self._get_ip_by_interface(iface) + or self._get_ip_os_fallback(iface) + or "" + ) if found_ip: return found_ip except Exception as exc: @@ -2289,8 +2464,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: @@ -2340,14 +2525,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() @@ -2434,15 +2626,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() @@ -2450,11 +2634,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] = { @@ -2487,8 +2667,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 @@ -2586,6 +2765,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( @@ -2894,5 +3079,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 ef77b14b..b3c7ed45 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). @@ -184,12 +221,15 @@ def _init_instance_variables(self) -> None: self._is_connecting = False self._target_ssid: str | None = None 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]] = {} @@ -231,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. @@ -256,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: @@ -278,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 @@ -306,95 +350,20 @@ def _on_network_state_changed(self, state: NetworkState) -> None: # 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. @@ -416,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. @@ -432,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) @@ -458,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) @@ -599,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: @@ -619,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, @@ -630,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 ] @@ -745,7 +702,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). """ @@ -769,16 +726,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) @@ -809,7 +763,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) @@ -900,52 +854,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() @@ -1009,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.""" @@ -1147,14 +1101,6 @@ def _on_hotspot_activate(self) -> None: self._show_error_popup("Hotspot password must be at least 8 characters.") return - # Mutual exclusion: turn off Wi-Fi and Ethernet - wifi_btn = self.wifi_button.toggle_button - eth_btn = self.ethernet_button.toggle_button - with QtCore.QSignalBlocker(wifi_btn): - wifi_btn.state = wifi_btn.State.OFF - with QtCore.QSignalBlocker(eth_btn): - eth_btn.state = eth_btn.State.OFF - hotspot_btn = self.hotspot_button.toggle_button with QtCore.QSignalBlocker(hotspot_btn): hotspot_btn.state = hotspot_btn.State.ON @@ -1163,6 +1109,10 @@ def _on_hotspot_activate(self) -> None: self._pending_operation = PendingOperation.HOTSPOT_ON self.setCurrentIndex(self.indexOf(self.main_network_page)) self._set_loading_state(True) + + # The worker keeps the cable for a hotspot, so a user activate drops it here. + self._claim_link(self.hotspot_button) + self._nm.disconnect_ethernet() self._nm.create_hotspot(new_name, new_password, "wpa-psk") def _show_hotspot_qr(self, ssid: str, password: str, security: str) -> None: @@ -1177,6 +1127,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 ( @@ -1187,6 +1138,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() @@ -1228,31 +1180,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( @@ -1261,12 +1218,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() @@ -1299,10 +1257,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)) @@ -1319,7 +1278,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) @@ -1374,8 +1333,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 @@ -1513,8 +1472,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 @@ -1522,9 +1494,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 @@ -1571,10 +1541,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() @@ -1584,14 +1550,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: @@ -1634,6 +1592,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() @@ -1652,12 +1611,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). @@ -1668,7 +1629,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: @@ -1677,17 +1638,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() + @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() @@ -2568,6 +2539,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) @@ -3303,7 +3307,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() ) @@ -3562,6 +3566,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 ( @@ -3751,12 +3756,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: @@ -3765,6 +3772,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) @@ -3828,6 +3836,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..e6f607cd 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``. @@ -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 d740274a..15b3d886 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( @@ -484,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 # ───────────────────────────────────────────────────────────────────────────── @@ -534,7 +545,7 @@ def test_ethernet_plug_keeps_wifi_enabled(self, win): # ───────────────────────────────────────────────────────────────────────────── -# _on_network_state_changed — loading state machine +# _on_network_state_changed: loading state machine # ───────────────────────────────────────────────────────────────────────────── @@ -576,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): @@ -702,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 @@ -757,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): @@ -784,6 +808,49 @@ 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 TestOnHotspotActivate: + """The hotspot page's activate button follows the same one-link rule as the toggle.""" + + @staticmethod + def _activate(w, name="MyHotspot", password="validpass"): + w.hotspot_name_input_field.setText(name) + w.hotspot_password_input_field.setText(password) + w._on_hotspot_activate() + + def test_creates_hotspot_and_drops_ethernet(self, win): + w, nm = win + self._activate(w) + nm.disconnect_ethernet.assert_called_once() + nm.create_hotspot.assert_called_once_with("MyHotspot", "validpass", "wpa-psk") + assert w._pending_operation == PendingOperation.HOTSPOT_ON + + def test_turns_other_toggles_off(self, win): + w, nm = win + _all_on(w) + self._activate(w) + assert _off(w.wifi_button) and _off(w.ethernet_button) + assert not _off(w.hotspot_button) + + def test_short_password_touches_nothing(self, win): + w, nm = win + with patch.object(w, "_show_error_popup") as mock_err: + self._activate(w, password="short") + mock_err.assert_called_once() + nm.disconnect_ethernet.assert_not_called() + nm.create_hotspot.assert_not_called() + class TestEthernetToggle: def test_ethernet_on_calls_connect_ethernet(self, win): @@ -806,6 +873,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 @@ -937,13 +1020,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 @@ -967,7 +1050,7 @@ def _stub_init(self, *_a, **_kw): # ───────────────────────────────────────────────────────────────────────────── -# Step 4a: Helper classes — PixmapCache, WifiIconProvider, IPAddressLineEdit +# Step 4a: Helper classes PixmapCache, WifiIconProvider, IPAddressLineEdit # ───────────────────────────────────────────────────────────────────────────── @@ -1883,3 +1966,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_worker_helpers_unit.py b/tests/network/test_worker_helpers_unit.py index 95052235..9459d315 100644 --- a/tests/network/test_worker_helpers_unit.py +++ b/tests/network/test_worker_helpers_unit.py @@ -147,9 +147,17 @@ class TestWaitForProfileIp: async def test_returns_ip_once_ssid_matches(self, qapp): w = _worker(qapp) w._get_current_ssid = AsyncMock(side_effect=["Other", "home"]) - w._get_current_ip.return_value = "10.0.0.7" + w._get_ip_by_interface.return_value = "10.0.0.7" assert await w._wait_for_profile_ip("Home", timeout=5.0) == "10.0.0.7" assert w._get_current_ssid.await_count == 2 + w._get_ip_by_interface.assert_awaited_with("wlan0") + + @pytest.mark.asyncio + async def test_primary_ethernet_ip_does_not_count(self, qapp): + w = _worker(qapp) + w._get_current_ssid = AsyncMock(return_value="Home") + w._get_current_ip.return_value = "192.168.0.10" + assert await w._wait_for_profile_ip("Home", timeout=0.01) == "" @pytest.mark.asyncio async def test_os_fallback_when_nm_has_no_ip(self, qapp): @@ -162,7 +170,7 @@ async def test_os_fallback_when_nm_has_no_ip(self, qapp): async def test_lookup_error_is_retried(self, qapp): w = _worker(qapp) w._get_current_ssid = AsyncMock(side_effect=[RuntimeError("dbus"), "Home"]) - w._get_current_ip.return_value = "10.0.0.9" + w._get_ip_by_interface.return_value = "10.0.0.9" assert await w._wait_for_profile_ip("Home", timeout=5.0) == "10.0.0.9" @pytest.mark.asyncio diff --git a/tests/network/test_worker_unit.py b/tests/network/test_worker_unit.py index bd0abe5d..836f35ed 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 @@ -65,11 +65,15 @@ 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._wifi_failures = 0 + w._system_bus = MagicMock(name="mock_system_bus") w._primary_wifi_path = ( "/org/freedesktop/NetworkManager/Devices/2" if with_wifi else "" ) @@ -94,9 +98,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() @@ -106,15 +107,19 @@ 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._wifi_failures = 0 + w._system_bus = None w._primary_wifi_path = "" w._primary_wifi_iface = "" w._primary_wired_path = "" @@ -134,9 +139,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 @@ -149,8 +151,12 @@ 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._wifi_failures = 0 + 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 "" @@ -170,14 +176,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: @@ -268,6 +285,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._ensure_wired_autoconnect = AsyncMock() w._is_ethernet_connected = AsyncMock(return_value=False) w._activate_saved_vlans = AsyncMock() w._start_signal_listeners = AsyncMock() @@ -609,18 +627,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 = [] @@ -735,12 +768,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) @@ -751,17 +798,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) @@ -837,7 +931,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=[], @@ -856,7 +950,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) @@ -888,7 +982,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) @@ -904,13 +998,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) @@ -930,7 +1044,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) @@ -948,7 +1062,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() @@ -1180,6 +1295,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): @@ -1348,7 +1487,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 @@ -1623,10 +1762,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): @@ -1814,7 +1953,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 @@ -1840,7 +1979,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" @@ -1870,7 +2009,7 @@ class TestAddNetworkBackup: async def test_not_found_keeps_saved_profile(self, qapp): w = _add_ready_worker(qapp) w._find_ap_props = AsyncMock(return_value=None) - result = await w._add_network_impl("Net", "pass", 0) + result = await w._add_network_impl("Net", "password", 0) assert result.error_code == "not_found" w._backup_and_drop_existing.assert_not_awaited() @@ -1878,7 +2017,7 @@ async def test_not_found_keeps_saved_profile(self, qapp): async def test_unsupported_security_keeps_saved_profile(self, qapp): w = _add_ready_worker(qapp) w._build_connection_properties = MagicMock(return_value={}) - result = await w._add_network_impl("Net", "pass", 0) + result = await w._add_network_impl("Net", "password", 0) assert result.error_code == "unsupported_security" w._backup_and_drop_existing.assert_not_awaited() @@ -1886,7 +2025,7 @@ async def test_unsupported_security_keeps_saved_profile(self, qapp): async def test_add_failure_restores_backup(self, qapp): w = _add_ready_worker(qapp) w._nm_settings().add_connection.side_effect = RuntimeError("boom") - result = await w._add_network_impl("Net", "pass", 0) + result = await w._add_network_impl("Net", "password", 0) assert result.error_code == "add_failed" w._restore_profile.assert_awaited_once_with("Net", _BACKUP) @@ -1894,7 +2033,7 @@ async def test_add_failure_restores_backup(self, qapp): async def test_add_failure_without_backup_restores_nothing(self, qapp): w = _add_ready_worker(qapp, backup=None) w._nm_settings().add_connection.side_effect = RuntimeError("boom") - result = await w._add_network_impl("Net", "pass", 0) + result = await w._add_network_impl("Net", "password", 0) assert result.error_code == "add_failed" w._restore_profile.assert_not_awaited() @@ -1902,7 +2041,7 @@ async def test_add_failure_without_backup_restores_nothing(self, qapp): async def test_activation_timeout_restores_backup(self, qapp): w = _add_ready_worker(qapp) w._wait_for_connection = AsyncMock(return_value=False) - result = await w._add_network_impl("Net", "pass", 0) + result = await w._add_network_impl("Net", "password", 0) assert result.error_code == "auth_failed" assert "previously saved password was kept" in result.message w._delete_network_impl.assert_awaited_once_with("Net") @@ -1912,7 +2051,7 @@ async def test_activation_timeout_restores_backup(self, qapp): async def test_activation_timeout_without_backup_reports_removal(self, qapp): w = _add_ready_worker(qapp, backup=None) w._wait_for_connection = AsyncMock(return_value=False) - result = await w._add_network_impl("Net", "pass", 0) + result = await w._add_network_impl("Net", "password", 0) assert result.error_code == "auth_failed" assert "saved profile has been removed" in result.message w._restore_profile.assert_not_awaited() @@ -1922,13 +2061,13 @@ async def test_failed_restore_reports_removal(self, qapp): w = _add_ready_worker(qapp) w._wait_for_connection = AsyncMock(return_value=False) w._restore_profile = AsyncMock(return_value=False) - result = await w._add_network_impl("Net", "pass", 0) + result = await w._add_network_impl("Net", "password", 0) assert "saved profile has been removed" in result.message @pytest.mark.asyncio async def test_success_leaves_new_profile(self, qapp): w = _add_ready_worker(qapp) - result = await w._add_network_impl("Net", "pass", 0) + result = await w._add_network_impl("Net", "password", 0) assert result.success w._delete_network_impl.assert_not_awaited() w._restore_profile.assert_not_awaited() @@ -2072,7 +2211,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( @@ -2106,6 +2245,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): @@ -2113,15 +2256,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 == [] @@ -2417,15 +2561,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): @@ -2566,15 +2713,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 @@ -2582,8 +2736,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()) @@ -2970,7 +3125,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() @@ -2986,6 +3141,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() @@ -3002,6 +3158,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 @@ -3120,27 +3278,69 @@ def test_activate_connection_exception_returns_failure(self, qapp): class TestWaitForConnection: - """_wait_for_connection covers timeout and consecutive-empty early-exit.""" + """_wait_for_connection: Wi-Fi's own IP means up, a latched FAILED ends it early.""" + + @staticmethod + def _fast(w, timeout=10.0): + """Run the wait with every sleep collapsed to one loop tick.""" + real_sleep = asyncio.sleep + with patch.object(_worker_mod.asyncio, "sleep", lambda _d: real_sleep(0)): + return _run(w._wait_for_connection("HomeNet", timeout=timeout)) - def test_wait_for_connection_consecutive_empty_stops_early(self, qapp): - """Three consecutive empty SSID polls triggers early False return.""" + def test_latched_failure_stops_early(self, qapp): w = _make_worker(qapp) - # Return empty SSID three times → consecutive_empty hits 3 - w._get_current_ssid = AsyncMock(return_value="") - w._get_current_ip = AsyncMock(return_value="") - result = _run(w._wait_for_connection("HomeNet", timeout=10.0)) - assert result is False - def test_wait_for_connection_inner_exception_continues(self, qapp): - """Exceptions inside the loop are silently swallowed (pass block).""" + async def ssid(): + w._wifi_failures += 1 # the listener saw FAILED between polls + return "" + + w._get_current_ssid = AsyncMock(side_effect=ssid) + assert self._fast(w) is False + w._get_current_ssid.assert_awaited_once() + + def test_failure_before_the_wait_is_ignored(self, qapp): + w = _make_worker(qapp) + w._wifi_failures = 3 + w._get_current_ssid = AsyncMock(return_value="HomeNet") + w._get_ip_by_interface = AsyncMock(return_value="192.168.1.20") + assert self._fast(w) is True + + def test_ip_is_read_from_the_wifi_interface(self, qapp): + w = _make_worker(qapp) + w._get_current_ssid = AsyncMock(return_value="homenet") + w._get_ip_by_interface = AsyncMock(return_value="192.168.1.20") + assert self._fast(w) is True + w._get_ip_by_interface.assert_awaited_with("wlan0") + + def test_primary_ethernet_ip_does_not_count(self, qapp): + w = _make_worker(qapp) + w._get_current_ssid = AsyncMock(return_value="HomeNet") + w._get_current_ip = AsyncMock(return_value="10.0.0.2") + w._get_ip_by_interface = AsyncMock(return_value="") + assert self._fast(w, timeout=0.05) is False + + def test_inner_exception_keeps_polling(self, qapp): w = _make_worker(qapp) - # Raise on first call, then return empty (hits consecutive_empty path) w._get_current_ssid = AsyncMock( - side_effect=[RuntimeError("transient"), "", "", ""] + side_effect=[RuntimeError("transient"), "HomeNet"] ) - w._get_current_ip = AsyncMock(return_value="") - result = _run(w._wait_for_connection("HomeNet", timeout=10.0)) - assert result is False + w._get_ip_by_interface = AsyncMock(return_value="192.168.1.20") + assert self._fast(w) is True + + +class TestWifiFailureLatch: + def test_failed_transition_is_counted(self, qapp): + w = _make_worker(qapp) + w._schedule_debounced_state_rebuild = MagicMock() + + async def transitions(): + for step in ((50, 40, 0), (120, 60, 7), (30, 120, 0)): + yield step + + w._signal_wifi = SimpleNamespace(state_changed=transitions()) + _run(w._listen_wifi_state_changed()) + assert w._wifi_failures == 1 + assert w._schedule_debounced_state_rebuild.call_count == 3 class TestGetSavedNetworksHandlesMalformedEntry: @@ -3189,6 +3389,201 @@ def test_malformed_entry_skipped_returns_valid_entries(self, qapp): assert result[0].ssid == "GoodNet" +class TestValidatePsk: + """NM's verify_secrets psk rule, checked before we touch a profile.""" + + @staticmethod + def _check(password, key_mgmt="wpa-psk"): + return NetworkManagerWorker._validate_psk(password, key_mgmt) + + def test_seven_chars_rejected(self): + result = self._check("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 self._check("12345678") is None + + def test_sixty_three_chars_accepted(self): + assert self._check("a" * 63) is None + + def test_sixty_four_hex_accepted(self): + assert self._check("a" * 64) is None + + def test_sixty_four_uppercase_hex_accepted(self): + assert self._check("ABCDEF01" * 8) is None + + def test_sixty_four_non_hex_rejected(self): + assert self._check("z" * 64) is not None + + def test_sixty_five_chars_rejected(self): + assert self._check("a" * 65) is not None + + def test_empty_rejected(self): + assert self._check("") is not None + + def test_length_counts_utf8_bytes(self): + # 62 characters but 64 non-hex bytes, which NM refuses. + assert self._check("é" * 2 + "a" * 60) is not None + + def test_sae_has_no_length_rule(self): + assert self._check("short", "sae") is None + + +def _secured_add_worker(qapp, key_mgmt): + """Add-ready worker whose built profile carries *key_mgmt* security.""" + w = _add_ready_worker(qapp) + w._build_connection_properties = MagicMock( + return_value={"802-11-wireless-security": {"key-mgmt": ("s", key_mgmt)}} + ) + return w + + +def _update_ready_worker(qapp, key_mgmt): + """Worker with one saved *key_mgmt* profile, returning it plus the settings proxy.""" + w = _make_worker(qapp) + w._get_connection_path = AsyncMock(return_value="/conn/1") + conn = AsyncProxyMock( + get_settings=AsyncMock( + return_value={ + "connection": {"id": ("s", "Net")}, + "802-11-wireless-security": {"key-mgmt": ("s", key_mgmt)}, + } + ), + get_secrets=AsyncMock(return_value={}), + update=AsyncMock(), + ) + w._conn_settings = lambda _path: conn + return w, conn + + +class TestPskGuardPlacement: + """The psk check runs once key-mgmt is known and before any profile change.""" + + @pytest.mark.asyncio + async def test_add_short_wpa_psk_rejected_before_backup(self, qapp): + w = _secured_add_worker(qapp, "wpa-psk") + result = await w._add_network_impl("Net", "short", 0) + assert result.error_code == "invalid_password_length" + w._backup_and_drop_existing.assert_not_awaited() + + @pytest.mark.asyncio + async def test_add_short_sae_password_accepted(self, qapp): + w = _secured_add_worker(qapp, "sae") + result = await w._add_network_impl("Net", "short", 0) + assert result.success is True + + @pytest.mark.asyncio + async def test_add_open_network_ignores_password(self, qapp): + w = _add_ready_worker(qapp) + result = await w._add_network_impl("Net", "x", 0) + assert result.success is True + + @pytest.mark.asyncio + async def test_update_short_wpa_psk_rejected_without_writing(self, qapp): + w, conn = _update_ready_worker(qapp, "wpa-psk") + result = await w._update_network_impl("Net", "short", None) + assert result.error_code == "invalid_password_length" + conn.update.assert_not_awaited() + + @pytest.mark.asyncio + async def test_update_short_sae_password_written(self, qapp): + w, conn = _update_ready_worker(qapp, "sae") + result = await w._update_network_impl("Net", "short", None) + assert result.success is True + conn.update.assert_awaited_once() + + +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 == () + + def _stub_redetect(w, new_paths): """Make every path look stale; re-detection then finds *new_paths* (wifi, wired).""" w._primary_paths_alive = AsyncMock(return_value=False)