refactor(notifications): update the NotificationWidget when it is v…

…isible and a new notification with the same id is dispatched

CHANGELOG.md

@@ -5,6 +5,7 @@
 - refactor(core): use `dpkg-query` instead of `apt` python api as loading `Cache` in `apt` is slow and use it in docker service
 - refactor(system): add response for docker commands and service commands
 - feat(lightdm): add installation options for lightdm package
+- refactor(notifications): update the `NotificationWidget` when it is visible and a new notification with the same id is dispatched
 
 ## Version 0.15.9
 

scripts/deploy.sh

@@ -11,17 +11,17 @@ function cleanup() {
 trap cleanup ERR
 trap cleanup EXIT
 
-perl -i -pe 's/^(packages = \[.*)$/\1\nexclude = ["ubo_app\/services\/*-voice\/models\/*"]/' pyproject.toml
-poetry build
-cleanup
-
 LATEST_VERSION=$(basename $(ls -rt dist/*.whl | tail -n 1))
 deps=${deps:-"False"}
 bootstrap=${bootstrap:-"False"}
 run=${run:-"False"}
 restart=${restart:-"False"}
 env=${env:-"False"}
 
+perl -i -pe 's/^(packages = \[.*)$/\1\nexclude = ["ubo_app\/services\/*-voice\/models\/*"]/' pyproject.toml
+poetry build
+cleanup
+
 function run_on_pod() {
   if [ $# -lt 1 ]; then
     echo "Usage: run_on_pod <command>"

scripts/test_on_device.sh

@@ -4,6 +4,13 @@ set -o errexit
 set -o pipefail
 set -o nounset
 
+# Signal handler
+function cleanup() {
+  run_on_pod "killall -9 pytest"
+}
+trap cleanup ERR
+trap cleanup EXIT
+
 copy=${copy:-"False"}
 deps=${deps:-"False"}
 run=${run:-"False"}

tests/flows/test_wireless.py

@@ -2,16 +2,11 @@
 
 from __future__ import annotations
 
-from dataclasses import asdict
 from typing import TYPE_CHECKING
 
 import pytest
 from tenacity import wait_fixed
 
-from ubo_app.store.services.wifi import (
-    WiFiConnection,
-    WiFiState,
-)
 from ubo_app.utils import IS_RPI
 
 if TYPE_CHECKING:
@@ -26,6 +21,7 @@
     )
     from tests.fixtures.menu import WaitForEmptyMenu, WaitForMenuItem
     from ubo_app.store.main import RootState
+    from ubo_app.store.services.wifi import WiFiState
 
 
 @pytest.mark.skipif(not IS_RPI, reason='Only runs on Raspberry Pi')
@@ -39,42 +35,29 @@ async def test_wireless_flow(
     camera: MockCamera,
     wait_for_menu_item: WaitForMenuItem,
     wait_for_empty_menu: WaitForEmptyMenu,
+    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     """Test the wireless flow."""
+    from sdbus_async.networkmanager import (  # pyright: ignore [reportMissingModuleSource]
+        AccessPoint,
+    )
+
+    async def strength() -> int:
+        return 100
+
+    monkeypatch.setattr(
+        AccessPoint,
+        'strength',
+        property(lambda self: (self, strength())[1]),
+    )
+
     from ubo_app.menu_app.menu import MenuApp
     from ubo_app.store.core import ChooseMenuItemByIconEvent, ChooseMenuItemByLabelEvent
     from ubo_app.store.main import dispatch, store
     from ubo_app.store.services.keypad import Key, KeypadKeyPressAction
 
-    def store_snapshot_selector(state: RootState) -> WiFiState | None:
-        """Select the store snapshot."""
-        wifi_state = state.wifi
-
-        return WiFiState(
-            connections=[
-                WiFiConnection(
-                    **dict(
-                        asdict(connection),
-                        signal_strength=100 if connection.signal_strength > 0 else 0,
-                    ),
-                )
-                for connection in wifi_state.connections
-            ]
-            if wifi_state.connections is not None
-            else None,
-            state=wifi_state.state,
-            current_connection=WiFiConnection(
-                **dict(
-                    asdict(wifi_state.current_connection),
-                    signal_strength=100
-                    if wifi_state.current_connection.signal_strength > 0
-                    else 0,
-                ),
-            )
-            if wifi_state.current_connection is not None
-            else None,
-            has_visited_onboarding=wifi_state.has_visited_onboarding,
-        )
+    def store_snapshot_selector(state: RootState) -> WiFiState:
+        return state.wifi
 
     app = MenuApp()
     app_context.set_app(app)

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

@@ -64,7 +64,7 @@
       "status": "not_available"
     },
     "service": {
-      "status": "not_running",
+      "status": "not_installed",
       "usernames": {}
     }
   },
@@ -80,9 +80,9 @@
     "is_connected": true
   },
   "lightdm": {
-    "is_active": true,
-    "is_enabled": true,
-    "is_installed": true,
+    "is_active": false,
+    "is_enabled": false,
+    "is_installed": false,
     "is_installing": false
   },
   "main": {
@@ -469,7 +469,7 @@
                               1,
                               1
                             ],
-                            "icon": "[color=#008000]󰪥[/color]",
+                            "icon": "[color=#ffff00]󰝦[/color]",
                             "is_short": false,
                             "key": "lightdm",
                             "label": "LightDM",

ubo_app/menu_app/menu_notification_handler.py

@@ -3,17 +3,20 @@
 
 import functools
 from dataclasses import replace
+from typing import TYPE_CHECKING
 
 from kivy.clock import Clock, mainthread
 from ubo_gui.app import UboApp
 from ubo_gui.constants import DANGER_COLOR, INFO_COLOR
+from ubo_gui.menu.stack_item import StackApplicationItem
 from ubo_gui.notification import NotificationWidget
 from ubo_gui.page import PAGE_MAX_ITEMS
 
 from ubo_app.menu_app.notification_info import NotificationInfo
 from ubo_app.store.core import CloseApplicationEvent, OpenApplicationEvent
 from ubo_app.store.main import dispatch, subscribe_event
 from ubo_app.store.services.notifications import (
+    Notification,
     NotificationActionItem,
     NotificationDisplayType,
     NotificationsClearAction,
@@ -22,20 +25,32 @@
 )
 from ubo_app.store.services.voice import VoiceReadTextAction
 
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+    from ubo_gui.menu.menu_widget import MenuWidget
+
 
 class MenuNotificationHandler(UboApp):
+    menu_widget: MenuWidget
+
     @mainthread
     def display_notification(  # noqa: C901
         self: MenuNotificationHandler,
         event: NotificationsDisplayEvent,
     ) -> None:
-        def run_notification_action(action: NotificationActionItem) -> None:
-            result = action.action()
-            if action.dismiss_notification:
-                dismiss()
-            else:
-                close()
-            return result
+        if (
+            event.notification.id
+            and any(
+                isinstance(stack_item, StackApplicationItem)
+                and isinstance(stack_item.application, NotificationWidget)
+                and stack_item.application.notification_id == event.notification.id
+                for stack_item in self.menu_widget.stack
+            )
+        ) or event.notification.display_type is NotificationDisplayType.BACKGROUND:
+            return
+
+        subscriptions = []
 
         notification = event.notification
         is_closed = False
@@ -46,20 +61,79 @@ def close(_: object = None) -> None:
             if is_closed:
                 return
             is_closed = True
-            unsubscribe()
+            for unsubscribe in subscriptions:
+                unsubscribe()
             notification_application.unbind(on_close=close)
             dispatch(CloseApplicationEvent(application=notification_application))
             if notification.dismiss_on_close:
                 dispatch(NotificationsClearAction(notification=notification))
             if notification.on_close:
                 notification.on_close()
 
+        notification_application = NotificationWidget(
+            notification_title=notification.title,
+            content=notification.content,
+            icon=notification.icon,
+            color=notification.color,
+            items=self._notification_items(notification, close),
+            title=f'Notification ({event.index + 1}/{event.count})'
+            if event.index is not None
+            else ' ',
+        )
+        notification_application.notification_id = notification.id
+
+        dispatch(OpenApplicationEvent(application=notification_application))
+
+        if notification.display_type is NotificationDisplayType.FLASH:
+            Clock.schedule_once(close, notification.flash_time)
+
+        notification_application.bind(on_close=close)
+
+        @mainthread
+        def clear_notification(event: NotificationsClearEvent) -> None:
+            if event.notification == notification:
+                close()
+
+        def renew_notification(event: NotificationsDisplayEvent) -> None:
+            nonlocal notification
+            if event.notification.id == notification.id:
+                notification = event.notification
+                self._update_notification_widget(notification_application, event, close)
+
+        subscriptions.append(
+            subscribe_event(
+                NotificationsClearEvent,
+                clear_notification,
+            ),
+        )
+        if notification.id is not None:
+            subscriptions.append(
+                subscribe_event(
+                    NotificationsDisplayEvent,
+                    renew_notification,
+                    keep_ref=False,
+                ),
+            )
+
+    def _notification_items(
+        self: MenuNotificationHandler,
+        notification: Notification,
+        close: Callable[[], None],
+    ) -> list[NotificationActionItem | None]:
         def dismiss(_: object = None) -> None:
             close()
             if not notification.dismiss_on_close:
                 dispatch(NotificationsClearAction(notification=notification))
 
-        items = []
+        def run_notification_action(action: NotificationActionItem) -> None:
+            result = action.action()
+            if action.dismiss_notification:
+                dismiss()
+            else:
+                close()
+            return result
+
+        items: list[NotificationActionItem | None] = []
 
         if notification.extra_information:
             text = notification.extra_information.text
@@ -106,32 +180,25 @@ def open_info() -> None:
                 ),
             )
 
-        items = [None] * (PAGE_MAX_ITEMS - len(items)) + items
+        return [None] * (PAGE_MAX_ITEMS - len(items)) + items
 
-        notification_application = NotificationWidget(
-            notification_title=notification.title,
-            content=notification.content,
-            icon=notification.icon,
-            color=notification.color,
-            items=items,
-            title=f'Notification ({event.index + 1}/{event.count})'
-            if event.index is not None
-            else ' ',
+    @mainthread
+    def _update_notification_widget(
+        self: MenuNotificationHandler,
+        notification_application: NotificationWidget,
+        event: NotificationsDisplayEvent,
+        close: Callable[[], None],
+    ) -> None:
+        notification_application.notification_title = event.notification.title
+        notification_application.content = event.notification.content
+        notification_application.icon = event.notification.icon
+        notification_application.color = event.notification.color
+        notification_application.items = self._notification_items(
+            event.notification,
+            close,
         )
-
-        dispatch(OpenApplicationEvent(application=notification_application))
-
-        if notification.display_type is NotificationDisplayType.FLASH:
-            Clock.schedule_once(close, notification.flash_time)
-
-        notification_application.bind(on_close=close)
-
-        @mainthread
-        def clear_notification(event: NotificationsClearEvent) -> None:
-            if event.notification == notification:
-                close()
-
-        unsubscribe = subscribe_event(
-            NotificationsClearEvent,
-            clear_notification,
+        notification_application.title = (
+            f'Notification ({event.index + 1}/{event.count})'
+            if event.index is not None
+            else ' '
         )