diff --git a/BlocksScreen/lib/network/worker.py b/BlocksScreen/lib/network/worker.py index e8085d2b..8ea24744 100644 --- a/BlocksScreen/lib/network/worker.py +++ b/BlocksScreen/lib/network/worker.py @@ -8,7 +8,7 @@ import socket as _socket import struct import threading -from typing import Callable +from collections.abc import Callable from uuid import uuid4 import sdbus @@ -284,11 +284,13 @@ 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). + + 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: @@ -297,7 +299,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 +368,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. @@ -749,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)) @@ -1289,17 +1279,14 @@ async def _add_network_impl( ) -> ConnectionResult: """Scan for the SSID, build a connection profile, add it to NM, and activate it. - Deletes any pre-existing profile for the same SSID before adding. + Any pre-existing profile for the same SSID is backed up before being + replaced, and restored if the new credentials fail to activate. Returns a failed ConnectionResult if the SSID is not visible, the security type is unsupported, or the 20-second activation wait times out. """ 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() - try: await self._wifi().request_scan({}) except Exception as exc: @@ -1320,10 +1307,14 @@ async def _add_network_impl( "unsupported_security", ) + # 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: nm_settings = self._nm_settings() conn_path = await nm_settings.add_connection(conn_props) 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( @@ -1338,15 +1329,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) @@ -1361,6 +1344,66 @@ 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: + """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.""" try: @@ -1585,16 +1628,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: @@ -1632,7 +1672,13 @@ async def _async_disconnect_ethernet(self) -> None: return try: await self._deactivate_all_vlans() - await self._wired().disconnect() + try: + await self._wired().disconnect() + except Exception as exc: + # Already inactive is the goal state, not a failure. + if "not active" not in str(exc).lower(): + raise + logger.debug("Ethernet already inactive: %s", exc) loop = asyncio.get_running_loop() deadline = loop.time() + 4.0 while loop.time() < deadline: @@ -1642,9 +1688,15 @@ async def _async_disconnect_ethernet(self) -> None: logger.info("Ethernet disconnected") except Exception as exc: logger.error("Failed to disconnect ethernet: %s", exc) + finally: + # Only user toggles reach here, so record intent even if teardown failed. + await self._set_wired_profiles_autoconnect(False) 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 @@ -1652,16 +1704,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) @@ -1695,18 +1738,9 @@ async def _async_create_vlan( if self._is_hotspot_active: await self._async_toggle_hotspot(False) - if self._primary_wifi_path: - try: - await self._wifi().disconnect() - except Exception as exc: - logger.debug("Pre-VLAN disconnect ignored: %s", exc) - await asyncio.sleep(0.5) - - if await self._nm().wireless_enabled: - await self._nm().wireless_enabled.set_async(False) - await self._wait_for_wifi_radio(False, timeout=8.0) - + # A VLAN rides on eth0; Wi-Fi is orthogonal and stays up. if not await self._is_ethernet_connected(): + await self._ensure_wired_autoconnect() await self._nm().activate_connection("/", self._primary_wired_path, "/") await asyncio.sleep(1.5) diff --git a/BlocksScreen/lib/panels/networkWindow.py b/BlocksScreen/lib/panels/networkWindow.py index 7e5b44ff..ef77b14b 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__) @@ -182,7 +183,6 @@ def _init_instance_variables(self) -> None: self._current_network_is_hidden = False self._is_connecting = False self._target_ssid: str | None = None - self._was_ethernet_connected: bool = False self._initial_priority: ConnectionPriority = ConnectionPriority.MEDIUM self._pending_operation: PendingOperation = PendingOperation.NONE self._pending_expected_ip: str = ( @@ -264,7 +264,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) @@ -302,30 +302,9 @@ def _on_network_state_changed(self, state: NetworkState) -> None: self._handle_first_run(state) self._emit_status_icon(state) self._is_first_run = False - self._was_ethernet_connected = state.ethernet_connected return - # Cable just plugged in while Wi-Fi is active -> disable Wi-Fi - if ( - state.ethernet_connected - and not self._was_ethernet_connected - and state.wifi_enabled - and not self._is_connecting - ): - logger.info("Ethernet connected — turning off Wi-Fi") - self._was_ethernet_connected = True - wifi_btn = self.wifi_button.toggle_button - hotspot_btn = self.hotspot_button.toggle_button - with QtCore.QSignalBlocker(wifi_btn): - wifi_btn.state = wifi_btn.State.OFF - with QtCore.QSignalBlocker(hotspot_btn): - hotspot_btn.state = hotspot_btn.State.OFF - self._nm.set_wifi_enabled(False) - self._sync_ethernet_panel(state) - self._emit_status_icon(state) - return - - self._was_ethernet_connected = state.ethernet_connected + # Exclusivity applies to user toggles only; a cable never kills the radio. # Ethernet panel visibility is pure hardware state (carrier + # connection) and must update even while a loading operation is @@ -419,9 +398,7 @@ def _on_network_state_changed(self, state: NetworkState) -> None: return # Normal (not connecting) display updates. - if state.ethernet_connected: - self._display_connected_state(state) - elif ( + if state.ethernet_connected or ( state.current_ssid and state.current_ip and state.connectivity @@ -573,7 +550,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 @@ -690,8 +667,7 @@ def _handle_first_run(self, state: NetworkState) -> None: hotspot_on = False if state.ethernet_connected: - if state.wifi_enabled: - self._nm.set_wifi_enabled(False) + # Display only: never force the radio off here, it is the recovery path. self._display_connected_state(state) elif state.connectivity == ConnectivityState.FULL and state.current_ssid: wifi_on = True @@ -734,6 +710,7 @@ def _sync_toggle_states(self, state: NetworkState) -> None: wifi_on = False hotspot_on = False + # One link at a time: the cable wins the display, then hotspot, then Wi-Fi. if state.ethernet_connected: pass elif state.hotspot_enabled: @@ -1007,6 +984,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 +1013,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 +1021,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 +1058,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 +1073,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..d740274a 100644 --- a/tests/network/test_network_ui.py +++ b/tests/network/test_network_ui.py @@ -178,11 +178,14 @@ def test_ethernet_connected_shows_connected_state(self, win): assert w.netlist_ssuid.isVisible() assert w.netlist_ssuid.text() == "Ethernet" - def test_ethernet_disables_wifi_if_enabled(self, win): + def test_ethernet_never_kills_radio_at_boot(self, win): + """Boot is display-only: the radio is the recovery path, never touched here.""" w, nm = win state = _eth_state(wifi_enabled=True) w._handle_first_run(state) - nm.set_wifi_enabled.assert_called_once_with(False) + nm.set_wifi_enabled.assert_not_called() + wifi_btn = w.wifi_button.toggle_button + assert wifi_btn.state == wifi_btn.State.OFF def test_ethernet_does_not_disable_wifi_if_already_off(self, win): w, nm = win @@ -490,7 +493,6 @@ def _prime(self, w): """Mark first-run as done so the normal display path runs.""" w._is_first_run = False w._is_connecting = False - w._was_ethernet_connected = False def test_ethernet_shows_connected(self, win): w, _ = win @@ -522,14 +524,13 @@ def test_first_run_flag_cleared_after_first_call(self, win): w._on_network_state_changed(_disconnected_state()) assert w._is_first_run is False - def test_ethernet_plug_disables_wifi(self, win): - """Ethernet cable plugged in during Wi-Fi session -> Wi-Fi disabled.""" + def test_ethernet_plug_keeps_wifi_enabled(self, win): + """Wi-Fi is the recovery path; a plugged cable must never kill the radio.""" w, nm = win self._prime(w) - w._was_ethernet_connected = False state = _eth_state(wifi_enabled=True) w._on_network_state_changed(state) - nm.set_wifi_enabled.assert_called_with(False) + nm.set_wifi_enabled.assert_not_called() # ───────────────────────────────────────────────────────────────────────────── @@ -698,7 +699,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 +746,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..7bace8e5 100644 --- a/tests/network/test_worker_unit.py +++ b/tests/network/test_worker_unit.py @@ -236,7 +236,6 @@ async def test_happy_path_sets_running(self, qapp): w = _make_worker(qapp, running=False) # Mock all async calls in initialize w._detect_interfaces = AsyncMock() - w._enforce_boot_mutual_exclusion = AsyncMock() w._is_ethernet_connected = AsyncMock(return_value=False) w._activate_saved_vlans = AsyncMock() w._start_signal_listeners = AsyncMock() @@ -752,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 @@ -1730,6 +1746,132 @@ async def test_unsupported_security_returns_error(self, qapp): assert result.error_code == "unsupported_security" +_BACKUP = {"connection": {"id": ("s", "Net")}, "802-11-wireless-security": {}} + + +def _add_ready_worker(qapp, *, backup=_BACKUP): + """Worker whose add path reaches NM, with the backup helpers mocked.""" + w = _make_worker(qapp) + w._wifi = _ProxyFactory(AsyncProxyMock(request_scan=AsyncMock(), interface="wlan0")) + w._find_ap_props = AsyncMock(return_value={"ssid": b"Net"}) + w._build_connection_properties = MagicMock(return_value={"connection": {}}) + w._backup_and_drop_existing = AsyncMock(return_value=backup) + w._restore_profile = AsyncMock(return_value=True) + w._delete_network_impl = AsyncMock() + w._reload_connections = AsyncMock() + w._wait_for_connection = AsyncMock(return_value=True) + w._nm_settings = _ProxyFactory( + AsyncProxyMock(add_connection=AsyncMock(return_value="/conn/9")) + ) + w._nm = _ProxyFactory(AsyncProxyMock(activate_connection=AsyncMock())) + return w + + +class TestAddNetworkBackup: + @pytest.mark.asyncio + 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) + assert result.error_code == "not_found" + w._backup_and_drop_existing.assert_not_awaited() + + @pytest.mark.asyncio + 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) + assert result.error_code == "unsupported_security" + w._backup_and_drop_existing.assert_not_awaited() + + @pytest.mark.asyncio + 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) + assert result.error_code == "add_failed" + w._restore_profile.assert_awaited_once_with("Net", _BACKUP) + + @pytest.mark.asyncio + 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) + assert result.error_code == "add_failed" + w._restore_profile.assert_not_awaited() + + @pytest.mark.asyncio + 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) + assert result.error_code == "auth_failed" + assert "previously saved password was kept" in result.message + w._delete_network_impl.assert_awaited_once_with("Net") + w._restore_profile.assert_awaited_once_with("Net", _BACKUP) + + @pytest.mark.asyncio + 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) + assert result.error_code == "auth_failed" + assert "saved profile has been removed" in result.message + w._restore_profile.assert_not_awaited() + + @pytest.mark.asyncio + 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) + 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) + assert result.success + w._delete_network_impl.assert_not_awaited() + w._restore_profile.assert_not_awaited() + + @pytest.mark.asyncio + async def test_drop_existing_skips_unknown_ssid(self, qapp): + w = _make_worker(qapp) + w._is_known = AsyncMock(return_value=False) + w._delete_network_impl = AsyncMock() + assert await w._backup_and_drop_existing("Net") is None + w._delete_network_impl.assert_not_awaited() + + @pytest.mark.asyncio + async def test_drop_existing_backs_up_then_deletes(self, qapp): + w = _make_worker(qapp) + w._saved_cache_dirty = False + w._is_known = AsyncMock(return_value=True) + w._backup_profile = AsyncMock(return_value=_BACKUP) + w._delete_network_impl = AsyncMock() + assert await w._backup_and_drop_existing("Net") == _BACKUP + w._delete_network_impl.assert_awaited_once_with("Net") + assert w._saved_cache_dirty is True + + @pytest.mark.asyncio + async def test_restore_readds_settings_and_dirties_cache(self, qapp): + w = _make_worker(qapp) + w._saved_cache_dirty = False + add = AsyncMock() + w._nm_settings = _ProxyFactory(AsyncProxyMock(add_connection=add)) + assert await w._restore_profile("Net", _BACKUP) is True + add.assert_awaited_once_with(_BACKUP) + assert w._saved_cache_dirty is True + + @pytest.mark.asyncio + async def test_restore_failure_returns_false(self, qapp): + w = _make_worker(qapp) + add = AsyncMock(side_effect=RuntimeError("nm down")) + w._nm_settings = _ProxyFactory(AsyncProxyMock(add_connection=add)) + assert await w._restore_profile("Net", _BACKUP) is False + + class TestDeleteConnectionsById: @pytest.mark.asyncio async def test_deletes_matching(self, qapp): @@ -2028,44 +2170,113 @@ 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 +class TestWiredProfilesAutoconnect: + """Device.Autoconnect dies on NM restart; only the profile flag persists.""" + + @staticmethod + def _wire_profiles(w, conn_type="802-3-ethernet", autoconnect=True): + nm_settings_proxy = AsyncProxyMock( + list_connections=AsyncMock(return_value=["/conn/eth"]) ) + w._nm_settings = _ProxyFactory(nm_settings_proxy) + settings = { + "connection": { + "type": ("s", conn_type), + "autoconnect": ("b", autoconnect), + "timestamp": ("t", 123), + }, + "ipv4": {"method": ("s", "auto")}, + } + w._gather_settings = AsyncMock(return_value=[("/conn/eth", settings)]) + conn_proxy = AsyncProxyMock(update=AsyncMock()) + w._conn_settings = lambda path: conn_proxy + return conn_proxy + + @pytest.mark.asyncio + async def test_disables_wired_profile(self, qapp): + w = _make_worker(qapp) + conn = self._wire_profiles(w, autoconnect=True) + await w._set_wired_profiles_autoconnect(False) + props = conn.update.await_args[0][0] + assert props["connection"]["autoconnect"] == ("b", False) + + @pytest.mark.asyncio + async def test_strips_timestamp_nm_will_not_accept(self, qapp): + w = _make_worker(qapp) + conn = self._wire_profiles(w, autoconnect=True) + await w._set_wired_profiles_autoconnect(False) + assert "timestamp" not in conn.update.await_args[0][0]["connection"] - def test_ethernet_active_wifi_on_disables_wifi(self, qapp): + @pytest.mark.asyncio + async def test_reenables_wired_profile(self, qapp): + w = _make_worker(qapp) + conn = self._wire_profiles(w, autoconnect=False) + await w._set_wired_profiles_autoconnect(True) + assert conn.update.await_args[0][0]["connection"]["autoconnect"] == ("b", True) + + @pytest.mark.asyncio + async def test_skips_when_already_correct(self, qapp): + w = _make_worker(qapp) + conn = self._wire_profiles(w, autoconnect=True) + await w._set_wired_profiles_autoconnect(True) + conn.update.assert_not_awaited() + + @pytest.mark.asyncio + async def test_ignores_non_ethernet_profiles(self, qapp): + w = _make_worker(qapp) + conn = self._wire_profiles(w, conn_type="802-11-wireless", autoconnect=True) + await w._set_wired_profiles_autoconnect(False) + conn.update.assert_not_awaited() + + @pytest.mark.asyncio + async def test_exception_is_non_fatal(self, qapp): + w = _make_worker(qapp) + w._nm_settings = MagicMock(side_effect=RuntimeError("boom")) + await w._set_wired_profiles_autoconnect(False) # must not raise + + +class TestEnsureWiredAutoconnect: + def test_no_wired_device_returns_early(self, qapp): + w = _make(qapp, wired=False) + wired = AsyncProxyMock(state=30, autoconnect=False) + _wire(w, wired_proxy=wired) + _run(w._ensure_wired_autoconnect()) + wired.autoconnect.set_async.assert_not_awaited() + + def test_autoconnect_off_is_rearmed(self, qapp): w = _make(qapp) - 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 + wired = AsyncProxyMock(state=30, autoconnect=False) + _wire(w, wired_proxy=wired) + _run(w._ensure_wired_autoconnect()) + wired.autoconnect.set_async.assert_awaited_once_with(True) - def test_ethernet_active_wifi_off_no_action(self, qapp): + def test_autoconnect_on_is_left_alone(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() + wired = AsyncProxyMock(state=100, autoconnect=True) + _wire(w, wired_proxy=wired) + _run(w._ensure_wired_autoconnect()) + wired.autoconnect.set_async.assert_not_awaited() + + def test_profiles_are_rearmed_too(self, qapp): + w = _make(qapp) + w._set_wired_profiles_autoconnect = AsyncMock() + _wire(w, wired_proxy=AsyncProxyMock(state=30, autoconnect=False)) + _run(w._ensure_wired_autoconnect()) + w._set_wired_profiles_autoconnect.assert_awaited_once_with(True) def test_exception_is_non_fatal(self, qapp): w = _make(qapp) - w._is_ethernet_connected = AsyncMock(side_effect=RuntimeError("boom")) - _run(w._enforce_boot_mutual_exclusion()) # must not raise + w._generic = MagicMock(side_effect=RuntimeError("boom")) + _run(w._ensure_wired_autoconnect()) # must not raise + + def test_wifi_radio_is_never_touched(self, qapp): + w = _make(qapp) + nm = AsyncProxyMock(wireless_enabled=True) + _wire(w, nm=nm) + wired = AsyncProxyMock(state=30, autoconnect=False) + _wire(w, wired_proxy=wired) + _run(w._ensure_wired_autoconnect()) + nm.wireless_enabled.set_async.assert_not_awaited() class TestWaitForWifiRadio: @@ -2105,17 +2316,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): @@ -2164,6 +2373,54 @@ def test_calls_disconnect(self, qapp): wired.disconnect.assert_awaited_once() w._deactivate_all_vlans.assert_awaited_once() + def test_persists_choice_in_the_profile(self, qapp): + w = _make(qapp) + wired = AsyncProxyMock() + wired.disconnect = AsyncMock() + _wire(w, wired_proxy=wired) + w._is_ethernet_connected = AsyncMock(return_value=False) + w._deactivate_all_vlans = AsyncMock() + w._set_wired_profiles_autoconnect = AsyncMock() + _run(w._async_disconnect_ethernet()) + w._set_wired_profiles_autoconnect.assert_awaited_once_with(False) + + def test_already_inactive_is_not_an_error(self, qapp): + w = _make(qapp) + wired = AsyncProxyMock() + wired.disconnect = AsyncMock( + side_effect=RuntimeError("This device is not active") + ) + _wire(w, wired_proxy=wired) + w._is_ethernet_connected = AsyncMock(return_value=False) + w._deactivate_all_vlans = AsyncMock() + with patch.object(_worker_mod.logger, "error") as err: + _run(w._async_disconnect_ethernet()) + err.assert_not_called() + + def test_persists_choice_even_when_already_inactive(self, qapp): + w = _make(qapp) + wired = AsyncProxyMock() + wired.disconnect = AsyncMock( + side_effect=RuntimeError("This device is not active") + ) + _wire(w, wired_proxy=wired) + w._is_ethernet_connected = AsyncMock(return_value=False) + w._deactivate_all_vlans = AsyncMock() + w._set_wired_profiles_autoconnect = AsyncMock() + _run(w._async_disconnect_ethernet()) + w._set_wired_profiles_autoconnect.assert_awaited_once_with(False) + + def test_persists_choice_even_when_teardown_fails(self, qapp): + w = _make(qapp) + wired = AsyncProxyMock() + wired.disconnect = AsyncMock(side_effect=RuntimeError("boom")) + _wire(w, wired_proxy=wired) + w._is_ethernet_connected = AsyncMock(return_value=False) + w._deactivate_all_vlans = AsyncMock() + w._set_wired_profiles_autoconnect = AsyncMock() + _run(w._async_disconnect_ethernet()) + w._set_wired_profiles_autoconnect.assert_awaited_once_with(False) + class TestConnectEthernetAsync: def test_no_wired_path_emits_error(self, qapp): @@ -2178,11 +2435,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 +2444,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 +2855,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 +2871,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