core: services: wifi: WifiManager: Start without a wlan interface

core/frontend/src/components/wifi/WifiManager.vue

@@ -80,8 +80,13 @@
             flat
             class="text-body-1 text-center"
           >
-            No wifi networks available :( <br>
-            Rescanning...
+            <template v-if="wifi_status?.state === 'unavailable'">
+              No wifi adapter
+            </template>
+            <template v-else>
+              No wifi networks available :( <br>
+              Rescanning...
+            </template>
           </v-card-text>
         </div>
         <div v-else>

core/frontend/src/components/wifi/WifiUpdater.vue

@@ -40,6 +40,15 @@ export default Vue.extend({
         timeout: 10000,
       })
         .then((response) => {
+          if (response.data.state === 'unavailable') {
+            wifi.setNetworkStatus(response.data)
+            wifi.setCurrentNetwork(null)
+            wifi.setAvailableNetworks([])
+            this.fetch_network_status_task.setDelay(30000)
+            return
+          }
+
+          this.fetch_network_status_task.setDelay(5000)
           wifi.setNetworkStatus(response.data)
 
           if (response.data.wpa_state !== 'COMPLETED') {

core/frontend/src/types/wifi.ts

@@ -27,6 +27,7 @@ export interface WifiStatus {
     p2p_device_address: string
     address: string
     uuid: string
+    state?: string
 }
 
 export interface WPANetwork {

core/services/wifi/test_wifi_manager.py

@@ -0,0 +1,46 @@
+import sys
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+_WIFI_DIR = str(Path(__file__).resolve().parent)
+sys.path.insert(0, _WIFI_DIR)
+_saved = {name: sys.modules.get(name) for name in ("exceptions", "settings", "typedefs")}
+for name in _saved:
+    sys.modules.pop(name, None)
+
+from wifi_handlers.wpa_supplicant.WifiManager import WifiManager
+
+for name, module in _saved.items():
+    if module is not None:
+        sys.modules[name] = module
+    else:
+        sys.modules.pop(name, None)
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_reports_unavailable_without_a_socket() -> None:
+    manager = WifiManager.__new__(WifiManager)
+
+    assert (await manager.status()).state == "unavailable"
+    assert await manager.get_wifi_available() == []
+    assert await manager.get_saved_wifi_network() == []
+    assert await manager.supports_hotspot() is False
+    assert await manager.hotspot_is_running() is False
+
+
+async def test_reports_available_after_connecting(monkeypatch: pytest.MonkeyPatch) -> None:
+    manager = WifiManager.__new__(WifiManager)
+    monkeypatch.setattr(manager.wpa, "run", lambda _target: None)
+    monkeypatch.setattr(WifiManager, "get_wifi_available", _no_networks)
+
+    # The udp socket is the fallback when there is no wpa_supplicant socket, it still means wifi works
+    await manager.connect(("127.0.0.1", 6664))
+
+    assert manager.wpa_path is not None
+
+
+async def _no_networks(_self: Any) -> list[Any]:
+    return []

core/services/wifi/wifi_handlers/networkmanager/networkmanager.py

@@ -420,7 +420,7 @@ async def supports_hotspot(self) -> bool:
 
     async def status(self) -> WifiStatus:
         if not self._device_path:
-            return WifiStatus(state="disconnected")
+            return WifiStatus(state="unavailable")
 
         device = NetworkDeviceWireless(self._device_path, self._bus)
         state = await device.state

core/services/wifi/wifi_handlers/wpa_supplicant/WifiManager.py

@@ -28,6 +28,7 @@
 class WifiManager(AbstractWifiManager):
     wpa = WPASupplicant()
     wpa_path: Optional[str] = None
+    _hotspot: Optional[HotspotManager] = None
 
     async def can_work(self) -> bool:
         return bool(get_host_os() == HostOs.Bullseye)
@@ -94,6 +95,8 @@ async def connect(self, path: Any) -> None:
             path {[tuple/str]} -- Can be a tuple to connect (ip/port) or unix socket file
         """
         self.wpa.run(path)
+        # run only returns when the socket is connected, before that there is no adapter to talk to
+        self.wpa_path = str(path)
         self._scan_task: Optional[asyncio.Task[bytes]] = None
         self._updated_scan_results: Optional[List[ScannedWifiNetwork]] = None
         self._ignored_reconnection_networks: List[str] = []
@@ -104,7 +107,6 @@ async def connect(self, path: Any) -> None:
         await self.get_wifi_available()
 
         try:
-            self._hotspot: Optional[HotspotManager] = None
             ssid, password = (
                 self._settings_manager.settings.hotspot_ssid,
                 self._settings_manager.settings.hotspot_password,
@@ -192,6 +194,8 @@ def hotspot(self) -> HotspotManager:
 
     async def get_wifi_available(self) -> List[ScannedWifiNetwork]:
         """Get a dict from the wifi signals available"""
+        if self.wpa_path is None:
+            return []
 
         async def perform_new_scan() -> None:
             try:
@@ -231,6 +235,8 @@ async def perform_new_scan() -> None:
 
     async def get_saved_wifi_network(self) -> List[SavedWifiNetwork]:
         """Get a list of saved wifi networks"""
+        if self.wpa_path is None:
+            return []
         try:
             data = await self.wpa.send_command_list_networks()
             networks_list = WifiManager.__dict_from_table(data)
@@ -331,6 +337,8 @@ async def connect_to_network(self, network_id: int, timeout: float = 20.0) -> No
 
     async def status(self) -> WifiStatus:
         """Check wpa_supplicant status"""
+        if self.wpa_path is None:
+            return WifiStatus(state="unavailable")
         try:
             data = await self.wpa.send_command_status()
             return WifiStatus(**WifiManager.__dict_from_list(data))
@@ -473,6 +481,9 @@ async def set_hotspot_credentials(self, credentials: WifiCredentials) -> None:
         self._settings_manager.settings.hotspot_password = credentials.password
         self._settings_manager.save()
 
+        if self.wpa_path is None:
+            return
+
         self.hotspot.set_credentials(credentials)
 
         if self.hotspot.is_running():
@@ -481,6 +492,9 @@ async def set_hotspot_credentials(self, credentials: WifiCredentials) -> None:
             await self.enable_hotspot(save_settings=False)
 
     def hotspot_credentials(self) -> WifiCredentials:
+        if self.wpa_path is None:
+            settings = self._settings_manager.settings
+            return WifiCredentials(ssid=settings.hotspot_ssid or "", password=settings.hotspot_password or "")
         credentials: WifiCredentials = self.hotspot.credentials
         return credentials
 
@@ -559,22 +573,26 @@ def is_socket(file_path: str) -> bool:
                 socket_name = available_sockets[-1]
                 logger.info(f"Going to use {socket_name} file")
             WLAN_SOCKET = os.path.join(wpa_socket_folder, socket_name)
-            self.wpa_path = WLAN_SOCKET
             await self.connect(WLAN_SOCKET)
         except Exception as socket_connection_error:
             logger.warning(f"Could not connect with wifi socket. {socket_connection_error}")
             logger.info("Connecting via internet wifi socket.")
             try:
                 await self.connect(("127.0.0.1", 6664))
             except Exception as udp_connection_error:
-                logger.error(f"Could not connect with internet socket: {udp_connection_error}. Exiting.")
-                raise udp_connection_error
+                logger.error(f"Could not connect with internet socket: {udp_connection_error}.")
+                self.wpa_path = None
+                return
         loop = asyncio.get_event_loop()
         loop.create_task(self.auto_reconnect(60))
         loop.create_task(self.start_hotspot_watchdog())
 
     async def supports_hotspot(self) -> bool:
+        if self.wpa_path is None:
+            return False
         return self.hotspot.supports_hotspot
 
     async def hotspot_is_running(self) -> bool:
+        if self.wpa_path is None:
+            return False
         return self.hotspot.is_running()