Merge pull request #198 from samschott/get-real-current-notifications

Get real current notifications on macOS

Author: samschott

src/desktop_notifier/backends/macos.py

@@ -33,6 +33,7 @@
 foundation = load_library("Foundation")
 uns = load_library("UserNotifications")
 
+UNNotification = ObjCClass("UNNotification")
 UNUserNotificationCenter = ObjCClass("UNUserNotificationCenter")
 UNMutableNotificationContent = ObjCClass("UNMutableNotificationContent")
 UNNotificationRequest = ObjCClass("UNNotificationRequest")
@@ -195,6 +196,28 @@ def handler(settings: objc_id) -> None:
 
         return authorized
 
+    async def get_current_notifications(self) -> list[str]:
+        future: Future[list[UNNotification]] = Future()  # type:ignore[valid-type]
+
+        def handler(notifications: objc_id) -> None:
+            notifications = py_from_ns(notifications)
+            for notification in notifications:
+                notification.retain()
+            future.set_result(notifications)
+
+        self.nc.getDeliveredNotificationsWithCompletionHandler(handler)
+
+        notifications = await asyncio.wrap_future(future)
+        identifiers = [
+            str(n.request.identifier)  # type:ignore[attr-defined]
+            for n in notifications
+        ]
+
+        for notification in notifications:
+            notification.autorelease()  # type:ignore[attr-defined]
+
+        return identifiers
+
     async def _send(self, notification: Notification) -> None:
         """
         Uses UNUserNotificationCenter to schedule a notification.

tests/test_api.py

@@ -1,4 +1,6 @@
+import asyncio
 import sys
+import time
 from pathlib import Path
 
 import pytest
@@ -15,6 +17,19 @@
 )
 
 
+async def wait_for_notifications(
+    notifier: DesktopNotifier, notification_count: int = 1, timeout_sec: float = 2.0
+) -> None:
+    t0 = time.monotonic()
+
+    while time.monotonic() - t0 < timeout_sec:
+        if len(await notifier.get_current_notifications()) == notification_count:
+            return
+        await asyncio.sleep(0.2)
+
+    raise TimeoutError("Timed out while waiting for notifications")
+
+
 @pytest.mark.asyncio
 async def test_request_authorisation(notifier: DesktopNotifier) -> None:
     """
@@ -48,6 +63,7 @@ async def test_send(notifier: DesktopNotifier) -> None:
         thread="test_notifications",
         timeout=5,
     )
+    await wait_for_notifications(notifier)
     assert notification in await notifier.get_current_notifications()
 
 
@@ -125,7 +141,9 @@ async def test_clear(notifier: DesktopNotifier) -> None:
         title="Julius Caesar",
         message="Et tu, Brute?",
     )
+    await wait_for_notifications(notifier, 2)
     current_notifications = await notifier.get_current_notifications()
+
     assert n0 in current_notifications
     assert n1 in current_notifications
 
@@ -143,8 +161,9 @@ async def test_clear_all(notifier: DesktopNotifier) -> None:
         title="Julius Caesar",
         message="Et tu, Brute?",
     )
-
+    await wait_for_notifications(notifier, 2)
     current_notifications = await notifier.get_current_notifications()
+
     assert n0 in current_notifications
     assert n1 in current_notifications
 

tests/test_sync_api.py

@@ -1,4 +1,5 @@
 import sys
+import time
 
 import pytest
 
@@ -11,6 +12,19 @@
 )
 
 
+def wait_for_notifications(
+    notifier: DesktopNotifierSync, notification_count: int = 1, timeout_sec: float = 2.0
+) -> None:
+    t0 = time.monotonic()
+
+    while time.monotonic() - t0 < timeout_sec:
+        if len(notifier.get_current_notifications()) == notification_count:
+            return
+        time.sleep(0.2)
+
+    raise TimeoutError("Timed out while waiting for notifications")
+
+
 def test_send(notifier_sync: DesktopNotifierSync) -> None:
     notification = notifier_sync.send(
         title="Julius Caesar",
@@ -33,6 +47,7 @@ def test_send(notifier_sync: DesktopNotifierSync) -> None:
         thread="test_notifications",
         timeout=5,
     )
+    wait_for_notifications(notifier_sync)
     assert notification in notifier_sync.get_current_notifications()
 
 
@@ -49,7 +64,9 @@ def test_clear(notifier_sync: DesktopNotifierSync) -> None:
         title="Julius Caesar",
         message="Et tu, Brute?",
     )
+    wait_for_notifications(notifier_sync, 2)
     current_notifications = notifier_sync.get_current_notifications()
+
     assert n0 in current_notifications
     assert n1 in current_notifications
 
@@ -66,8 +83,9 @@ def test_clear_all(notifier_sync: DesktopNotifierSync) -> None:
         title="Julius Caesar",
         message="Et tu, Brute?",
     )
-
+    wait_for_notifications(notifier_sync, 2)
     current_notifications = notifier_sync.get_current_notifications()
+
     assert n0 in current_notifications
     assert n1 in current_notifications