refactor(tests): wait for wifi status icons in test_wireless

CHANGELOG.md

@@ -13,6 +13,7 @@
 - fix(keypad): keypad becoming unresponsive if a key was pressed while the app was loading - closes #118
 - fix(camera): closing the camera viewfinder will close the picamera instance so that it can be used again
 - refactor(core): use python-fake for faking
+- refactor(tests): wait for wifi status icons in `test_wireless`
 
 ## Version 0.15.4
 

pyproject.toml

@@ -26,7 +26,7 @@ headless-kivy = [
     "test",
   ] },
 ]
-python-redux = "^0.15.9"
+python-redux = "^0.15.10"
 pyzbar = "^0.1.9"
 sdbus-networkmanager = { version = "^2.0.0", markers = "platform_machine=='aarch64'" }
 rpi_ws281x = { version = "^5.0.0", markers = "platform_machine=='aarch64'" }

tests/flows/results/test_wireless/wireless_flow/store-rpi-003.jsonc

@@ -1,2 +1,23 @@
 // store-rpi-003
-null
+{
+  "connections": [
+    {
+      "hidden": false,
+      "password": null,
+      "signal_strength": 100,
+      "ssid": "ubo-test-ssid",
+      "state": "Connected",
+      "type": null
+    }
+  ],
+  "current_connection": {
+    "hidden": false,
+    "password": null,
+    "signal_strength": 100,
+    "ssid": "ubo-test-ssid",
+    "state": "Connected",
+    "type": null
+  },
+  "has_visited_onboarding": true,
+  "state": "Connected"
+}

tests/flows/results/test_wireless/wireless_flow/window-rpi-006.hash

@@ -1,2 +1,2 @@
 // window-rpi-006
-e0ab61ea525c3fcedc7232fd5b60c094847602f2a7dae50affc5cd31ffb1a1f4
+6aa0aa68a5e7047343d85d3099ac0efbbd9ca72399be7959659a4f39fa2009d5

tests/flows/test_wireless.py

@@ -9,8 +9,6 @@
 from tenacity import wait_fixed
 
 from ubo_app.store.services.wifi import (
-    ConnectionState,
-    GlobalWiFiState,
     WiFiConnection,
     WiFiState,
 )
@@ -50,25 +48,10 @@ async def test_wireless_flow(
     from ubo_app.store.main import dispatch, store
     from ubo_app.store.services.keypad import Key, KeypadKeyPressAction
 
-    is_adding_connection = False
-
     def store_snapshot_selector(state: RootState) -> WiFiState | None:
         """Select the store snapshot."""
         wifi_state = state.wifi
 
-        if is_adding_connection and wifi_state.connections == []:
-            return None
-
-        if (
-            any(
-                connection.state is ConnectionState.CONNECTING
-                for connection in wifi_state.connections or []
-            )
-            or wifi_state.state is GlobalWiFiState.PENDING
-        ):
-            # Connecting state does not happen consistently in multiple runs
-            return None
-
         return WiFiState(
             connections=[
                 WiFiConnection(
@@ -99,8 +82,8 @@ def store_snapshot_selector(state: RootState) -> WiFiState | None:
     app_context.set_app(app)
     load_services(['camera', 'wifi', 'notifications'])
 
-    @wait_for(timeout=5.0, wait=wait_fixed(0.5), run_async=True)
-    def check_icon() -> None:
+    @wait_for(timeout=20.0, wait=wait_fixed(1), run_async=True)
+    def check_icon(expected_icon: str) -> None:
         state = store._state  # noqa: SLF001
 
         assert state is not None
@@ -111,10 +94,10 @@ def check_icon() -> None:
         )
 
         assert icon is not None, 'wifi icon not registered'
-        assert icon.symbol != '󰖩', 'wifi is already connected'
-        assert icon.symbol == '󰖪', f'unexpected wifi icon {icon.symbol}'
+        assert icon.symbol == expected_icon
+
+    await check_icon('󰖪')
 
-    await check_icon()
     await stability()
     store_snapshot.take(selector=store_snapshot_selector)
 
@@ -151,25 +134,22 @@ def check_icon() -> None:
     # Set QR Code image of the WiFi credentials before camera is started
     camera.set_image('qrcode/wifi')
 
-    is_adding_connection = True
-
     # Select "QR code" to scan a QR code for credentials
     dispatch(ChooseMenuItemByIconEvent(icon='󰄀'))
 
     # Success notification should be shown
     window_snapshot.take()
 
     # Dismiss the notification informing the user that the connection was added
+    await check_icon('󰤨')
     await wait_for_menu_item(label='', icon='󰆴')
     dispatch(ChooseMenuItemByIconEvent(icon='󰆴'))
     await stability()
 
-    is_adding_connection = False
-
     # Select "Select" to open the wireless connection list and see the new connection
     dispatch(ChooseMenuItemByLabelEvent(label='Select'))
 
-    @wait_for(timeout=10.0, wait=wait_fixed(0.5), run_async=True)
+    @wait_for(timeout=20.0, wait=wait_fixed(1), run_async=True)
     def check_connections() -> None:
         state = store._state  # noqa: SLF001
 
@@ -185,31 +165,34 @@ def check_connections() -> None:
     dispatch(ChooseMenuItemByLabelEvent(label='ubo-test-ssid'))
 
     # Wait for the "Disconnect" item to show up
-    await wait_for_menu_item(label='Disconnect')
+    await wait_for_menu_item(label='Disconnect', timeout=10)
     await stability()
     window_snapshot.take()
     dispatch(ChooseMenuItemByLabelEvent(label='Disconnect'))
 
     # Wait for the "Connect" item to show up
-    await wait_for_menu_item(label='Connect')
+    await wait_for_menu_item(label='Connect', timeout=10)
+    await check_icon('󰖪')
     await stability()
     store_snapshot.take(selector=store_snapshot_selector)
     window_snapshot.take()
     dispatch(ChooseMenuItemByLabelEvent(label='Connect'))
 
-    await wait_for_menu_item(label='Disconnect')
+    await wait_for_menu_item(label='Disconnect', timeout=10)
+    await check_icon('󰤨')
     await stability()
     store_snapshot.take(selector=store_snapshot_selector)
     window_snapshot.take()
     dispatch(ChooseMenuItemByLabelEvent(label='Delete'))
 
-    @wait_for(timeout=10.0, wait=wait_fixed(0.5), run_async=True)
+    @wait_for(timeout=20.0, wait=wait_fixed(1), run_async=True)
     def check_no_connections() -> None:
         state = store._state  # noqa: SLF001
         assert state
         assert state.wifi.connections == []
 
     await check_no_connections()
+    await check_icon('󰖪')
     store_snapshot.take(selector=store_snapshot_selector)
 
     # Dismiss the notification informing the user that the connection was deleted

tests/integration/results/test_services/all_services_register/store-desktop-000.jsonc

@@ -639,7 +639,16 @@
     "is_access_key_set": false
   },
   "wifi": {
-    "connections": [],
+    "connections": [
+      {
+        "hidden": false,
+        "password": null,
+        "signal_strength": "<function:>",
+        "ssid": "<function:>",
+        "state": "Unknown",
+        "type": null
+      }
+    ],
     "current_connection": null,
     "has_visited_onboarding": true,
     "state": "Disconnected"

tests/setup.sh

@@ -3,11 +3,8 @@
 IS_RPI=false
 
 # Check if the script is running on a Raspberry Pi
-if [ -e /etc/os-release ]; then
-    source /etc/os-release
-    if [[ $ID == "raspbian" ]]; then
-        IS_RPI=true
-    fi
+if [ -e /etc/rpi-issue ]; then
+  IS_RPI=true
 fi
 
 if [ "$IS_RPI" = true ]; then

ubo_app/services/030-wifi/pages/main.py

@@ -6,7 +6,7 @@
 from constants import get_signal_icon
 from debouncer import DebounceOptions, debounce
 from kivy.clock import mainthread
-from kivy.properties import BooleanProperty
+from kivy.properties import StringProperty
 from ubo_gui.menu.types import (
     ActionItem,
     ApplicationItem,
@@ -19,6 +19,7 @@
     disconnect_wireless_connection,
     forget_wireless_connection,
     get_active_connection_ssid,
+    get_active_connection_state,
     get_wifi_device,
 )
 
@@ -39,12 +40,12 @@
 
 class WiFiConnectionPage(PromptWidget):
     ssid: str
-    is_active = BooleanProperty(defaultvalue=None)
+    state: ConnectionState = StringProperty(defaultvalue=ConnectionState.UNKNOWN)
 
     def first_option_callback(self: WiFiConnectionPage) -> None:
-        if self.is_active:
+        if self.state is ConnectionState.CONNECTED:
             create_task(disconnect_wireless_connection())
-        else:
+        elif self.state is ConnectionState.DISCONNECTED:
             create_task(connect_wireless_connection(self.ssid))
         dispatch(WiFiUpdateRequestAction(reset=True))
 
@@ -56,38 +57,56 @@ def second_option_callback(self: WiFiConnectionPage) -> None:
         )
 
     def update(self: WiFiConnectionPage, *_: tuple[Any, ...]) -> None:
-        self.first_option_background_color = (
-            PromptWidget.first_option_background_color.defaultvalue
-        )
-        if self.is_active:
+        if self.state is ConnectionState.CONNECTED:
             self.first_option_label = 'Disconnect'
             self.first_option_icon = '󰖪'
+            self.first_option_color = 'black'
+            self.first_option_background_color = (
+                PromptWidget.first_option_background_color.defaultvalue
+            )
             self.icon = '󰖩'
-        else:
+        elif self.state is ConnectionState.DISCONNECTED:
             self.first_option_label = 'Connect'
             self.first_option_icon = '󰖩'
+            self.first_option_color = 'black'
+            self.first_option_background_color = (
+                PromptWidget.first_option_background_color.defaultvalue
+            )
             self.icon = '󰖪'
+        elif self.state is ConnectionState.CONNECTING:
+            self.first_option_label = 'Connecting...'
+            self.first_option_icon = ''
+            self.first_option_color = 'white'
+            self.first_option_background_color = 'black'
+            self.icon = ''
+        elif self.state is ConnectionState.UNKNOWN:
+            self.first_option_label = ''
+            self.first_option_icon = ''
+            self.first_option_color = 'white'
+            self.first_option_background_color = 'black'
+            self.icon = ''
 
     def __init__(self: WiFiConnectionPage, **kwargs: object) -> None:
         super().__init__(**kwargs, items=None)
         self.prompt = f'SSID: {self.ssid}'
-        self.icon = ''
-        self.first_option_background_color = 'black'
-        self.first_option_label = ''
         self.first_option_is_short = False
         self.second_option_label = 'Delete'
         self.second_option_icon = '󰆴'
         self.second_option_is_short = False
 
-        self.bind(is_active=self.update)
+        self.bind(state=self.update)
+        self.update()
 
         @debounce(
             wait=0.5,
-            options=DebounceOptions(leading=False, trailing=True, time_window=2),
+            options=DebounceOptions(leading=False, trailing=True, time_window=0.5),
         )
         async def update_status() -> None:
-            is_active = await get_active_connection_ssid() == self.ssid
-            mainthread(lambda: setattr(self, 'is_active', is_active))()
+            if await get_active_connection_ssid() == self.ssid:
+                state = await get_active_connection_state()
+            else:
+                state = ConnectionState.DISCONNECTED
+            mainthread(lambda: setattr(self, 'state', state))()
 
         create_task(update_status())
 

ubo_app/services/030-wifi/setup.py

@@ -39,7 +39,7 @@
 
 @debounce(
     wait=0.5,
-    options=DebounceOptions(leading=True, trailing=False, time_window=2),
+    options=DebounceOptions(leading=True, trailing=False, time_window=0.5),
 )
 async def update_wifi_list(_: WiFiUpdateRequestEvent | None = None) -> None:
     connections = await get_connections()