diff --git a/scripts/utils/wifi_monitor_daemon.py b/scripts/utils/wifi_monitor_daemon.py index 53f0497c..414c16dc 100755 --- a/scripts/utils/wifi_monitor_daemon.py +++ b/scripts/utils/wifi_monitor_daemon.py @@ -78,21 +78,17 @@ class WiFiMonitorDaemon: while self.running: try: - # Get current status before checking - status = self.wifi_manager.get_wifi_status() - ethernet_connected = self.wifi_manager._is_ethernet_connected() - - # Check WiFi status and manage AP mode - state_changed = self.wifi_manager.check_and_manage_ap_mode() - - # Get updated status after check - updated_status = self.wifi_manager.get_wifi_status() - updated_ethernet = self.wifi_manager._is_ethernet_connected() - + # One combined check that also returns the state it observed — + # the previous flow fetched status before AND after the check + # on top of the check's own internal fetch, each one several + # nmcli subprocess forks, every 30s, forever. + (state_changed, updated_status, updated_ethernet, + ap_active) = self.wifi_manager.check_and_manage_ap_mode_with_state() + current_state = { 'connected': updated_status.connected, 'ethernet_connected': updated_ethernet, - 'ap_active': updated_status.ap_mode_active, + 'ap_active': ap_active, 'ssid': updated_status.ssid } @@ -109,7 +105,7 @@ class WiFiMonitorDaemon: else: logger.debug("Ethernet not connected") - if updated_status.ap_mode_active: + if ap_active: logger.info(f"AP mode ACTIVE - SSID: {ap_ssid} (IP: 192.168.4.1)") else: logger.debug("AP mode inactive") @@ -123,16 +119,16 @@ class WiFiMonitorDaemon: # Log periodic status (less verbose) if updated_status.connected: logger.debug(f"Status check: WiFi={updated_status.ssid} ({updated_status.signal}%), " - f"Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}") + f"Ethernet={updated_ethernet}, AP={ap_active}") else: - logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}") + logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={ap_active}") # Escalating recovery: if nmcli reports connected but actual internet # is unreachable for several consecutive checks, restart NetworkManager. # This is done HERE (not inside check_and_manage_ap_mode) to keep the # AP-enable trigger clean and avoid false-positive AP enables from # transient packet loss on otherwise working WiFi. - if updated_status.connected and not updated_status.ap_mode_active: + if updated_status.connected and not ap_active: if not self.wifi_manager.check_internet_connectivity(): self._consecutive_internet_failures += 1 logger.warning( diff --git a/src/wifi_manager.py b/src/wifi_manager.py index 0983ff3f..808767ff 100644 --- a/src/wifi_manager.py +++ b/src/wifi_manager.py @@ -2564,11 +2564,35 @@ address=/detectportal.firefox.com/192.168.4.1 Returns: True if AP mode state changed, False otherwise """ + changed, _status, _ethernet, _ap = self.check_and_manage_ap_mode_with_state() + return changed + + def check_and_manage_ap_mode_with_state(self) -> Tuple[bool, WiFiStatus, bool, bool]: + """Like check_and_manage_ap_mode, but also returns the state it + observed, so callers (the wifi monitor daemon) don't have to re-run + the same nmcli subprocess battery before AND after the check — + each status fetch is several process forks. + + Returns: + (state_changed, WiFiStatus, ethernet_connected, ap_active_after) + """ try: # Get status with retry for more reliable detection status = self._get_wifi_status_with_retry() ethernet_connected = self._is_ethernet_connected() ap_active = self._is_ap_mode_active() + changed = self._manage_ap_mode(status, ethernet_connected, ap_active) + # State only ever changes via one enable or one disable, so the + # post-state is the inverse of the pre-state when changed. + ap_after = (not ap_active) if changed else ap_active + return changed, status, ethernet_connected, ap_after + except Exception as e: + logger.error(f"Error checking AP mode: {e}", exc_info=True) + return False, WiFiStatus(connected=False), False, False + + def _manage_ap_mode(self, status: WiFiStatus, ethernet_connected: bool, ap_active: bool) -> bool: + """AP-mode decision logic against an already-fetched state snapshot.""" + try: auto_enable = self.config.get("auto_enable_ap_mode", True) # Default: True (safe due to grace period) # Log current state for debugging diff --git a/test/test_wifi_check_state.py b/test/test_wifi_check_state.py new file mode 100644 index 00000000..349e80a0 --- /dev/null +++ b/test/test_wifi_check_state.py @@ -0,0 +1,92 @@ +"""Tests for check_and_manage_ap_mode_with_state (src/wifi_manager.py). + +The wifi monitor daemon used to fetch WiFi status before AND after each +check on top of the check's own internal fetch — every fetch is several +nmcli subprocess forks, every 30s, forever. The new API returns the state +the check observed, so the daemon runs exactly one fetch battery per tick. +""" + +import os +import sys +from unittest.mock import MagicMock, patch + +import pytest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from src.wifi_manager import WiFiManager, WiFiStatus # noqa: E402 + + +@pytest.fixture +def wm(tmp_path): + with patch.object(WiFiManager, "_load_config", return_value={}, create=True): + manager = WiFiManager.__new__(WiFiManager) + # minimal attribute setup without running the real __init__ + manager.config = {"auto_enable_ap_mode": True} + manager._disconnected_checks = 0 + manager._disconnected_checks_required = 3 + manager._ap_enabled_at = None + return manager + + +def _wire(wm, connected, ethernet, ap_active): + wm._get_wifi_status_with_retry = MagicMock( + return_value=WiFiStatus(connected=connected, ssid="net" if connected else None)) + wm._is_ethernet_connected = MagicMock(return_value=ethernet) + wm._is_ap_mode_active = MagicMock(return_value=ap_active) + wm.enable_ap_mode = MagicMock(return_value=(True, "ok")) + wm.disable_ap_mode = MagicMock(return_value=(True, "ok")) + wm.scan_networks = MagicMock(return_value=([], False)) + wm._save_cached_scan = MagicMock() + wm._FORCE_AP_FLAG_PATH = MagicMock() + wm._FORCE_AP_FLAG_PATH.exists.return_value = False + + +class TestWithState: + def test_single_fetch_per_call(self, wm): + _wire(wm, connected=True, ethernet=False, ap_active=False) + wm.check_and_manage_ap_mode_with_state() + assert wm._get_wifi_status_with_retry.call_count == 1 + assert wm._is_ethernet_connected.call_count == 1 + assert wm._is_ap_mode_active.call_count == 1 + + def test_returns_observed_state(self, wm): + _wire(wm, connected=True, ethernet=True, ap_active=False) + changed, status, ethernet, ap_after = wm.check_and_manage_ap_mode_with_state() + assert changed is False + assert status.connected is True + assert ethernet is True + assert ap_after is False + + def test_ap_after_inverts_on_disable(self, wm): + """WiFi reconnects while AP is up -> auto-disable -> ap_after False.""" + _wire(wm, connected=True, ethernet=False, ap_active=True) + changed, _status, _ethernet, ap_after = wm.check_and_manage_ap_mode_with_state() + assert changed is True + assert ap_after is False + wm.disable_ap_mode.assert_called_once() + + def test_ap_after_inverts_on_enable(self, wm): + """Grace period exhausted with nothing connected -> enable -> True.""" + _wire(wm, connected=False, ethernet=False, ap_active=False) + wm._disconnected_checks = 2 # this call is the 3rd + changed, _status, _ethernet, ap_after = wm.check_and_manage_ap_mode_with_state() + assert changed is True + assert ap_after is True + wm.enable_ap_mode.assert_called_once() + + def test_bool_wrapper_is_back_compatible(self, wm): + _wire(wm, connected=True, ethernet=False, ap_active=False) + assert wm.check_and_manage_ap_mode() is False + _wire(wm, connected=True, ethernet=False, ap_active=True) + assert wm.check_and_manage_ap_mode() is True + + def test_exception_path_never_raises(self, wm): + wm._get_wifi_status_with_retry = MagicMock(side_effect=RuntimeError("nmcli gone")) + changed, status, _ethernet, _ap_after = wm.check_and_manage_ap_mode_with_state() + assert changed is False + assert status.connected is False + + +if __name__ == "__main__": + sys.exit(pytest.main([__file__, "-v"]))