From ca87bb42be69e1fdca70069b19a92a623b52beea Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Fri, 28 Aug 2026 11:20:13 +0100 Subject: [PATCH 1/3] fix(network): enforce one active link and persist ethernet-off intent --- BlocksScreen/lib/network/worker.py | 159 ++++++++++++++--------- BlocksScreen/lib/panels/networkWindow.py | 67 +++++----- tests/network/test_network_ui.py | 4 +- tests/network/test_worker_unit.py | 58 +-------- 4 files changed, 137 insertions(+), 151 deletions(-) diff --git a/BlocksScreen/lib/network/worker.py b/BlocksScreen/lib/network/worker.py index c708f076..b983dd87 100644 --- a/BlocksScreen/lib/network/worker.py +++ b/BlocksScreen/lib/network/worker.py @@ -284,11 +284,10 @@ def hotspot_password(self) -> str: async def _async_initialize(self) -> None: """Bootstrap the worker on the asyncio thread. - Detects network interfaces, enforces the boot-time ethernet/Wi-Fi - mutual exclusion, activates any saved VLANs if ethernet is present, - triggers an initial Wi-Fi scan, and starts all D-Bus signal listeners. - Emits ``initialized`` when done (even on failure, so the manager can - unblock its caller). + Detects network interfaces, activates any saved VLANs if ethernet is + present, triggers an initial Wi-Fi scan, and starts all D-Bus signal + listeners. Emits ``initialized`` when done (even on failure, so the + manager can unblock its caller). """ try: if not self._system_bus: @@ -297,7 +296,6 @@ async def _async_initialize(self) -> None: self._running = True await self._detect_interfaces() - await self._enforce_boot_mutual_exclusion() if await self._is_ethernet_connected(): await self._activate_saved_vlans() @@ -367,31 +365,39 @@ async def _detect_interfaces(self) -> None: # Ethernet-only or Wi-Fi driver still loading — log but don't alarm. logger.warning("No Wi-Fi interface detected; ethernet-only mode") - async def _enforce_boot_mutual_exclusion(self) -> None: - """Disable Wi-Fi at boot if ethernet is already connected. + async def _set_wired_profiles_autoconnect(self, enabled: bool) -> None: + """Persist autoconnect on every wired profile; Device.Autoconnect dies on NM restart.""" + try: + paths = await self._nm_settings().list_connections() + for path, settings in await self._gather_settings(list(paths)): + conn = settings.get("connection", {}) + if conn.get("type", (None, ""))[1] != "802-3-ethernet": + continue + if bool(conn.get("autoconnect", ("b", True))[1]) == enabled: + continue + props = {k: dict(v) for k, v in settings.items()} + props["connection"]["autoconnect"] = ("b", enabled) + props["connection"].pop("timestamp", None) + await self._conn_settings(path).update(props) + logger.info("Wired profile %s autoconnect -> %s", path, enabled) + except Exception as exc: + logger.warning("Wired profile autoconnect (%s) failed: %s", enabled, exc) + + async def _ensure_wired_autoconnect(self) -> None: + """Re-arm wired autoconnect on both the device and the saved profiles. - Prevents the device from simultaneously using both interfaces at - startup. If ethernet is active and the Wi-Fi radio is on, the Wi-Fi - device is disconnected and the radio is disabled, then we wait up to - 8 s for the radio to confirm it is off. Failures are logged but not - propagated — a non-fatal best-effort action at boot. + Called only when the user asks for ethernet, so autoconnect staying off + keeps meaning "user turned it off". Best-effort: never propagates. """ + if not self._primary_wired_path: + return + await self._set_wired_profiles_autoconnect(True) try: - if not await self._is_ethernet_connected(): - return - if not await self._nm().wireless_enabled: - return - logger.info("Boot: ethernet active + Wi-Fi enabled — disabling Wi-Fi") - if self._primary_wifi_path: - try: - await self._wifi().disconnect() - except Exception as exc: - logger.debug("Pre-radio-disable disconnect ignored: %s", exc) - await self._nm().wireless_enabled.set_async(False) - await self._wait_for_wifi_radio(False, timeout=8.0) - self._is_hotspot_active = False + wired = self._generic(self._primary_wired_path) + if not await wired.autoconnect: + await wired.autoconnect.set_async(True) except Exception as exc: - logger.warning("Boot mutual exclusion failed (non-fatal): %s", exc) + logger.debug("Device autoconnect re-arm ignored: %s", exc) async def _start_signal_listeners(self) -> None: """Create persistent proxies and spawn all D-Bus signal listeners. @@ -1295,9 +1301,7 @@ async def _add_network_impl( if not self._primary_wifi_path or not self._system_bus: return ConnectionResult(False, "No Wi-Fi interface", "no_interface") - if await self._is_known(ssid): - await self._delete_network_impl(ssid) - self._invalidate_saved_cache() + backup = await self._backup_and_drop_existing(ssid) try: await self._wifi().request_scan({}) @@ -1337,15 +1341,7 @@ async def _add_network_impl( try: await self._nm().activate_connection(conn_path) if not await self._wait_for_connection(ssid, timeout=_WIFI_CONNECT_TIMEOUT): - await self._delete_network_impl(ssid) - self._invalidate_saved_cache() - return ConnectionResult( - False, - f"Authentication failed for '{ssid}'.\n" - "The saved profile has been removed.\n" - "Please check the password and try again.", - "auth_failed", - ) + return await self._rollback_failed_add(ssid, backup) return ConnectionResult(True, f"Network '{ssid}' added and connecting") except Exception as act_err: logger.warning("Activate after add failed: %s", act_err) @@ -1360,13 +1356,65 @@ async def _reload_connections(self) -> None: except Exception as reload_err: logger.debug("reload_connections non-fatal: %s", reload_err) + async def _backup_profile(self, ssid: str) -> dict | None: + """Snapshot a saved profile's settings plus secrets so it can be re-added.""" + conn_path = await self._get_connection_path(ssid) + if not conn_path: + return None + try: + cs = self._conn_settings(conn_path) + settings = dict(await cs.get_settings()) + await self._merge_wifi_secrets(cs, settings) + settings.get("connection", {}).pop("timestamp", None) + logger.debug("backup_profile: '%s' sections=%s", ssid, sorted(settings)) + return settings + except Exception as exc: + logger.warning("backup_profile: could not snapshot '%s': %s", ssid, exc) + return None + + async def _restore_profile(self, ssid: str, settings: dict) -> bool: + """Re-add a backed-up profile after a failed replacement; True when restored.""" + try: + await self._nm_settings().add_connection(settings) + self._invalidate_saved_cache() + logger.info("restore_profile: '%s' restored after failed add", ssid) + return True + except Exception as exc: + logger.error("restore_profile: could not restore '%s': %s", ssid, exc) + return False + + async def _rollback_failed_add( + self, ssid: str, backup: dict | None + ) -> ConnectionResult: + """Delete the profile that never activated and restore *backup* if there is one.""" + logger.warning("add_network: '%s' never activated, rolling back", ssid) + await self._delete_network_impl(ssid) + self._invalidate_saved_cache() + if backup and await self._restore_profile(ssid, backup): + return ConnectionResult( + False, + f"Could not connect to '{ssid}'.\n" + "The previously saved password was kept.\n" + "Please check the password and try again.", + "auth_failed", + ) + return ConnectionResult( + False, + f"Authentication failed for '{ssid}'.\n" + "The saved profile has been removed.\n" + "Please check the password and try again.", + "auth_failed", + ) + async def _backup_and_drop_existing(self, ssid: str) -> dict | None: - """Return the scanned AP properties for *ssid*, or None if it is not visible.""" - ap_paths = await self._wifi().get_all_access_points() - for _ap_path, props in await self._gather_ap_properties(ap_paths): - if self._decode_ssid(props.get("ssid", b"")) == ssid: - return props - return None + """Back up and delete a saved profile for *ssid* so it can be re-added cleanly.""" + if not await self._is_known(ssid): + return None + backup = await self._backup_profile(ssid) + logger.info("add_network: replacing saved '%s' (backup=%s)", ssid, bool(backup)) + await self._delete_network_impl(ssid) + self._invalidate_saved_cache() + return backup async def _async_connect_network(self, ssid: str) -> None: """Activate an existing saved Wi-Fi profile and emit connection_result.""" @@ -1592,16 +1640,13 @@ async def _update_network_impl( return ConnectionResult(False, str(exc), "update_failed") async def _async_set_wifi_enabled(self, enabled: bool) -> None: - """Enable or disable the Wi-Fi radio, handling ethernet mutual exclusion.""" + """Enable or disable the Wi-Fi radio. Ethernet is left untouched.""" try: if not self._system_bus: return if not enabled: self._is_hotspot_active = False - if enabled and await self._is_ethernet_connected(): - await self._async_disconnect_ethernet() - current = await self._nm().wireless_enabled if current != enabled: if not enabled: @@ -1651,7 +1696,10 @@ async def _async_disconnect_ethernet(self) -> None: logger.error("Failed to disconnect ethernet: %s", exc) async def _async_connect_ethernet(self) -> None: - """Disable Wi-Fi/hotspot, activate the wired device, and restore saved VLANs.""" + """Activate the wired device and restore saved VLANs. + + Mechanism only: the one-link-at-a-time policy lives in the UI toggles. + """ if not self._primary_wired_path: self.error_occurred.emit("connect_ethernet", "No wired device found") return @@ -1659,16 +1707,7 @@ async def _async_connect_ethernet(self) -> None: if self._is_hotspot_active: await self._async_toggle_hotspot(False) - if self._primary_wifi_path: - try: - await self._wifi().disconnect() - except Exception as exc: - logger.debug("Pre-VLAN disconnect ignored: %s", exc) - await asyncio.sleep(0.5) - - if await self._nm().wireless_enabled: - await self._nm().wireless_enabled.set_async(False) - await self._wait_for_wifi_radio(False, timeout=8.0) + await self._ensure_wired_autoconnect() await self._nm().activate_connection("/", self._primary_wired_path, "/") await asyncio.sleep(1.5) diff --git a/BlocksScreen/lib/panels/networkWindow.py b/BlocksScreen/lib/panels/networkWindow.py index 7e5b44ff..e22d331d 100644 --- a/BlocksScreen/lib/panels/networkWindow.py +++ b/BlocksScreen/lib/panels/networkWindow.py @@ -37,6 +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 logger = logging.getLogger(__name__) @@ -264,7 +265,7 @@ def _on_reconnect_complete(self) -> None: def _init_timers(self) -> None: """Initialize timers.""" - self._load_timer = QtCore.QTimer(self) + self._load_timer = QTimer(self) self._load_timer.setSingleShot(True) self._load_timer.timeout.connect(self._handle_load_timeout) @@ -573,7 +574,7 @@ def _on_operation_complete(self, result: ConnectionResult) -> None: result.message, ) ssid = self._target_ssid - QtCore.QTimer.singleShot( + QTimer.singleShot( 2000, lambda _ssid=ssid: self._nm.connect_network(_ssid) ) return # Keep loading visible; state machine handles completion @@ -1007,6 +1008,15 @@ def _configure_info_box_centered(self) -> None: self.mn_info_box.setWordWrap(True) self.mn_info_box.setAlignment(QtCore.Qt.AlignmentFlag.AlignCenter) + def _claim_link(self, winner) -> None: + """Force the other two link toggles OFF; only one link may be on at a time.""" + for btn in (self.wifi_button, self.hotspot_button, self.ethernet_button): + if btn is winner: + continue + toggle = btn.toggle_button + with QtCore.QSignalBlocker(toggle): + toggle.state = toggle.State.OFF + @QtCore.pyqtSlot(object, name="stateChange") def _on_toggle_state(self, new_state) -> None: """Route a toggle-button state change to the correct handler (Wi-Fi or hotspot).""" @@ -1027,7 +1037,7 @@ def _on_toggle_state(self, new_state) -> None: # when the worker emits the disconnected state. def _handle_wifi_toggle(self, is_on: bool) -> None: - """Enable or disable Wi-Fi, enforcing the ethernet/hotspot mutual-exclusion rule.""" + """Enable or disable Wi-Fi; turning it on drops the hotspot and the cable.""" if not is_on: self._target_ssid = None self._pending_operation = PendingOperation.WIFI_OFF @@ -1035,45 +1045,36 @@ def _handle_wifi_toggle(self, is_on: bool) -> None: self._nm.set_wifi_enabled(False) return - hotspot_btn = self.hotspot_button.toggle_button - eth_btn = self.ethernet_button.toggle_button - with QtCore.QSignalBlocker(hotspot_btn): - hotspot_btn.state = hotspot_btn.State.OFF - with QtCore.QSignalBlocker(eth_btn): - eth_btn.state = eth_btn.State.OFF + # Guard before touching links: a state update mid-teardown bounces the toggle. + self._target_ssid = None + self._pending_operation = PendingOperation.WIFI_ON + self._set_loading_state(True) + self._claim_link(self.wifi_button) + self._nm.disconnect_ethernet() self._nm.set_wifi_enabled(True) - # NOTE: set_wifi_enabled is dispatched to the worker — cached state - # is STALE here (may still show ethernet). Always proceed to the - # saved-network connection path. - saved = self._nm.saved_networks wifi_networks = [n for n in saved if "ap" not in n.mode] if not wifi_networks: + self._clear_loading() self._show_warning_popup("No saved Wi-Fi networks. Please add one first.") self._display_wifi_on_no_connection() return - # Sort by priority descending (highest priority first), - # then by timestamp as tiebreaker — this gives "reconnect to - # highest-priority saved network" behaviour. + # Reconnect to the highest-priority saved network, newest breaking ties. wifi_networks.sort(key=lambda n: (n.priority, n.timestamp), reverse=True) self._target_ssid = wifi_networks[0].ssid - self._pending_operation = PendingOperation.WIFI_ON - self._set_loading_state(True) # Non-blocking: disable hotspot then connect self._nm.toggle_hotspot(False) _ssid_to_connect = self._target_ssid - QtCore.QTimer.singleShot( - 500, lambda: self._nm.connect_network(_ssid_to_connect) - ) + QTimer.singleShot(500, lambda: self._nm.connect_network(_ssid_to_connect)) def _handle_hotspot_toggle(self, is_on: bool) -> None: - """Enable or disable the hotspot, enforcing the ethernet/Wi-Fi mutual-exclusion rule.""" + """Enable or disable the hotspot; turning it on drops Wi-Fi client and the cable.""" if not is_on: self._target_ssid = None self._pending_operation = PendingOperation.HOTSPOT_OFF @@ -1081,17 +1082,13 @@ def _handle_hotspot_toggle(self, is_on: bool) -> None: self._nm.toggle_hotspot(False) return - 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 - self._target_ssid = None self._pending_operation = PendingOperation.HOTSPOT_ON self._set_loading_state(True) + self._claim_link(self.hotspot_button) + self._nm.disconnect_ethernet() + hotspot_name = self.hotspot_name_input_field.text() or "" hotspot_pass = self.hotspot_password_input_field.text() or "" hotspot_sec = "wpa-psk" @@ -1100,18 +1097,14 @@ def _handle_hotspot_toggle(self, is_on: bool) -> None: self._nm.create_hotspot(hotspot_name, hotspot_pass, hotspot_sec) def _handle_ethernet_toggle(self, is_on: bool) -> None: - """Handle ethernet toggle with mutual exclusion.""" + """Connect or disconnect the cable; connecting drops Wi-Fi and the hotspot.""" if is_on: - wifi_btn = self.wifi_button.toggle_button - hotspot_btn = self.hotspot_button.toggle_button - with QtCore.QSignalBlocker(wifi_btn): - wifi_btn.state = wifi_btn.State.OFF - with QtCore.QSignalBlocker(hotspot_btn): - hotspot_btn.state = hotspot_btn.State.OFF - self._target_ssid = None self._pending_operation = PendingOperation.ETHERNET_ON self._set_loading_state(True) + + self._claim_link(self.ethernet_button) + self._nm.set_wifi_enabled(False) self._nm.connect_ethernet() return diff --git a/tests/network/test_network_ui.py b/tests/network/test_network_ui.py index 7250aaaa..9fd4e750 100644 --- a/tests/network/test_network_ui.py +++ b/tests/network/test_network_ui.py @@ -698,7 +698,7 @@ def test_transient_mismatch_retries(self, win, qapp): message="not compatible with device", error_code="nm_error", ) - with patch("BlocksScreen.lib.panels.networkWindow.QtCore.QTimer") as mock_timer: + 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 @@ -745,7 +745,7 @@ def test_wifi_on_with_saved_networks_starts_connect(self, win): ) ] nm.saved_networks = saved - with patch("BlocksScreen.lib.panels.networkWindow.QtCore.QTimer") as mock_timer: + with patch("BlocksScreen.lib.panels.networkWindow.QTimer") as mock_timer: w._handle_wifi_toggle(True) mock_timer.singleShot.assert_called() assert w._pending_operation == PendingOperation.WIFI_ON diff --git a/tests/network/test_worker_unit.py b/tests/network/test_worker_unit.py index b8a8fa7a..dd09b6ec 100644 --- a/tests/network/test_worker_unit.py +++ b/tests/network/test_worker_unit.py @@ -2028,46 +2028,6 @@ def test_calls_state_and_connectivity(self, qapp): w._async_load_saved_networks.assert_awaited_once() -class TestEnforceBootMutualExclusion: - def test_no_ethernet_returns_early(self, qapp): - w = _make(qapp) - nm = AsyncProxyMock(wireless_enabled=True) - _wire(w, nm=nm) - w._is_ethernet_connected = AsyncMock(return_value=False) - _run(w._enforce_boot_mutual_exclusion()) - # wireless_enabled.set_async should NOT be called - assert ( - not hasattr(nm.wireless_enabled, "set_async") - or not nm.wireless_enabled.set_async.called - ) - - def test_ethernet_active_wifi_on_disables_wifi(self, qapp): - w = _make(qapp) - nm = AsyncProxyMock(wireless_enabled=True) - _wire(w, nm=nm) - wifi = AsyncProxyMock() - wifi.disconnect = AsyncMock() - _wire(w, wifi_proxy=wifi) - w._is_ethernet_connected = AsyncMock(return_value=True) - w._wait_for_wifi_radio = AsyncMock(return_value=True) - _run(w._enforce_boot_mutual_exclusion()) - nm.wireless_enabled.set_async.assert_awaited_once_with(False) - assert w._is_hotspot_active is False - - def test_ethernet_active_wifi_off_no_action(self, qapp): - w = _make(qapp) - nm = AsyncProxyMock(wireless_enabled=False) - _wire(w, nm=nm) - w._is_ethernet_connected = AsyncMock(return_value=True) - _run(w._enforce_boot_mutual_exclusion()) - nm.wireless_enabled.set_async.assert_not_awaited() - - def test_exception_is_non_fatal(self, qapp): - w = _make(qapp) - w._is_ethernet_connected = AsyncMock(side_effect=RuntimeError("boom")) - _run(w._enforce_boot_mutual_exclusion()) # must not raise - - class TestWaitForWifiRadio: def test_returns_true_when_already_matching(self, qapp): w = _make(qapp) @@ -2105,17 +2065,15 @@ 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_disconnects_ethernet(self, qapp): + def test_enable_wifi_leaves_ethernet_untouched(self, qapp): 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_awaited_once() nm.wireless_enabled.set_async.assert_awaited_once_with(True) def test_already_matching_skips_toggle(self, qapp): @@ -2178,11 +2136,7 @@ def test_happy_path(self, qapp): nm = AsyncProxyMock(wireless_enabled=True) nm.activate_connection = AsyncMock() _wire(w, nm=nm) - 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._ensure_wired_autoconnect = AsyncMock() w._build_current_state = AsyncMock(return_value=NetworkState()) w._activate_saved_vlans = AsyncMock() w._is_hotspot_active = False @@ -2191,15 +2145,17 @@ def test_happy_path(self, qapp): w.connection_result.connect(results.append) _run(w._async_connect_ethernet()) - nm.wireless_enabled.set_async.assert_awaited_once_with(False) + w._ensure_wired_autoconnect.assert_awaited_once() nm.activate_connection.assert_awaited_once() assert len(results) == 1 assert results[0].success is True def test_exception_emits_error_and_state(self, qapp): w = _make(qapp) - nm = AsyncProxyMock(wireless_enabled=AsyncMock(side_effect=RuntimeError("x"))) + nm = AsyncProxyMock(wireless_enabled=True) + nm.activate_connection = AsyncMock(side_effect=RuntimeError("x")) w._nm = _ProxyFactory(nm) + w._ensure_wired_autoconnect = AsyncMock() w._build_current_state = AsyncMock(return_value=NetworkState()) errors = [] @@ -2600,7 +2556,6 @@ class TestAsyncInitializeFull: def test_happy_path_full_init(self, qapp): w = _make(qapp, running=False) w._detect_interfaces = AsyncMock() - w._enforce_boot_mutual_exclusion = AsyncMock() w._is_ethernet_connected = AsyncMock(return_value=False) w._activate_saved_vlans = AsyncMock() w._start_signal_listeners = AsyncMock() @@ -2617,7 +2572,6 @@ def test_happy_path_full_init(self, qapp): assert w._running is True w._detect_interfaces.assert_awaited_once() - w._enforce_boot_mutual_exclusion.assert_awaited_once() w._start_signal_listeners.assert_awaited_once() assert len(init_signals) == 1 assert len(hotspot_info) == 1 From 1b985296cd28573f2a0fbbaba5e1d77af5e92ee3 Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Fri, 25 Sep 2026 15:52:54 +0100 Subject: [PATCH 2/3] fix(network): never disable Wi-Fi from state refresh when a cable is connected --- BlocksScreen/lib/network/worker.py | 28 ++++++---------------------- tests/network/test_worker_unit.py | 17 +++++++++++++++++ 2 files changed, 23 insertions(+), 22 deletions(-) diff --git a/BlocksScreen/lib/network/worker.py b/BlocksScreen/lib/network/worker.py index 965bb119..8ea24744 100644 --- a/BlocksScreen/lib/network/worker.py +++ b/BlocksScreen/lib/network/worker.py @@ -286,8 +286,11 @@ async def _async_initialize(self) -> None: Detects network interfaces, activates any saved VLANs if ethernet is present, triggers an initial Wi-Fi scan, and starts all D-Bus signal - listeners. Emits ``initialized`` when done (even on failure, so the + listeners. Emits ``initialized`` when done (even on failure, so the manager can unblock its caller). + + Wired autoconnect is deliberately not re-armed here: NM's latch is what + persists the user's "ethernet off" choice across reboots. """ try: if not self._system_bus: @@ -755,31 +758,12 @@ async def _wait_for_wifi_device_ready(self, timeout: float = 8.0) -> bool: return False async def _async_get_current_state(self) -> None: - """Rebuild and emit the full NetworkState, enforcing runtime mutual exclusion.""" + """Rebuild and emit the full NetworkState. Read-only: never mutates NM.""" try: if not await self._ensure_dbus_connection(): self.state_changed.emit(NetworkState()) return - state = await self._build_current_state() - if ( - state.ethernet_connected - and state.wifi_enabled - and not state.hotspot_enabled - and not self._is_hotspot_active - ): - logger.info( - "Runtime mutual exclusion: ethernet active + " - "Wi-Fi — disabling Wi-Fi" - ) - if self._primary_wifi_path: - try: - await self._wifi().disconnect() - except Exception as exc: - logger.debug("Disconnect before Wi-Fi disable ignored: %s", exc) - await self._nm().wireless_enabled.set_async(False) - await asyncio.sleep(0.5) - state = await self._build_current_state() - self.state_changed.emit(state) + self.state_changed.emit(await self._build_current_state()) except Exception as exc: logger.error("Failed to get current state: %s", exc) self.error_occurred.emit("get_current_state", str(exc)) diff --git a/tests/network/test_worker_unit.py b/tests/network/test_worker_unit.py index a164ae5b..7bace8e5 100644 --- a/tests/network/test_worker_unit.py +++ b/tests/network/test_worker_unit.py @@ -751,6 +751,23 @@ async def test_exception_emits_error(self, qapp): await w._async_get_current_state() assert len(errors) == 1 + @pytest.mark.asyncio + async def test_cable_with_radio_on_never_kills_radio(self, qapp): + w = _make_worker(qapp) + w._ensure_dbus_connection = AsyncMock(return_value=True) + state = NetworkState(ethernet_connected=True, wifi_enabled=True) + w._build_current_state = AsyncMock(return_value=state) + nm_proxy = AsyncProxyMock(wireless_enabled=True) + wifi_proxy = AsyncProxyMock() + w._nm = _ProxyFactory(nm_proxy) + w._wifi = _ProxyFactory(wifi_proxy) + received = [] + w.state_changed.connect(received.append) + await w._async_get_current_state() + assert received == [state] + nm_proxy.wireless_enabled.set_async.assert_not_awaited() + wifi_proxy.disconnect.assert_not_awaited() + class TestBuildCurrentState: @pytest.mark.asyncio From 0266085fd5445226b9a119f9437948c7c48f073e Mon Sep 17 00:00:00 2001 From: Guilherme Costa Date: Mon, 28 Sep 2026 10:24:35 +0100 Subject: [PATCH 3/3] refactor(network): extract helpers and cut cyclomatic complexity (#309) # Description - [x] Code refactor Stacked on #2, review after it merges. Behaviour-neutral motion only, extracts helpers to cut cyclomatic complexity below the C901 threshold (10). Reviewable as pure motion. - Extract `_gather_ap_properties`, `_gather_settings`, `_vlan_*`, `_prefix_to_mask`, `_parse_ipv4_settings`, `_unwrap`, `_setting`, `_delete_connections_where`, `_delete_all_connections_by_id`, `_find_ap_props`, `_reload_connections` - `models.py` cleanup # Tests All 5 gates pass standing alone, including `ruff check --select C901` over the network layer. No behaviour change (pure extraction). --- BlocksScreen/lib/network/models.py | 4 + BlocksScreen/lib/network/worker.py | 811 ++++++++++++---------- tests/network/test_worker_helpers_unit.py | 214 ++++++ 3 files changed, 648 insertions(+), 381 deletions(-) create mode 100644 tests/network/test_worker_helpers_unit.py diff --git a/BlocksScreen/lib/network/models.py b/BlocksScreen/lib/network/models.py index acf9c79a..5bae85ae 100644 --- a/BlocksScreen/lib/network/models.py +++ b/BlocksScreen/lib/network/models.py @@ -237,6 +237,10 @@ 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 = "" + netmask: str = "" + gateway: str = "" + dns_servers: tuple[str, ...] = () @dataclass(frozen=True, slots=True) diff --git a/BlocksScreen/lib/network/worker.py b/BlocksScreen/lib/network/worker.py index 9b13b673..e8085d2b 100644 --- a/BlocksScreen/lib/network/worker.py +++ b/BlocksScreen/lib/network/worker.py @@ -8,6 +8,7 @@ import socket as _socket import struct import threading +from typing import Callable from uuid import uuid4 import sdbus @@ -808,43 +809,11 @@ async def _build_current_state(self) -> NetworkState: eth_connected = await self._is_ethernet_connected() if eth_connected: - current_ip = await self._get_ip_by_interface( - self._primary_wired_iface or "eth0" - ) - if not current_ip: - current_ip = self._get_ip_os_fallback( - self._primary_wired_iface or "eth0" - ) current_ssid = "" - elif current_ssid: - current_ip = await self._get_ip_by_interface("wlan0") - if not current_ip: - current_ip = await self._get_current_ip() - else: - current_ip = "" - - if not current_ip and connectivity in ( - ConnectivityState.FULL, - ConnectivityState.LIMITED, - ): - for _iface in ( - self._primary_wired_iface or "eth0", - "wlan0", - ): - _fallback = self._get_ip_os_fallback(_iface) - if _fallback: - current_ip = _fallback - if _iface != "wlan0": - eth_connected = True - break - - signal = 0 - sec_type = "" - if current_ssid: - signal_map = await self._build_signal_map() - signal = signal_map.get(current_ssid.lower(), 0) - saved = await self._get_saved_network_cached(current_ssid) - sec_type = saved.security_type if saved else "" + current_ip, eth_connected = await self._resolve_current_ip( + "wlan0", current_ssid, eth_connected, connectivity + ) + signal, sec_type = await self._wifi_signal_and_security(current_ssid) hotspot_enabled = current_ssid == self._hotspot_config.ssid @@ -873,6 +842,46 @@ async def _build_current_state(self) -> NetworkState: logger.error("Error building current state: %s", exc) return NetworkState() + async def _resolve_current_ip( + self, + wifi_iface: str, + ssid: str, + eth_connected: bool, + connectivity: ConnectivityState, + ) -> tuple[str, bool]: + """Resolve the active IPv4 address, returning it plus the ethernet flag.""" + wired = self._primary_wired_iface or "eth0" + if eth_connected: + current_ip = await self._get_ip_by_interface(wired) + if not current_ip: + current_ip = self._get_ip_os_fallback(wired) + elif ssid: + current_ip = await self._get_ip_by_interface(wifi_iface) + if not current_ip: + current_ip = await self._get_current_ip() + else: + current_ip = "" + + if current_ip or connectivity not in ( + ConnectivityState.FULL, + ConnectivityState.LIMITED, + ): + return current_ip, eth_connected + + for iface in (wired, wifi_iface): + fallback = self._get_ip_os_fallback(iface) + if fallback: + return fallback, eth_connected or iface != wifi_iface + return current_ip, eth_connected + + async def _wifi_signal_and_security(self, ssid: str) -> tuple[int, str]: + """Return the scanned signal strength and saved security type for ``ssid``.""" + if not ssid: + return 0, "" + signal = (await self._build_signal_map()).get(ssid.lower(), 0) + saved = await self._get_saved_network_cached(ssid) + return signal, saved.security_type if saved else "" + @staticmethod def _map_connectivity(value: int) -> ConnectivityState: """Map a raw NM connectivity integer to a ConnectivityState enum member.""" @@ -1042,6 +1051,15 @@ async def _get_all_ap_properties(self, ap_path: str) -> dict[str, object]: logger.debug("GetAll failed for AP %s: %s", ap_path, exc) return {} + async def _gather_ap_properties( + self, ap_paths: list[str] + ) -> list[tuple[str, dict[str, object]]]: + """Read every AP's properties in one concurrent batch, preserving order.""" + props = await asyncio.gather( + *(self._get_all_ap_properties(p) for p in ap_paths) + ) + return list(zip(ap_paths, props)) + 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] = {} @@ -1167,18 +1185,32 @@ async def _async_load_saved_networks(self) -> None: self.error_occurred.emit("load_saved_networks", str(exc)) self.saved_networks_loaded.emit([]) + async def _gather_settings(self, paths: list[str]) -> list[tuple[str, dict]]: + """Read every profile's settings in one concurrent batch, dropping failures.""" + + async def one(path: str) -> tuple[str, dict | None]: + """Fetch one profile's settings, returning None if the read fails.""" + try: + return path, await self._conn_settings(path).get_settings() + except Exception as exc: + logger.debug("GetSettings failed for %s: %s", path, exc) + return path, None + + results = await asyncio.gather(*(one(p) for p in paths)) + return [(p, s) for p, s in results if s is not None] + async def _get_saved_networks_impl(self) -> list[SavedNetwork]: """Enumerate NM connection profiles and return infrastructure Wi-Fi ones.""" if not self._system_bus: return [] try: - connections = await self._nm_settings().list_connections() - signal_map = await self._build_signal_map() + connections, signal_map = await asyncio.gather( + self._nm_settings().list_connections(), self._build_signal_map() + ) saved: list[SavedNetwork] = [] - 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 @@ -1198,10 +1230,9 @@ async def _get_saved_networks_impl(self) -> list[SavedNetwork]: )[1] timestamp = settings["connection"].get("timestamp", (None, 0))[1] signal = signal_map.get(ssid.lower(), 0) - ipv4_method = settings.get("ipv4", {}).get( - "method", (None, "auto") - )[1] - is_dhcp = ipv4_method != "manual" + ipv4 = settings.get("ipv4", {}) + is_dhcp = ipv4.get("method", (None, "auto"))[1] != "manual" + ip_addr, netmask, gateway, dns = self._parse_ipv4_settings(ipv4) saved.append( SavedNetwork( @@ -1214,6 +1245,10 @@ async def _get_saved_networks_impl(self) -> list[SavedNetwork]: signal_strength=signal, timestamp=int(timestamp or 0), is_dhcp=is_dhcp, + ip_address=ip_addr, + netmask=netmask, + gateway=gateway, + dns_servers=dns, ) ) except Exception as exc: @@ -1270,17 +1305,8 @@ async def _add_network_impl( except Exception as exc: logger.debug("Pre-connect scan request ignored: %s", exc) - ap_paths = await self._wifi().get_all_access_points() - target_ap_path: str | None = None - target_ap_props: dict[str, object] = {} - for ap_path in ap_paths: - props = await self._get_all_ap_properties(ap_path) - if self._decode_ssid(props.get("ssid", b"")) == ssid: - target_ap_path = ap_path - target_ap_props = props - break - - if not target_ap_path: + target_ap_props = await self._find_ap_props(ssid) + if target_ap_props is None: return ConnectionResult(False, f"Network '{ssid}' not found", "not_found") interface = await self._wifi().interface @@ -1307,11 +1333,7 @@ async def _add_network_impl( ) return ConnectionResult(False, str(exc), "add_failed") - if _CAN_RELOAD_CONNECTIONS: - try: - await self._nm_settings().reload_connections() - except Exception as reload_err: - logger.debug("reload_connections non-fatal: %s", reload_err) + await self._reload_connections() try: await self._nm().activate_connection(conn_path) @@ -1330,99 +1352,14 @@ async def _add_network_impl( logger.warning("Activate after add failed: %s", act_err) return ConnectionResult(True, f"Network '{ssid}' added (activate manually)") - def _build_connection_properties( - self, - ssid: str, - password: str, - interface: str, - priority: int, - ap_props: dict[str, object], - ) -> dict[str, object] | None: - """Build NM connection property dict for *ssid* from its AP capability flags. - - Returns None if the security type is unsupported (e.g. WPA-EAP). - Handles OPEN, WPA-PSK, WPA2-PSK, and WPA3-SAE (including SAE-transition). - """ - flags = int(ap_props.get("flags", 0)) - wpa_flags = int(ap_props.get("wpa_flags", 0)) - rsn_flags = int(ap_props.get("rsn_flags", 0)) - - props: dict[str, object] = { - "connection": { - "id": ("s", ssid), - "uuid": ("s", str(uuid4())), - "type": ("s", "802-11-wireless"), - "interface-name": ("s", interface), - "autoconnect": ("b", True), - "autoconnect-priority": ("i", priority), - }, - "802-11-wireless": { - "mode": ("s", "infrastructure"), - "ssid": ("ay", ssid.encode("utf-8")), - }, - "ipv4": { - "method": ("s", "auto"), - "route-metric": ("i", 200), - }, - "ipv6": {"method": ("s", "auto")}, - } - - if (flags & 1) == 0: - return props - - props["802-11-wireless"]["security"] = ( - "s", - "802-11-wireless-security", - ) - security = self._determine_security_type(flags, wpa_flags, rsn_flags) - - if not is_connectable_security(security): - logger.warning( - "Rejecting connection to '%s': unsupported security %s", - ssid, - security.value, - ) - return None - - if security == SecurityType.WPA3_SAE: - has_psk = bool((rsn_flags & 0x100) or wpa_flags) - if has_psk: - logger.debug( - "SAE transition for '%s' — using wpa-psk + PMF optional", - ssid, - ) - props["802-11-wireless-security"] = { - "key-mgmt": ("s", "wpa-psk"), - "auth-alg": ("s", "open"), - "psk": ("s", password), - "pmf": ("u", 2), # OPTIONAL — required for SAE-transition APs - } - else: - logger.debug("Pure SAE detected for '%s'", ssid) - props["802-11-wireless-security"] = { - "key-mgmt": ("s", "sae"), - "auth-alg": ("s", "open"), - "psk": ("s", password), - "pmf": ("u", 3), # REQUIRED — mandatory for pure WPA3-SAE - } - elif security in ( - SecurityType.WPA2_PSK, - SecurityType.WPA_PSK, - ): - props["802-11-wireless-security"] = { - "key-mgmt": ("s", "wpa-psk"), - "auth-alg": ("s", "open"), - "psk": ("s", password), - } - else: - logger.warning( - "Unsupported security type '%s' for '%s'", - security.value, - ssid, - ) - return None - - return props + async def _reload_connections(self) -> None: + """Ask NM to re-read connection files; non-fatal and root-only.""" + if not _CAN_RELOAD_CONNECTIONS: + return + try: + await self._nm_settings().reload_connections() + except Exception as reload_err: + logger.debug("reload_connections non-fatal: %s", reload_err) async def _async_connect_network(self, ssid: str) -> None: """Activate an existing saved Wi-Fi profile and emit connection_result.""" @@ -1566,11 +1503,7 @@ async def _delete_network_impl(self, ssid: str) -> ConnectionResult: try: await self._conn_settings(conn_path).delete() - if _CAN_RELOAD_CONNECTIONS: - try: - await self._nm_settings().reload_connections() - except Exception as reload_err: - logger.debug("reload_connections non-fatal: %s", reload_err) + await self._reload_connections() current_ssid = await self._get_current_ssid() if current_ssid and current_ssid.lower() == ssid.lower(): @@ -1779,36 +1712,17 @@ async def _async_create_vlan( iface = self._primary_wired_iface or "eth0" - try: - existing_conns = await self._nm_settings().list_connections() - for existing_path in existing_conns: - try: - s = await self._conn_settings(existing_path).get_settings() - if ( - s.get("connection", {}).get("type", (None, ""))[1] == "vlan" - and s.get("vlan", {}).get("id", (None, -1))[1] == vlan_id - and s.get("vlan", {}).get("parent", (None, ""))[1] == iface - ): - self.connection_result.emit( - ConnectionResult( - False, - f"VLAN {vlan_id} already exists on " - f"{iface}.\nRemove it first before " - "creating a new one.", - "duplicate_vlan", - ) - ) - return - except Exception as exc: - logger.debug( - "Skipping connection in duplicate VLAN check: %s", exc - ) - continue - except Exception as dup_err: - logger.debug( - "Duplicate VLAN check failed (non-fatal): %s", - dup_err, + if await self._vlan_profile_exists(vlan_id, iface): + self.connection_result.emit( + ConnectionResult( + False, + f"VLAN {vlan_id} already exists on " + f"{iface}.\nRemove it first before " + "creating a new one.", + "duplicate_vlan", + ) ) + return vlan_conn_id = f"VLAN {vlan_id}" @@ -1818,38 +1732,13 @@ async def _async_create_vlan( await self._delete_all_connections_by_id(vlan_conn_id) await asyncio.sleep(0.5) - prefix = self._mask_to_prefix(subnet_mask) - ip_uint = self._ip_to_nm_uint32(ip_address) - gw_uint = self._ip_to_nm_uint32(gateway) if gateway else 0 - dns_list: list[int] = [] - if dns1: - dns_list.append(self._ip_to_nm_uint32(dns1)) - if dns2: - dns_list.append(self._ip_to_nm_uint32(dns2)) - - conn_props: dict[str, object] = { - "connection": { - "id": ("s", vlan_conn_id), - "uuid": ("s", str(uuid4())), - "type": ("s", "vlan"), - "autoconnect": ("b", False), - }, - "vlan": { - "id": ("u", vlan_id), - "parent": ("s", iface), - }, - "ipv4": { - "method": ("s", "manual"), - "addresses": ( - "aau", - [[ip_uint, prefix, gw_uint]], - ), - "gateway": ("s", gateway or ""), - "dns": ("au", dns_list), - "route-metric": ("i", 500), - }, - "ipv6": {"method": ("s", "ignore")}, - } + conn_props = self._build_vlan_properties( + vlan_conn_id, + vlan_id, + iface, + (ip_address, subnet_mask, gateway), + (dns1, dns2), + ) conn_path = await self._nm_settings().add_connection(conn_props) await self._nm().activate_connection(conn_path, "/", "/") @@ -1865,6 +1754,62 @@ async def _async_create_vlan( self.error_occurred.emit("create_vlan", str(exc)) self.state_changed.emit(await self._build_current_state()) + async def _vlan_profile_exists(self, vlan_id: int, iface: str) -> bool: + """Check whether a VLAN profile with *vlan_id* already rides on *iface*.""" + try: + for existing_path in await self._nm_settings().list_connections(): + try: + s = await self._conn_settings(existing_path).get_settings() + except Exception as exc: + logger.debug("Skipping connection in duplicate VLAN check: %s", exc) + continue + if ( + s.get("connection", {}).get("type", (None, ""))[1] == "vlan" + and s.get("vlan", {}).get("id", (None, -1))[1] == vlan_id + and s.get("vlan", {}).get("parent", (None, ""))[1] == iface + ): + return True + except Exception as dup_err: + logger.debug("Duplicate VLAN check failed (non-fatal): %s", dup_err) + return False + + def _build_vlan_properties( + self, + conn_id: str, + vlan_id: int, + iface: str, + ipv4: tuple[str, str, str], + dns: tuple[str, str], + ) -> dict[str, object]: + """Build the NM property dict for a static-IP VLAN profile on *iface*.""" + ip_address, subnet_mask, gateway = ipv4 + dns_list = [self._ip_to_nm_uint32(d) for d in dns if d] + addr = [ + self._ip_to_nm_uint32(ip_address), + self._mask_to_prefix(subnet_mask), + self._ip_to_nm_uint32(gateway) if gateway else 0, + ] + return { + "connection": { + "id": ("s", conn_id), + "uuid": ("s", str(uuid4())), + "type": ("s", "vlan"), + "autoconnect": ("b", False), + }, + "vlan": { + "id": ("u", vlan_id), + "parent": ("s", iface), + }, + "ipv4": { + "method": ("s", "manual"), + "addresses": ("aau", [addr]), + "gateway": ("s", gateway or ""), + "dns": ("au", dns_list), + "route-metric": ("i", 500), + }, + "ipv6": {"method": ("s", "ignore")}, + } + async def _async_delete_vlan(self, vlan_id: int) -> None: """Delete all NM connection profiles for *vlan_id* and emit connection_result.""" try: @@ -1886,83 +1831,71 @@ async def _get_active_vlans(self) -> tuple[VlanInfo, ...]: """Return a tuple of VlanInfo for all currently active VLAN connections.""" vlans: list[VlanInfo] = [] try: - active_paths = await self._nm().active_connections - for active_path in active_paths: + for active_path in await self._nm().active_connections: try: - ac = self._active_conn(active_path) - conn_path = await ac.connection - settings = await self._conn_settings(conn_path).get_settings() - conn_type = settings.get("connection", {}).get("type", (None, ""))[ - 1 - ] - if conn_type != "vlan": - continue - - vlan_id = settings.get("vlan", {}).get("id", (None, 0))[1] - iface = settings.get("connection", {}).get( - "interface-name", (None, "") - )[1] - if not iface: - parent = settings.get("vlan", {}).get("parent", (None, "eth0"))[ - 1 - ] - iface = f"{parent}.{vlan_id}" - - ipv4_method = settings.get("ipv4", {}).get( - "method", (None, "auto") - )[1] - is_dhcp = ipv4_method != "manual" - - dns_data = settings.get("ipv4", {}).get("dns-data", (None, []))[1] - dns_servers: tuple[str, ...] = () - if dns_data: - dns_servers = tuple(str(d) for d in dns_data) - else: - dns_raw = settings.get("ipv4", {}).get("dns", (None, []))[1] - if dns_raw: - dns_servers = tuple( - self._nm_uint32_to_ip(d) for d in dns_raw - ) - - ip_addr = "" - gateway = "" - try: - ip4_path = await self._active_conn(active_path).ip4_config - if ip4_path and ip4_path != "/": - ip4_cfg = self._ipv4(ip4_path) - addr_data = await ip4_cfg.address_data - if addr_data: - ip_addr = str(addr_data[0]["address"][1]) - gw = await ip4_cfg.gateway - if gw: - gateway = str(gw) - except Exception as exc: - logger.debug( - "D-Bus IP read for VLAN failed, falling back to OS: %s", exc - ) - if iface: - ip_addr = await self._get_ip_by_interface(iface) - - if not ip_addr and iface: - ip_addr = self._get_ip_os_fallback(iface) - - vlans.append( - VlanInfo( - vlan_id=int(vlan_id), - ip_address=ip_addr, - interface=iface, - gateway=gateway, - dns_servers=dns_servers, - is_dhcp=is_dhcp, - ) - ) + vlan = await self._vlan_from_active(active_path) except Exception as exc: logger.debug("Skipping connection in active VLAN list: %s", exc) continue + if vlan: + vlans.append(vlan) except Exception as exc: logger.debug("Error getting active VLANs: %s", exc) return tuple(vlans) + async def _vlan_from_active(self, active_path: str) -> VlanInfo | None: + """Build VlanInfo for an active connection path, or None if it is not a VLAN.""" + conn_path = await self._active_conn(active_path).connection + settings = await self._conn_settings(conn_path).get_settings() + conn = settings.get("connection", {}) + if conn.get("type", (None, ""))[1] != "vlan": + return None + + vlan_cfg = settings.get("vlan", {}) + vlan_id = vlan_cfg.get("id", (None, 0))[1] + iface = conn.get("interface-name", (None, ""))[1] + if not iface: + iface = f"{vlan_cfg.get('parent', (None, 'eth0'))[1]}.{vlan_id}" + + ipv4 = settings.get("ipv4", {}) + ip_addr, gateway = await self._vlan_ip_gateway(active_path, iface) + return VlanInfo( + vlan_id=int(vlan_id), + ip_address=ip_addr, + interface=iface, + gateway=gateway, + dns_servers=self._vlan_dns(ipv4), + is_dhcp=ipv4.get("method", (None, "auto"))[1] != "manual", + ) + + def _vlan_dns(self, ipv4: dict) -> tuple[str, ...]: + """Extract DNS servers from an ipv4 settings dict, preferring ``dns-data``.""" + dns_data = ipv4.get("dns-data", (None, []))[1] + if dns_data: + return tuple(str(d) for d in dns_data) + return tuple(self._nm_uint32_to_ip(d) for d in ipv4.get("dns", (None, []))[1]) + + async def _vlan_ip_gateway(self, active_path: str, iface: str) -> tuple[str, str]: + """Read a VLAN's IPv4 address and gateway, falling back to the OS on failure.""" + ip_addr = "" + gateway = "" + try: + ip4_path = await self._active_conn(active_path).ip4_config + if ip4_path and ip4_path != "/": + ip4_cfg = self._ipv4(ip4_path) + addr_data = await ip4_cfg.address_data + if addr_data: + ip_addr = str(addr_data[0]["address"][1]) + gateway = str(await ip4_cfg.gateway or "") + except Exception as exc: + logger.debug("D-Bus IP read for VLAN failed, falling back to OS: %s", exc) + if iface: + ip_addr = await self._get_ip_by_interface(iface) + + if not ip_addr and iface: + ip_addr = self._get_ip_os_fallback(iface) + return ip_addr, gateway + async def _deactivate_all_vlans(self) -> None: """Deactivate all active VLAN connections via the NM D-Bus interface.""" try: @@ -2046,41 +1979,38 @@ async def _reconnect_wifi_profile(self, ssid: str) -> None: ) return + found_ip = await self._wait_for_profile_ip(ssid, timeout=10.0) + if not found_ip: + logger.warning("Reconnect for '%s': IP not assigned within 10 s", ssid) + return + + logger.info("Reconnect complete for '%s': IP=%s", ssid, found_ip) + try: + self._invalidate_saved_cache() + self.saved_networks_loaded.emit(await self._get_saved_networks_impl()) + except Exception as cache_err: + logger.debug("Cache refresh after reconnect failed: %s", cache_err) + + async def _wait_for_profile_ip(self, ssid: str, timeout: float) -> str: + """Poll for an IPv4 address on *ssid*, returning it or "" once *timeout* elapses.""" loop = asyncio.get_running_loop() - deadline = loop.time() + 10.0 + deadline = loop.time() + timeout while loop.time() < deadline: await asyncio.sleep(1.0) - found_ip: str = "" try: current = await self._get_current_ssid() - if current and current.lower() == ssid.lower(): - found_ip = await self._get_current_ip() or "" - if not found_ip: - found_ip = self._get_ip_os_fallback("wlan0") or "" + if not current or current.lower() != ssid.lower(): + continue + found_ip = await self._get_current_ip() or "" + if not found_ip: + found_ip = self._get_ip_os_fallback("wlan0") or "" + if found_ip: + return found_ip except Exception as exc: logger.debug( "IP address lookup during connection wait ignored: %s", exc ) - - if found_ip: - logger.info( - "Reconnect complete for '%s': IP=%s", - ssid, - found_ip, - ) - try: - self._invalidate_saved_cache() - self.saved_networks_loaded.emit( - await self._get_saved_networks_impl() - ) - except Exception as cache_err: - logger.debug( - "Cache refresh after reconnect failed: %s", - cache_err, - ) - return - - logger.warning("Reconnect for '%s': IP not assigned within 10 s", ssid) + return "" async def _async_update_wifi_static_ip( self, @@ -2466,27 +2396,144 @@ async def _deactivate_connection_by_id(self, conn_id: str) -> bool: logger.debug("Error deactivating '%s': %s", conn_id, exc) return False - async def _delete_all_connections_by_id(self, conn_id: str) -> int: - """Delete every NM connection profile whose id exactly matches *conn_id*.""" + @staticmethod + def _setting(settings: dict, section: str, key: str, default: str = "") -> str: + """Unwrap a NM settings value from its (signature, value) variant tuple.""" + try: + return settings.get(section, {}).get(key, (None, default))[1] + except (TypeError, IndexError, KeyError): + return default + + async def _delete_connections_where( + self, match: Callable[[dict], bool], label: str + ) -> int: + """Delete every profile whose settings satisfy *match*, returning the count.""" deleted = 0 try: - connections = await self._nm_settings().list_connections() - for conn_path in connections: + paths = await self._nm_settings().list_connections() + for conn_path, settings in await self._gather_settings(paths): try: - cs = self._conn_settings(conn_path) - settings = await cs.get_settings() - cid = settings.get("connection", {}).get("id", (None, ""))[1] - if cid == conn_id: - await cs.delete() - deleted += 1 + if not match(settings): + continue + await self._conn_settings(conn_path).delete() + deleted += 1 + logger.debug("Deleted profile at %s (%s)", conn_path, label) except Exception as exc: logger.debug( - "Skipping connection in cleanup for '%s': %s", conn_id, exc + "Skip %s during '%s' cleanup: %s", conn_path, label, exc ) except Exception as exc: - logger.error("Cleanup for '%s' failed: %s", conn_id, exc) + logger.error("Cleanup for '%s' failed: %s", label, exc) return deleted + async def _find_ap_props(self, ssid: str) -> dict[str, object] | None: + """Return the scanned AP properties for *ssid*, or None if it is not visible.""" + ap_paths = await self._wifi().get_all_access_points() + for _ap_path, props in await self._gather_ap_properties(ap_paths): + if self._decode_ssid(props.get("ssid", b"")) == ssid: + return props + return None + + def _build_connection_properties( + self, + ssid: str, + password: str, + interface: str, + priority: int, + ap_props: dict[str, object], + ) -> dict[str, object] | None: + """Build NM connection property dict for *ssid* from its AP capability flags. + + Returns None if the security type is unsupported (e.g. WPA-EAP). + Handles OPEN, WPA-PSK, WPA2-PSK, and WPA3-SAE (including SAE-transition). + """ + flags = int(ap_props.get("flags", 0)) + wpa_flags = int(ap_props.get("wpa_flags", 0)) + rsn_flags = int(ap_props.get("rsn_flags", 0)) + + props: dict[str, object] = { + "connection": { + "id": ("s", ssid), + "uuid": ("s", str(uuid4())), + "type": ("s", "802-11-wireless"), + "interface-name": ("s", interface), + "autoconnect": ("b", True), + "autoconnect-priority": ("i", priority), + }, + "802-11-wireless": { + "mode": ("s", "infrastructure"), + "ssid": ("ay", ssid.encode("utf-8")), + }, + "ipv4": { + "method": ("s", "auto"), + "route-metric": ("i", 200), + }, + "ipv6": {"method": ("s", "auto")}, + } + + if (flags & 1) == 0: + return props + + props["802-11-wireless"]["security"] = ( + "s", + "802-11-wireless-security", + ) + security = self._determine_security_type(flags, wpa_flags, rsn_flags) + + if not is_connectable_security(security): + logger.warning( + "Rejecting connection to '%s': unsupported security %s", + ssid, + security.value, + ) + return None + + if security == SecurityType.WPA3_SAE: + has_psk = bool((rsn_flags & 0x100) or wpa_flags) + if has_psk: + logger.debug( + "SAE transition for '%s': using wpa-psk + PMF optional", + ssid, + ) + props["802-11-wireless-security"] = { + "key-mgmt": ("s", "wpa-psk"), + "auth-alg": ("s", "open"), + "psk": ("s", password), + "pmf": ("u", 2), # OPTIONAL: required for SAE-transition APs + } + else: + logger.debug("Pure SAE detected for '%s'", ssid) + props["802-11-wireless-security"] = { + "key-mgmt": ("s", "sae"), + "auth-alg": ("s", "open"), + "psk": ("s", password), + "pmf": ("u", 3), # REQUIRED: mandatory for pure WPA3-SAE + } + elif security in ( + SecurityType.WPA2_PSK, + SecurityType.WPA_PSK, + ): + props["802-11-wireless-security"] = { + "key-mgmt": ("s", "wpa-psk"), + "auth-alg": ("s", "open"), + "psk": ("s", password), + } + else: + logger.warning( + "Unsupported security type '%s' for '%s'", + security.value, + ssid, + ) + return None + + return props + + async def _delete_all_connections_by_id(self, conn_id: str) -> int: + """Delete every NM connection profile whose id exactly matches *conn_id*.""" + return await self._delete_connections_where( + lambda s: self._setting(s, "connection", "id") == conn_id, conn_id + ) + async def _delete_all_ap_mode_connections(self) -> int: """Delete all saved Wi-Fi connections in AP mode. @@ -2495,67 +2542,24 @@ async def _delete_all_ap_mode_connections(self) -> int: old AP-mode profiles accumulate in NetworkManager and NM may auto-activate them on the next boot. """ - deleted = 0 - try: - connections = await self._nm_settings().list_connections() - for conn_path in connections: - try: - cs = self._conn_settings(conn_path) - settings = await cs.get_settings() - conn_type = settings.get("connection", {}).get("type", (None, ""))[ - 1 - ] - if conn_type != "802-11-wireless": - continue - mode = settings.get("802-11-wireless", {}).get("mode", (None, ""))[ - 1 - ] - if mode == "ap": - conn_id = settings.get("connection", {}).get("id", (None, ""))[ - 1 - ] - await cs.delete() - deleted += 1 - logger.debug( - "Removed stale AP profile '%s' at %s", conn_id, conn_path - ) - except Exception as exc: - logger.debug("Skipping connection in AP profile cleanup: %s", exc) - except Exception as exc: - logger.error("Failed to remove stale AP profiles: %s", exc) + + def is_ap_profile(s: dict) -> bool: + """True when the settings dict describes a Wi-Fi profile in AP mode.""" + return ( + self._setting(s, "connection", "type") == "802-11-wireless" + and self._setting(s, "802-11-wireless", "mode") == "ap" + ) + + deleted = await self._delete_connections_where(is_ap_profile, "stale AP mode") if deleted: self._invalidate_saved_cache() return deleted async def _delete_connections_by_id(self, ssid: str) -> int: """Delete every NM connection profile whose id matches *ssid* (case-insensitive).""" - deleted = 0 - try: - connections = await self._nm_settings().list_connections() - for conn_path in connections: - try: - cs = self._conn_settings(conn_path) - settings = await cs.get_settings() - conn_id = settings.get("connection", {}).get("id", (None, ""))[1] - if conn_id.lower() == ssid.lower(): - await cs.delete() - deleted += 1 - logger.debug( - "Deleted stale profile '%s' at %s", - conn_id, - conn_path, - ) - except Exception as exc: - logger.debug( - "Skip connection %s during cleanup: %s", - conn_path, - exc, - ) - except Exception as exc: - logger.error( - "Failed to enumerate connections for cleanup: %s", - exc, - ) + deleted = await self._delete_connections_where( + lambda s: self._setting(s, "connection", "id").lower() == ssid.lower(), ssid + ) if deleted: self._invalidate_saved_cache() return deleted @@ -2570,6 +2574,51 @@ def _nm_uint32_to_ip(uint_ip: int) -> str: """Convert a native-endian uint32 from NM back to a dotted-decimal IPv4 string.""" return str(ipaddress.IPv4Address(struct.pack("=I", uint_ip))) + @staticmethod + def _prefix_to_mask(prefix: int) -> str: + """Convert an integer prefix length to a dotted-decimal subnet mask.""" + if not 0 <= prefix <= 32: + return "" + return str(ipaddress.IPv4Network(f"0.0.0.0/{prefix}").netmask) + + @classmethod + def _parse_ipv4_settings(cls, ipv4: dict) -> tuple[str, str, str, tuple[str, ...]]: + """Extract address, mask, gateway and DNS from an NM ipv4 settings dict.""" + ip_addr, prefix = "", 0 + try: + # NM syncs legacy 'addresses' with 'address-data'; either one works. + addrs = ipv4.get("addresses", (None, []))[1] or [] + if addrs: + ip_addr = cls._nm_uint32_to_ip(addrs[0][0]) + prefix = int(addrs[0][1]) + else: + data = ipv4.get("address-data", (None, []))[1] or [] + if data: + ip_addr = str(cls._unwrap(data[0].get("address", ""))) + prefix = int(cls._unwrap(data[0].get("prefix", 0)) or 0) + except Exception as exc: + logger.debug("parse_ipv4_settings: address parse failed: %s", exc) + ip_addr, prefix = "", 0 + + gateway = str(ipv4.get("gateway", (None, ""))[1] or "") + dns = tuple(str(d) for d in (ipv4.get("dns-data", (None, []))[1] or [])) + if not dns: + try: + dns = tuple( + cls._nm_uint32_to_ip(d) + for d in ipv4.get("dns", (None, []))[1] or [] + ) + except Exception as exc: + logger.debug("parse_ipv4_settings: dns parse failed: %s", exc) + return ip_addr, cls._prefix_to_mask(prefix) if prefix else "", gateway, dns + + @staticmethod + def _unwrap(value: object) -> object: + """Return the payload of an NM (signature, value) variant tuple.""" + if isinstance(value, tuple) and len(value) == 2: + return value[1] + return value + @staticmethod def _mask_to_prefix(mask_str: str) -> int: """Convert a subnet mask or CIDR prefix string to an integer prefix length.""" diff --git a/tests/network/test_worker_helpers_unit.py b/tests/network/test_worker_helpers_unit.py new file mode 100644 index 00000000..95052235 --- /dev/null +++ b/tests/network/test_worker_helpers_unit.py @@ -0,0 +1,214 @@ +"""Unit tests for the NetworkManagerWorker helpers extracted to cut C901 complexity.""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from PyQt6.QtCore import QObject + +from BlocksScreen.lib.network import worker as _worker_mod +from BlocksScreen.lib.network.models import ConnectivityState +from BlocksScreen.lib.network.worker import NetworkManagerWorker +from tests.network.conftest import AsyncProxyMock, _ProxyFactory + +_ONLINE = ConnectivityState.FULL + + +def _worker(qapp) -> NetworkManagerWorker: + """Build a worker without its asyncio thread or D-Bus connection.""" + with patch.object( + NetworkManagerWorker, "__init__", lambda self: QObject.__init__(self) + ): + w = NetworkManagerWorker() + w._system_bus = MagicMock(name="mock_bus") + w._primary_wifi_iface = "wlan0" + w._primary_wired_iface = "eth0" + w._iface_to_device_path = {} + w._saved_cache_dirty = False + w._get_ip_by_interface = AsyncMock(return_value="") + w._get_current_ip = AsyncMock(return_value="") + w._get_ip_os_fallback = MagicMock(return_value="") + return w + + +class TestResolveCurrentIp: + @pytest.mark.asyncio + async def test_wired_uses_nm_address(self, qapp): + w = _worker(qapp) + w._get_ip_by_interface.return_value = "192.168.0.10" + assert await w._resolve_current_ip("wlan0", "", True, _ONLINE) == ( + "192.168.0.10", + True, + ) + w._get_ip_by_interface.assert_awaited_once_with("eth0") + + @pytest.mark.asyncio + async def test_wired_falls_back_to_os_address(self, qapp): + w = _worker(qapp) + w._get_ip_os_fallback.return_value = "192.168.0.11" + assert await w._resolve_current_ip("wlan0", "", True, _ONLINE) == ( + "192.168.0.11", + True, + ) + w._get_ip_os_fallback.assert_called_once_with("eth0") + + @pytest.mark.asyncio + async def test_wifi_falls_back_to_primary_connection_ip(self, qapp): + w = _worker(qapp) + w._get_current_ip.return_value = "10.0.0.5" + assert await w._resolve_current_ip("wlan0", "Home", False, _ONLINE) == ( + "10.0.0.5", + False, + ) + w._get_ip_by_interface.assert_awaited_once_with("wlan0") + + @pytest.mark.asyncio + async def test_os_fallback_on_wired_marks_ethernet(self, qapp): + w = _worker(qapp) + w._get_ip_os_fallback.side_effect = lambda i: ( + "192.168.0.12" if i == "eth0" else "" + ) + assert await w._resolve_current_ip("wlan0", "", False, _ONLINE) == ( + "192.168.0.12", + True, + ) + + @pytest.mark.asyncio + async def test_os_fallback_on_wifi_keeps_ethernet_flag(self, qapp): + w = _worker(qapp) + w._get_ip_os_fallback.side_effect = lambda i: "10.0.0.6" if i == "wlan0" else "" + assert await w._resolve_current_ip("wlan0", "", False, _ONLINE) == ( + "10.0.0.6", + False, + ) + + @pytest.mark.asyncio + async def test_offline_skips_os_fallback(self, qapp): + w = _worker(qapp) + result = await w._resolve_current_ip("wlan0", "", False, ConnectivityState.NONE) + assert result == ("", False) + w._get_ip_os_fallback.assert_not_called() + + +class TestWifiSignalAndSecurity: + @pytest.mark.asyncio + async def test_no_ssid_skips_dbus(self, qapp): + w = _worker(qapp) + w._build_signal_map = AsyncMock() + assert await w._wifi_signal_and_security("") == (0, "") + w._build_signal_map.assert_not_awaited() + + +class TestBuildVlanProperties: + def test_full_static_profile(self, qapp): + w = _worker(qapp) + props = w._build_vlan_properties( + "VLAN 10", + 10, + "eth0", + ("192.168.10.2", "255.255.255.0", "192.168.10.1"), + ("1.1.1.1", "8.8.8.8"), + ) + to_uint = NetworkManagerWorker._ip_to_nm_uint32 + assert props["connection"]["id"] == ("s", "VLAN 10") + assert props["connection"]["type"] == ("s", "vlan") + assert props["connection"]["autoconnect"] == ("b", False) + assert props["vlan"] == {"id": ("u", 10), "parent": ("s", "eth0")} + assert props["ipv4"]["addresses"] == ( + "aau", + [[to_uint("192.168.10.2"), 24, to_uint("192.168.10.1")]], + ) + assert props["ipv4"]["gateway"] == ("s", "192.168.10.1") + assert props["ipv4"]["dns"] == ("au", [to_uint("1.1.1.1"), to_uint("8.8.8.8")]) + assert props["ipv6"] == {"method": ("s", "ignore")} + + def test_optional_gateway_and_dns_omitted(self, qapp): + w = _worker(qapp) + props = w._build_vlan_properties( + "VLAN 20", 20, "eth0", ("10.20.0.2", "16", ""), ("", "9.9.9.9") + ) + assert props["ipv4"]["addresses"][1][0][1:] == [16, 0] + assert props["ipv4"]["gateway"] == ("s", "") + assert props["ipv4"]["dns"] == ( + "au", + [NetworkManagerWorker._ip_to_nm_uint32("9.9.9.9")], + ) + + def test_invalid_address_raises(self, qapp): + w = _worker(qapp) + with pytest.raises(ValueError): + w._build_vlan_properties( + "VLAN 30", 30, "eth0", ("999.1.1.1", "24", ""), ("", "") + ) + + +@patch.object(_worker_mod.asyncio, "sleep", new=AsyncMock()) +class TestWaitForProfileIp: + @pytest.mark.asyncio + 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" + assert await w._wait_for_profile_ip("Home", timeout=5.0) == "10.0.0.7" + assert w._get_current_ssid.await_count == 2 + + @pytest.mark.asyncio + async def test_os_fallback_when_nm_has_no_ip(self, qapp): + w = _worker(qapp) + w._get_current_ssid = AsyncMock(return_value="Home") + w._get_ip_os_fallback.return_value = "10.0.0.8" + assert await w._wait_for_profile_ip("Home", timeout=5.0) == "10.0.0.8" + + @pytest.mark.asyncio + 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" + assert await w._wait_for_profile_ip("Home", timeout=5.0) == "10.0.0.9" + + @pytest.mark.asyncio + async def test_timeout_returns_empty(self, qapp): + w = _worker(qapp) + w._get_current_ssid = AsyncMock(return_value="Other") + assert await w._wait_for_profile_ip("Home", timeout=0.01) == "" + + +class TestDeleteAllApModeConnections: + @pytest.mark.asyncio + async def test_deletes_only_ap_profiles_and_invalidates_cache(self, qapp): + w = _worker(qapp) + profiles = { + "/c/ap": { + "connection": {"type": ("s", "802-11-wireless")}, + "802-11-wireless": {"mode": ("s", "ap")}, + }, + "/c/sta": { + "connection": {"type": ("s", "802-11-wireless")}, + "802-11-wireless": {"mode": ("s", "infrastructure")}, + }, + "/c/eth": {"connection": {"type": ("s", "802-3-ethernet")}}, + } + conns = { + path: AsyncProxyMock( + get_settings=AsyncMock(return_value=s), delete=AsyncMock() + ) + for path, s in profiles.items() + } + w._nm_settings = _ProxyFactory( + AsyncProxyMock(list_connections=AsyncMock(return_value=list(conns))) + ) + w._conn_settings = lambda path: conns[path] + + assert await w._delete_all_ap_mode_connections() == 1 + conns["/c/ap"].delete.assert_awaited_once() + conns["/c/sta"].delete.assert_not_awaited() + conns["/c/eth"].delete.assert_not_awaited() + assert w._saved_cache_dirty is True + + @pytest.mark.asyncio + async def test_nothing_deleted_keeps_cache(self, qapp): + w = _worker(qapp) + w._nm_settings = _ProxyFactory( + AsyncProxyMock(list_connections=AsyncMock(return_value=[])) + ) + assert await w._delete_all_ap_mode_connections() == 0 + assert w._saved_cache_dirty is False