diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index ebb46540..284d2173 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -25,7 +25,7 @@ jobs: flake8 src tests examples - name: isort run: | - isort src tests examples + isort --check --diff src tests examples mypy: runs-on: ${{ matrix.os }} diff --git a/.gitignore b/.gitignore index 18e17bb3..8d06390f 100644 --- a/.gitignore +++ b/.gitignore @@ -51,6 +51,10 @@ venv/ .installed.cfg *.egg +# pytest / coverage +.coverage +coverage.xml + # IntelliJ Idea family of suites .idea *.iml diff --git a/CHANGES.md b/CHANGES.md index 6f3c794e..6889b236 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,3 +1,34 @@ +# v7.0.0 + +## Added: + +* New `DispatchedNotification` class with the platform native notification identifier + and runtime-updated status information about that notification +* New cross-platform implementation for `timeout` +* New `on_dispatched` event that is triggered when a notification was sent to the + notifications server. +* New `on_cleared` event that is triggered when a notification is closed without + user interaction (e.g. because it expired). +* Add class-level callbacks to `DesktopNotifierSync` as well + +## Changed: + +* `send` methods now return `DispatchedNotification` instances. +* `send` methods now also accept `DispatchedNotification` instances, allowing one to + replace existing notifications with updated information; to do so simply create a + new `DispatchedNotification` instance with the same `identifier`, but a different + `Notification` instance +* `Notification` now accepts floats as `timeout` +* `Notification` and `Button` instances can now be initialized with `identifier=None` +* Interaction callbacks at the `DesktopNotifier` level now truly receive all events + the notifications server signals; depending on the platform this might include + interactions with notifications of other applications as well + +## Fixed: + +* Fixed sending notifications with a named sound resource on Linux + (given that the notifications server has that capability) + # v6.0.0 ## Added: diff --git a/README.md b/README.md index fdf7a80b..1329488e 100644 --- a/README.md +++ b/README.md @@ -73,10 +73,7 @@ from desktop_notifier import DesktopNotifier, Urgency, Button, ReplyField, DEFAU async def main() -> None: - notifier = DesktopNotifier( - app_name="Sample App", - notification_limit=10, - ) + notifier = DesktopNotifier(app_name="Sample App") await notifier.send( title="Julius Caesar", @@ -91,9 +88,12 @@ async def main() -> None: reply_field=ReplyField( on_replied=lambda text: print("Brutus replied:", text), ), + on_dispatched=lambda: print("Notification showing"), + on_cleared=lambda: print("Notification timed out"), on_clicked=lambda: print("Notification clicked"), on_dismissed=lambda: print("Notification dismissed"), sound=DEFAULT_SOUND, + timeout=10, ) # Run the event loop forever to respond to user interactions with the notification. diff --git a/docs/background/platform_support.rst b/docs/background/platform_support.rst index 50466a16..b4a3d609 100644 --- a/docs/background/platform_support.rst +++ b/docs/background/platform_support.rst @@ -29,7 +29,7 @@ Please refer to the platform documentation for more detailed information: "sound", "Play a sound with the notification", "✓ [#f2]_", "✓ [#f5]_", "✓" "thread", "An identifier to group notifications together", "--", "✓", "✓" "attachment", "File attachment, e.g., an image", "✓ [#f2]_ [#f6]_", "✓ [#f6]_", "✓ [#f6]_" - "timeout", "Duration in seconds until notification auto-dismissal", "✓", "--", "--" + "timeout", "Duration in seconds until notification auto-dismissal", "✓", "✓", "✓" .. [#f1] App name and icon on macOS and Windows are automatically determined by the calling application. diff --git a/examples/eventloop.py b/examples/eventloop.py index 1e7cbfc3..c6747670 100644 --- a/examples/eventloop.py +++ b/examples/eventloop.py @@ -27,9 +27,12 @@ async def main() -> None: button_title="Send", on_replied=lambda text: print("Brutus replied:", text), ), + on_dispatched=lambda: print("Notification showing"), + on_cleared=lambda: print("Notification timed out"), on_clicked=lambda: print("Notification clicked"), on_dismissed=lambda: print("Notification dismissed"), sound=DEFAULT_SOUND, + timeout=10, ) # Run the event loop forever to respond to user interactions with the notification. diff --git a/examples/eventloop_handlers.py b/examples/eventloop_handlers.py index 87ffe300..65eee911 100644 --- a/examples/eventloop_handlers.py +++ b/examples/eventloop_handlers.py @@ -11,6 +11,14 @@ asyncio.set_event_loop_policy(EventLoopPolicy()) +def on_dispatched(identifier: str) -> None: + print(f"Notification '{identifier}' is showing now") + + +def on_cleared(identifier: str) -> None: + print(f"Notification '{identifier}' was cleared without user interaction") + + def on_clicked(identifier: str) -> None: print(f"Notification '{identifier}' was clicked") @@ -29,6 +37,8 @@ def on_replied(identifier: str, reply: str) -> None: async def main() -> None: notifier = DesktopNotifier(app_name="Sample App") + notifier.on_dispatched = on_dispatched + notifier.on_cleared = on_cleared notifier.on_clicked = on_clicked notifier.on_dismissed = on_dismissed notifier.on_button_pressed = on_button_pressed @@ -47,6 +57,7 @@ async def main() -> None: button_title="Send", ), sound=DEFAULT_SOUND, + timeout=10, ) # Run the event loop forever to respond to user interactions with the notification. diff --git a/examples/synchronous.py b/examples/synchronous.py index d840a598..37b2bfab 100644 --- a/examples/synchronous.py +++ b/examples/synchronous.py @@ -23,7 +23,10 @@ button_title="Send", on_replied=lambda text: print("Brutus replied:", text), ), + on_dispatched=lambda: print("Notification showing"), + on_cleared=lambda: print("Notification timed out"), on_clicked=lambda: print("Notification clicked"), on_dismissed=lambda: print("Notification dismissed"), sound=DEFAULT_SOUND, + timeout=10, ) diff --git a/pyproject.toml b/pyproject.toml index 61d0e28e..86c3016f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,8 +39,8 @@ classifiers = [ ] requires-python = ">=3.9" dependencies = [ - "bidict", "packaging", + "immutabledict", "dbus-fast;sys_platform=='linux'", "rubicon-objc;sys_platform=='darwin'", "winrt-windows.applicationmodel.core;sys_platform=='win32'", @@ -81,9 +81,7 @@ docs = [ desktop_notifier = ["**/*.png"] [tool.flake8] -ignore = "E203,E501,W503,H306" -per-file-ignores = """ -__init__.py: F401""" +ignore = "E501,W503" statistics = "True" [tool.mypy] diff --git a/src/desktop_notifier/__init__.py b/src/desktop_notifier/__init__.py index 6b59e201..beb7fb29 100644 --- a/src/desktop_notifier/__init__.py +++ b/src/desktop_notifier/__init__.py @@ -9,6 +9,7 @@ Button, Capability, DesktopNotifier, + DispatchedNotification, Icon, Notification, ReplyField, @@ -37,4 +38,5 @@ "Capability", "DEFAULT_SOUND", "DEFAULT_ICON", + "DispatchedNotification", ] diff --git a/src/desktop_notifier/backends/base.py b/src/desktop_notifier/backends/base.py index ae436ac7..1dec3e38 100644 --- a/src/desktop_notifier/backends/base.py +++ b/src/desktop_notifier/backends/base.py @@ -4,11 +4,13 @@ """ from __future__ import annotations +import asyncio import logging from abc import ABC, abstractmethod +from asyncio import Task from typing import Any, Callable -from ..common import Capability, Notification +from ..common import Capability, DispatchedNotification, Icon, Notification, uuid_str __all__ = [ "DesktopNotifierBackend", @@ -24,10 +26,14 @@ class DesktopNotifierBackend(ABC): :param app_name: Name to identify the application in the notification center. """ - def __init__(self, app_name: str) -> None: + def __init__(self, app_name: str, app_icon: Icon | None = None) -> None: self.app_name = app_name - self._notification_cache: dict[str, Notification] = dict() + self.app_icon = app_icon + self._notification_cache: dict[str, DispatchedNotification] = dict() + self._timeout_tasks: dict[str, Task[Any]] = dict() + self.on_dispatched: Callable[[str], Any] | None = None + self.on_cleared: Callable[[str], Any] | None = None self.on_clicked: Callable[[str], Any] | None = None self.on_dismissed: Callable[[str], Any] | None = None self.on_button_pressed: Callable[[str, str], Any] | None = None @@ -49,33 +55,81 @@ async def has_authorisation(self) -> bool: """ ... - async def send(self, notification: Notification) -> None: + async def send( + self, notification: Notification | DispatchedNotification + ) -> DispatchedNotification | None: """ Sends a desktop notification. :param notification: Notification to send. """ + dispatched_notification: DispatchedNotification | None = None + if isinstance(notification, DispatchedNotification): + dispatched_notification = notification + notification = dispatched_notification.notification + try: - await self._send(notification) + identifier = await self._send(notification, dispatched_notification) except Exception: # Notifications can fail for many reasons: # The dbus service may not be available, we might be in a headless session, # etc. Since notifications are not critical to an application, we only emit # a warning. - logger.warning("Notification failed", exc_info=True) + logger.warning( + "Failed sending notification: %s", notification, exc_info=True + ) + return None else: logger.debug("Notification sent: %s", notification) - self._notification_cache[notification.identifier] = notification + if not identifier: + if dispatched_notification: + identifier = dispatched_notification.identifier + elif notification.identifier not in self._notification_cache: + identifier = notification.identifier + else: + identifier = uuid_str() + + dispatched_notification = DispatchedNotification(identifier, notification) + self._notification_cache[identifier] = dispatched_notification + + self.handle_dispatched(identifier) + + if notification.timeout > 0: + timeout_task = self._timeout_task(identifier, notification.timeout) + self._timeout_tasks[identifier] = asyncio.create_task(timeout_task) + + return dispatched_notification + + async def _timeout_task(self, identifier: str, timeout: float) -> None: + """ + Waits until a notification's timeout delay is reached and clears the notification. + This asyncio task might be cancelled by :meth:`_clear_notification_from_cache`. + """ + await asyncio.sleep(timeout) + + # manually call handle_cleared() so that the event isn't misidentified as dismissal + self.handle_cleared(identifier) + + await self._clear(identifier) - def _clear_notification_from_cache(self, identifier: str) -> Notification | None: + def _clear_notification_from_cache( + self, identifier: str + ) -> DispatchedNotification | None: """ Removes the notification from our cache. Should be called by backends when the notification is closed. """ + if identifier in self._timeout_tasks: + self._timeout_tasks[identifier].cancel() + del self._timeout_tasks[identifier] return self._notification_cache.pop(identifier, None) @abstractmethod - async def _send(self, notification: Notification) -> None: + async def _send( + self, + notification: Notification, + replace_notification: DispatchedNotification | None = None, + ) -> str | None: """ Method to send a notification via the platform. This should be implemented by subclasses. @@ -93,6 +147,10 @@ async def get_current_notifications(self) -> list[str]: """Returns identifiers of all currently displayed notifications for this app.""" return list(self._notification_cache.keys()) + def get_cached_notifications(self) -> dict[str, DispatchedNotification]: + """Returns the notifications known to this Desktop Notifier instance.""" + return self._notification_cache.copy() + async def clear(self, identifier: str) -> None: """ Removes the given notification from the notification center. This is a wrapper @@ -103,7 +161,6 @@ async def clear(self, identifier: str) -> None: :param identifier: Notification identifier. """ await self._clear(identifier) - self._clear_notification_from_cache(identifier) @abstractmethod async def _clear(self, identifier: str) -> None: @@ -122,9 +179,7 @@ async def clear_all(self) -> None: :meth:`_clear_all` to actually clear the notifications. Platform implementations must implement :meth:`_clear_all`. """ - await self._clear_all() - self._notification_cache.clear() @abstractmethod async def _clear_all(self) -> None: @@ -142,46 +197,91 @@ async def get_capabilities(self) -> frozenset[Capability]: """ ... - def handle_clicked( - self, identifier: str, notification: Notification | None = None - ) -> None: - if notification and notification.on_clicked: - notification.on_clicked() - elif self.on_clicked: + def handle_dispatched(self, identifier: str) -> None: + if identifier in self._notification_cache: + dispatched_notification = self._notification_cache[identifier] + if dispatched_notification.notification.on_dispatched: + dispatched_notification.notification.on_dispatched() + return + if self.on_dispatched: + self.on_dispatched(identifier) + + def handle_cleared(self, identifier: str) -> None: + dispatched_notification = self._clear_notification_from_cache(identifier) + if dispatched_notification: + if dispatched_notification.cleared: + return + + object.__setattr__(dispatched_notification, "cleared", True) + + if dispatched_notification.notification.on_cleared: + dispatched_notification.notification.on_cleared() + return + if self.on_cleared: + self.on_cleared(identifier) + + def handle_clicked(self, identifier: str) -> None: + dispatched_notification = self._clear_notification_from_cache(identifier) + if dispatched_notification: + if dispatched_notification.cleared: + return + + object.__setattr__(dispatched_notification, "cleared", True) + object.__setattr__(dispatched_notification, "clicked", True) + + if dispatched_notification.notification.on_clicked: + dispatched_notification.notification.on_clicked() + return + if self.on_clicked: self.on_clicked(identifier) - def handle_dismissed( - self, identifier: str, notification: Notification | None = None - ) -> None: - if notification and notification.on_dismissed: - notification.on_dismissed() - elif self.on_dismissed: + def handle_dismissed(self, identifier: str) -> None: + dispatched_notification = self._clear_notification_from_cache(identifier) + if dispatched_notification: + if dispatched_notification.cleared: + return + + object.__setattr__(dispatched_notification, "cleared", True) + object.__setattr__(dispatched_notification, "dismissed", True) + + if dispatched_notification.notification.on_dismissed: + dispatched_notification.notification.on_dismissed() + return + if self.on_dismissed: self.on_dismissed(identifier) - def handle_replied( - self, identifier: str, reply_text: str, notification: Notification | None = None - ) -> None: - if ( - notification - and notification.reply_field - and notification.reply_field.on_replied - ): - notification.reply_field.on_replied(reply_text) - elif self.on_replied: - self.on_replied(identifier, reply_text) + def handle_replied(self, identifier: str, reply_text: str) -> None: + dispatched_notification = self._clear_notification_from_cache(identifier) + if dispatched_notification: + if dispatched_notification.cleared: + return - def handle_button( - self, - identifier: str, - button_identifier: str, - notification: Notification | None = None, - ) -> None: - if notification and button_identifier in notification._buttons_dict: - button = notification._buttons_dict[button_identifier] - else: - button = None + object.__setattr__(dispatched_notification, "cleared", True) + object.__setattr__(dispatched_notification, "replied", reply_text) + + reply_field = dispatched_notification.notification.reply_field + if reply_field and reply_field.on_replied: + reply_field.on_replied(reply_text) + return + if self.on_replied: + self.on_replied(identifier, reply_text) - if button and button.on_pressed: - button.on_pressed() - elif self.on_button_pressed: + def handle_button(self, identifier: str, button_identifier: str) -> None: + dispatched_notification = self._clear_notification_from_cache(identifier) + if dispatched_notification: + if dispatched_notification.cleared: + return + + object.__setattr__(dispatched_notification, "cleared", True) + object.__setattr__( + dispatched_notification, "button_clicked", button_identifier + ) + + buttons = dispatched_notification.notification.buttons_dict + if button_identifier in buttons: + button = buttons[button_identifier] + if button and button.on_pressed: + button.on_pressed() + return + if self.on_button_pressed: self.on_button_pressed(identifier, button_identifier) diff --git a/src/desktop_notifier/backends/dbus.py b/src/desktop_notifier/backends/dbus.py index 916ab2f3..3fbefd74 100644 --- a/src/desktop_notifier/backends/dbus.py +++ b/src/desktop_notifier/backends/dbus.py @@ -10,13 +10,12 @@ import logging from typing import TypeVar -from bidict import bidict from dbus_fast.aio.message_bus import MessageBus from dbus_fast.aio.proxy_object import ProxyInterface from dbus_fast.errors import DBusError from dbus_fast.signature import Variant -from ..common import Capability, Notification, Urgency +from ..common import Capability, DispatchedNotification, Icon, Notification, Urgency from .base import DesktopNotifierBackend __all__ = ["DBusDesktopNotifier"] @@ -48,10 +47,9 @@ class DBusDesktopNotifier(DesktopNotifierBackend): supported_hint_signatures = {"a{sv}", "a{ss}"} - def __init__(self, app_name: str) -> None: - super().__init__(app_name) + def __init__(self, app_name: str, app_icon: Icon | None = None) -> None: + super().__init__(app_name, app_icon) self.interface: ProxyInterface | None = None - self._platform_to_interface_notification_identifier: bidict[int, str] = bidict() async def request_authorisation(self) -> bool: """ @@ -90,7 +88,11 @@ async def _init_dbus(self) -> ProxyInterface: return self.interface - async def _send(self, notification: Notification) -> None: + async def _send( + self, + notification: Notification, + replace_notification: DispatchedNotification | None = None, + ) -> str | None: """ Asynchronously sends a notification via the Dbus interface. @@ -99,6 +101,11 @@ async def _send(self, notification: Notification) -> None: if not self.interface: self.interface = await self._init_dbus() + platform_id: int = 0 + if replace_notification: + # FreeDesktop.org notifications can be replaced seamlessly + platform_id = int(replace_notification.identifier) + # The "default" action is typically invoked when clicking on the # notification body itself, see # https://specifications.freedesktop.org/notification-spec. There are some @@ -115,7 +122,7 @@ async def _send(self, notification: Notification) -> None: if notification.sound: if notification.sound.is_named(): - hints_v["sound-name"] = Variant("s", "message-new-instant") + hints_v["sound-name"] = Variant("s", notification.sound.name) else: hints_v["sound-file"] = Variant("s", notification.sound.as_uri()) @@ -131,7 +138,7 @@ async def _send(self, notification: Notification) -> None: if hints_signature == "": logger.warning("Notification server not supported") - return + return None hints: dict[str, str] | dict[str, Variant] @@ -142,20 +149,28 @@ async def _send(self, notification: Notification) -> None: else: hints = {} - timeout = notification.timeout * 1000 if notification.timeout != -1 else -1 + # few notifications servers implement timeouts, thus we use our own implementation + timeout = -1 + + icon: str = "" if notification.icon: - if notification.icon.is_named(): - icon = notification.icon.name - else: - icon = notification.icon.as_uri() - else: - icon = "" + icon = ( + notification.icon.as_name() + if notification.icon.is_named() + else notification.icon.as_uri() + ) + elif self.app_icon: + icon = ( + self.app_icon.as_name() + if self.app_icon.is_named() + else self.app_icon.as_uri() + ) # dbus_next proxy APIs are generated at runtime. Silence the type checker but # raise an AttributeError if required. platform_id = await self.interface.call_notify( # type:ignore[attr-defined] self.app_name, - 0, + platform_id, icon, notification.title, notification.message, @@ -163,20 +178,17 @@ async def _send(self, notification: Notification) -> None: hints, timeout, ) - self._platform_to_interface_notification_identifier[platform_id] = ( - notification.identifier - ) + + return str(platform_id) async def _clear(self, identifier: str) -> None: """ Asynchronously removes a notification from the notification center """ if not self.interface: - return + self.interface = await self._init_dbus() - platform_id = self._platform_to_interface_notification_identifier.inverse[ - identifier - ] + platform_id = int(identifier) try: # dbus_next proxy APIs are generated at runtime. Silence the type checker @@ -189,22 +201,11 @@ async def _clear(self, identifier: str) -> None: # See https://specifications.freedesktop.org/notification-spec/latest/protocol.html#command-close-notification pass - try: - del self._platform_to_interface_notification_identifier.inverse[identifier] - except KeyError: - # Popping may have been handled already by _on_close callback. - pass - async def _clear_all(self) -> None: """ Asynchronously clears all notifications from notification center """ - if not self.interface: - return - - for identifier in list( - self._platform_to_interface_notification_identifier.values() - ): + for identifier in list(self._notification_cache.keys()): await self._clear(identifier) # Note that _on_action and _on_closed might be called for the same notification @@ -221,17 +222,11 @@ def _on_action(self, nid: int, action_key: str) -> None: :param action_key: A string identifying the action to take. We choose those keys ourselves when scheduling the notification. """ - identifier = self._platform_to_interface_notification_identifier.pop(nid, "") - notification = self._clear_notification_from_cache(identifier) - - if not notification: - return - + identifier = str(nid) if action_key == "default": - self.handle_clicked(identifier, notification) - return - - self.handle_button(identifier, action_key, notification) + self.handle_clicked(identifier) + else: + self.handle_button(identifier, action_key) def _on_closed(self, nid: int, reason: int) -> None: """ @@ -241,14 +236,11 @@ def _on_closed(self, nid: int, reason: int) -> None: :param nid: The platform's notification ID as an integer. :param reason: An integer describing the reason why the notification was closed. """ - identifier = self._platform_to_interface_notification_identifier.pop(nid, "") - notification = self._clear_notification_from_cache(identifier) - - if not notification: - return - + identifier = str(nid) if reason == NOTIFICATION_CLOSED_DISMISSED: - self.handle_dismissed(identifier, notification) + self.handle_dismissed(identifier) + else: + self.handle_cleared(identifier) async def get_capabilities(self) -> frozenset[Capability]: if not self.interface: @@ -258,8 +250,10 @@ async def get_capabilities(self) -> frozenset[Capability]: Capability.APP_NAME, Capability.ICON, Capability.TITLE, - Capability.TIMEOUT, Capability.URGENCY, + Capability.ON_DISPATCHED, + Capability.ON_CLEARED, + Capability.TIMEOUT, } # Capabilities supported by some notification servers. diff --git a/src/desktop_notifier/backends/dummy.py b/src/desktop_notifier/backends/dummy.py index 5b7a10a8..fd9abb22 100644 --- a/src/desktop_notifier/backends/dummy.py +++ b/src/desktop_notifier/backends/dummy.py @@ -4,16 +4,13 @@ """ from __future__ import annotations -from ..common import Capability, Notification +from ..common import Capability, DispatchedNotification, Notification from .base import DesktopNotifierBackend class DummyNotificationCenter(DesktopNotifierBackend): """A dummy backend for unsupported platforms""" - def __init__(self, app_name: str) -> None: - super().__init__(app_name) - async def request_authorisation(self) -> bool: """ Request authorisation to send notifications. @@ -28,7 +25,11 @@ async def has_authorisation(self) -> bool: """ return True - async def _send(self, notification: Notification) -> None: + async def _send( + self, + notification: Notification, + replace_notification: DispatchedNotification | None = None, + ) -> str | None: pass async def _clear(self, identifier: str) -> None: diff --git a/src/desktop_notifier/backends/macos.py b/src/desktop_notifier/backends/macos.py index ae13bfec..fb6757c6 100644 --- a/src/desktop_notifier/backends/macos.py +++ b/src/desktop_notifier/backends/macos.py @@ -22,7 +22,15 @@ from rubicon.objc import NSObject, ObjCClass, objc_method, py_from_ns from rubicon.objc.runtime import load_library, objc_block, objc_id -from ..common import DEFAULT_SOUND, Capability, Notification, Urgency +from ..common import ( + DEFAULT_SOUND, + Capability, + DispatchedNotification, + Icon, + Notification, + Urgency, + uuid_str, +) from .base import DesktopNotifierBackend from .macos_support import macos_version @@ -94,21 +102,17 @@ def userNotificationCenter_didReceiveNotificationResponse_withCompletionHandler_ self, center, response, completion_handler: objc_block ) -> None: identifier = py_from_ns(response.notification.request.identifier) - notification = self.implementation._clear_notification_from_cache(identifier) if response.actionIdentifier == UNNotificationDefaultActionIdentifier: - self.implementation.handle_clicked(identifier, notification) - + self.implementation.handle_clicked(identifier) elif response.actionIdentifier == UNNotificationDismissActionIdentifier: - self.implementation.handle_dismissed(identifier, notification) - + self.implementation.handle_dismissed(identifier) elif response.actionIdentifier == ReplyActionIdentifier: reply_text = py_from_ns(response.userText) - self.implementation.handle_replied(identifier, reply_text, notification) - + self.implementation.handle_replied(identifier, reply_text) else: action_id = py_from_ns(response.actionIdentifier) - self.implementation.handle_button(identifier, action_id, notification) + self.implementation.handle_button(identifier, action_id) completion_handler() @@ -129,8 +133,8 @@ class CocoaNotificationCenter(DesktopNotifierBackend): Urgency.Critical: UNNotificationInterruptionLevel.TimeSensitive, } - def __init__(self, app_name: str) -> None: - super().__init__(app_name) + def __init__(self, app_name: str, app_icon: Icon | None = None) -> None: + super().__init__(app_name, app_icon) self.nc = UNUserNotificationCenter.currentNotificationCenter() self.nc_delegate = NotificationCenterDelegate.alloc().init() self.nc_delegate.implementation = self @@ -218,12 +222,24 @@ def handler(notifications: objc_id) -> None: return identifiers - async def _send(self, notification: Notification) -> None: + async def _send( + self, + notification: Notification, + replace_notification: DispatchedNotification | None = None, + ) -> str | None: """ Uses UNUserNotificationCenter to schedule a notification. :param notification: Notification to send. """ + identifier = notification.identifier + if replace_notification: + await self._clear(replace_notification.identifier) + identifier = replace_notification.identifier + elif identifier in self._notification_cache: + # identifier is already in use, generate a new random identifier + identifier = uuid_str() + # On macOS, we need to register a new notification category for every # unique set of buttons. category_id = await self._find_or_create_notification_category(notification) @@ -264,7 +280,7 @@ async def _send(self, notification: Notification) -> None: content.attachments = [attachment] notification_request = UNNotificationRequest.requestWithIdentifier( - notification.identifier, content=content, trigger=None + identifier, content=content, trigger=None ) future: Future[NSError] = Future() # type:ignore[valid-type] @@ -287,6 +303,8 @@ def handler(error: objc_id) -> None: log_nserror(error, "Error when scheduling notification") error.autorelease() # type:ignore[attr-defined] + return identifier + async def _find_or_create_notification_category( self, notification: Notification ) -> str: @@ -398,12 +416,14 @@ async def get_capabilities(self) -> frozenset[Capability]: Capability.MESSAGE, Capability.BUTTONS, Capability.REPLY_FIELD, + Capability.ON_DISPATCHED, Capability.ON_CLICKED, Capability.ON_DISMISSED, Capability.SOUND, Capability.SOUND_NAME, Capability.THREAD, Capability.ATTACHMENT, + Capability.TIMEOUT, } if macos_version >= Version("12.0"): capabilities.add(Capability.URGENCY) diff --git a/src/desktop_notifier/backends/winrt.py b/src/desktop_notifier/backends/winrt.py index 1ddbfcad..c54be703 100644 --- a/src/desktop_notifier/backends/winrt.py +++ b/src/desktop_notifier/backends/winrt.py @@ -31,7 +31,15 @@ ) # local imports -from ..common import DEFAULT_SOUND, Capability, Notification, Urgency +from ..common import ( + DEFAULT_SOUND, + Capability, + DispatchedNotification, + Icon, + Notification, + Urgency, + uuid_str, +) from .base import DesktopNotifierBackend __all__ = ["WinRTDesktopNotifier"] @@ -69,11 +77,8 @@ class WinRTDesktopNotifier(DesktopNotifierBackend): Urgency.Critical: ToastNotificationPriority.HIGH, } - def __init__( - self, - app_name: str, - ) -> None: - super().__init__(app_name) + def __init__(self, app_name: str, app_icon: Icon | None = None) -> None: + super().__init__(app_name, app_icon) manager = ToastNotificationManager.get_default() @@ -115,12 +120,24 @@ async def has_authorisation(self) -> bool: # See https://github.com/samschott/desktop-notifier/issues/95. return True - async def _send(self, notification: Notification) -> None: + async def _send( + self, + notification: Notification, + replace_notification: DispatchedNotification | None = None, + ) -> str | None: """ Asynchronously sends a notification. :param notification: Notification to send. """ + identifier = notification.identifier + if replace_notification: + await self._clear(replace_notification.identifier) + identifier = replace_notification.identifier + elif identifier in self._notification_cache: + # identifier is already in use, generate a new random identifier + identifier = uuid_str() + toast_xml = Element("toast", {"launch": DEFAULT_ACTION}) visual_xml = SubElement(toast_xml, "visual") actions_xml = SubElement(toast_xml, "actions") @@ -145,13 +162,18 @@ async def _send(self, notification: Notification) -> None: message_xml = SubElement(binding, "text") message_xml.text = notification.message - if notification.icon and notification.icon.is_file(): + icon_obj: Icon | None = None + if notification.icon: + icon_obj = notification.icon + elif self.app_icon: + icon_obj = self.app_icon + if icon_obj and icon_obj.is_file(): SubElement( binding, "image", { "placement": "appLogoOverride", - "src": notification.icon.as_uri(), + "src": icon_obj.as_uri(), }, ) @@ -210,7 +232,7 @@ async def _send(self, notification: Notification) -> None: xml_document.load_xml(tostring(toast_xml, encoding="unicode")) native = ToastNotification(xml_document) - native.tag = notification.identifier + native.tag = identifier native.priority = self._to_native_urgency[notification.urgency] native.add_activated(self._on_activated) @@ -219,14 +241,14 @@ async def _send(self, notification: Notification) -> None: self.notifier.show(native) + return identifier + def _on_activated( self, sender: ToastNotification | None, boxed_activated_args: WinRTObject | None ) -> None: if not sender: return - notification = self._clear_notification_from_cache(sender.tag) - if not boxed_activated_args: return @@ -234,16 +256,14 @@ def _on_activated( action_id = activated_args.arguments if action_id == DEFAULT_ACTION: - self.handle_clicked(sender.tag, notification) - + self.handle_clicked(sender.tag) elif action_id == REPLY_ACTION and activated_args.user_input: boxed_reply = activated_args.user_input[REPLY_TEXTBOX_NAME] reply = unbox(boxed_reply) - self.handle_replied(sender.tag, reply, notification) - + self.handle_replied(sender.tag, reply) elif action_id.startswith(BUTTON_ACTION_PREFIX): button_id = action_id.replace(BUTTON_ACTION_PREFIX, "") - self.handle_button(sender.tag, button_id, notification) + self.handle_button(sender.tag, button_id) def _on_dismissed( self, @@ -253,13 +273,11 @@ def _on_dismissed( if not sender: return - notification = self._clear_notification_from_cache(sender.tag) - - if ( - dismissed_args - and dismissed_args.reason == ToastDismissalReason.USER_CANCELED - ): - self.handle_dismissed(sender.tag, notification) + if dismissed_args: + # ToastDismissalReason.APPLICATION_HIDDEN and ToastDismissalReason.TIMED_OUT + # both just indicate that the toast was sent to the notifications center + if dismissed_args.reason == ToastDismissalReason.USER_CANCELED: + self.handle_dismissed(sender.tag) def _on_failed( self, sender: ToastNotification | None, failed_args: ToastFailedEventArgs | None @@ -298,12 +316,14 @@ async def get_capabilities(self) -> frozenset[Capability]: Capability.ICON, Capability.BUTTONS, Capability.REPLY_FIELD, + Capability.ON_DISPATCHED, Capability.ON_CLICKED, Capability.ON_DISMISSED, Capability.THREAD, Capability.ATTACHMENT, Capability.SOUND, Capability.SOUND_NAME, + Capability.TIMEOUT, } # Custom audio is support only starting with the Windows 10 Anniversary update. # See https://learn.microsoft.com/en-us/windows/apps/design/shell/tiles-and-notifications/custom-audio-on-toasts#add-the-custom-audio. diff --git a/src/desktop_notifier/common.py b/src/desktop_notifier/common.py index 23698cc3..b76a1d34 100644 --- a/src/desktop_notifier/common.py +++ b/src/desktop_notifier/common.py @@ -14,6 +14,8 @@ from typing import Any, Callable from urllib.parse import unquote, urlparse +from immutabledict import immutabledict + __all__ = [ "Capability", "FileResource", @@ -28,6 +30,8 @@ "Notification", "DEFAULT_ICON", "DEFAULT_SOUND", + "DispatchedNotification", + "uuid_str", ] @@ -100,6 +104,11 @@ class Resource(FileResource): name: str | None = None """Name of the system resource""" + def as_name(self) -> str: + if self.name is not None: + return self.name + raise AttributeError("No resource name provided") + def is_named(self) -> bool: """Returns whether the instance was initialized with ``name``""" return self.name is not None @@ -170,6 +179,10 @@ class Button: identifier: str = dataclasses.field(default_factory=uuid_str) """A unique identifier to use in callbacks to specify with button was clicked""" + def __post_init__(self) -> None: + if self.identifier is None: + object.__setattr__(self, "identifier", uuid_str()) + @dataclass(frozen=True) class ReplyField: @@ -207,55 +220,94 @@ class Notification: message: str """Notification message""" - urgency: Urgency = Urgency.Normal + urgency: Urgency = field(default=Urgency.Normal, repr=False) """Notification urgency. Can determine stickiness, notification appearance and break through silencing.""" - icon: Icon | None = None + icon: Icon | None = field(default=None, repr=False) """Icon to use for the notification""" - buttons: tuple[Button, ...] = field(default_factory=tuple) + buttons: tuple[Button, ...] = field(default_factory=tuple, repr=False) """Buttons shown on an interactive notification""" - reply_field: ReplyField | None = None + buttons_dict: immutabledict[str, Button] = field( + default_factory=immutabledict, init=False, repr=False, compare=False + ) + """Buttons shown on an interactive notification, indexed by button identifier""" + + reply_field: ReplyField | None = field(default=None, repr=False) """Text field shown on an interactive notification. This can be used for example for messaging apps to reply directly from the notification.""" - on_clicked: Callable[[], Any] | None = None + on_dispatched: Callable[[], Any] | None = field(default=None, repr=False) + """Method to call when the notification was sent to the notifications server for display""" + + on_cleared: Callable[[], Any] | None = field(default=None, repr=False) + """Method to call when the notification is cleared without user interaction""" + + on_clicked: Callable[[], Any] | None = field(default=None, repr=False) """Method to call when the notification is clicked""" - on_dismissed: Callable[[], Any] | None = None + on_dismissed: Callable[[], Any] | None = field(default=None, repr=False) """Method to call when the notification is dismissed""" - attachment: Attachment | None = None + attachment: Attachment | None = field(default=None, repr=False) """A file attached to the notification which may be displayed as a preview""" - sound: Sound | None = None + sound: Sound | None = field(default=None, repr=False) """A sound to play on notification""" - thread: str | None = None + thread: str | None = field(default=None, repr=False) """An identifier to group related notifications together, e.g., from a chat space""" - timeout: int = -1 + timeout: float = field(default=-1, repr=False) """Duration in seconds for which the notification is shown""" identifier: str = field(default_factory=uuid_str) """A unique identifier for this notification. Generated automatically if not passed by the client.""" - _buttons_dict: dict[str, Button] = field(default_factory=dict) - def __post_init__(self) -> None: - for button in self.buttons: - self._buttons_dict[button.identifier] = button + if self.identifier is None: + object.__setattr__(self, "identifier", uuid_str()) - def __repr__(self) -> str: - return ( - f"<{self.__class__.__name__}(identifier='{self.identifier}', " - f"title='{self.title}', message='{self.message}')>" + object.__setattr__( + self, + "buttons_dict", + immutabledict({button.identifier: button for button in self.buttons}), ) +@dataclass(frozen=True) +class DispatchedNotification: + """A desktop notification that was sent to the notifications server earlier""" + + identifier: str + """A platform-dependant unique identifier for this dispatched notification. + The format varies from platform to platform. The identifier is unique among + the currently active notifications, but might be reused if a notification is + replaced. It might be identical to the randomly generated UUID of the source + notification, but you must not rely on this""" + + notification: Notification + """The notification that was sent to the notifications server""" + + cleared: bool = field(default=False, init=False, repr=False) + """Whether the notification has been cleared (both with and without user interaction, updated at runtime)""" + + clicked: bool = field(default=False, init=False, repr=False) + """Whether the user clicked on the notification (updated at runtime)""" + + dismissed: bool = field(default=False, init=False, repr=False) + """Whether the user dismissed the notification (updated at runtime)""" + + button_clicked: str | None = field(default=None, init=False, repr=False) + """The identifier of the button the user clicked on, if applicable (updated at runtime)""" + + replied: str | None = field(default=None, init=False, repr=False) + """The text the user entered into the notification's reply field, if applicable (updated at runtime)""" + + class Capability(Enum): """Notification capabilities that can be supported by a platform""" @@ -289,6 +341,13 @@ class Capability(Enum): ATTACHMENT = auto() """Supports notification attachments. Allowed file types vary by platform.""" + ON_DISPATCHED = auto() + """Supports on-dispatched callbacks""" + + ON_CLEARED = auto() + """Supports distinguishing between an user closing a notification, and clearing a + notification programmatically, and consequently supports on-cleared callbacks""" + ON_CLICKED = auto() """Supports on-clicked callbacks""" diff --git a/src/desktop_notifier/main.py b/src/desktop_notifier/main.py index 61c86999..851d320f 100644 --- a/src/desktop_notifier/main.py +++ b/src/desktop_notifier/main.py @@ -19,6 +19,7 @@ Attachment, Button, Capability, + DispatchedNotification, Icon, Notification, ReplyField, @@ -38,6 +39,7 @@ "Capability", "DEFAULT_SOUND", "DEFAULT_ICON", + "DispatchedNotification", ] logger = logging.getLogger(__name__) @@ -127,8 +129,6 @@ class DesktopNotifier: deprecated. """ - app_icon: Icon | None - def __init__( self, app_name: str = "Python", @@ -141,10 +141,8 @@ def __init__( category=DeprecationWarning, ) - self.app_icon = app_icon - backend = get_backend_class() - self._backend = backend(app_name) + self._backend = backend(app_name, app_icon) self._did_request_authorisation = False self._capabilities: frozenset[Capability] | None = None @@ -156,9 +154,17 @@ def app_name(self) -> str: @app_name.setter def app_name(self, value: str) -> None: - """Setter: app_name""" self._backend.app_name = value + @property + def app_icon(self) -> Icon | None: + """The application icon""" + return self._backend.app_icon + + @app_icon.setter + def app_icon(self, value: Icon | None) -> None: + self._backend.app_icon = value + async def request_authorisation(self) -> bool: """ Requests authorisation to send user notifications. This will be automatically @@ -178,7 +184,9 @@ async def has_authorisation(self) -> bool: """Returns whether we have authorisation to send notifications.""" return await self._backend.has_authorisation() - async def send_notification(self, notification: Notification) -> str: + async def send_notification( + self, notification: Notification | DispatchedNotification + ) -> DispatchedNotification | None: """ Sends a desktop notification. @@ -190,11 +198,7 @@ async def send_notification(self, notification: Notification) -> str: disturb" is enabled on macOS). :param notification: The notification to send. - :returns: An identifier for the scheduled notification. """ - if not notification.icon: - object.__setattr__(notification, "icon", self.app_icon) - # Ask for authorisation if not already done. On some platforms, this will # trigger a system dialog to ask the user for permission. if not self._did_request_authorisation: @@ -204,9 +208,7 @@ async def send_notification(self, notification: Notification) -> str: # We attempt to send the notification regardless of authorization. # The user may have changed settings in the meantime. - await self._backend.send(notification) - - return notification.identifier + return await self._backend.send(notification) async def send( self, @@ -216,21 +218,21 @@ async def send( icon: Icon | None = None, buttons: Sequence[Button] = (), reply_field: ReplyField | None = None, + on_dispatched: Callable[[], Any] | None = None, + on_cleared: Callable[[], Any] | None = None, on_clicked: Callable[[], Any] | None = None, on_dismissed: Callable[[], Any] | None = None, attachment: Attachment | None = None, sound: Sound | None = None, thread: str | None = None, timeout: int = -1, # in seconds - ) -> str: + ) -> DispatchedNotification | None: """ Sends a desktop notification This is a convenience function which creates a :class:`desktop_notifier.base.Notification` with the provided arguments and then calls :meth:`send_notification`. - - :returns: An identifier for the scheduled notification. """ notification = Notification( title, @@ -239,6 +241,8 @@ async def send( icon=icon, buttons=tuple(buttons), reply_field=reply_field, + on_dispatched=on_dispatched, + on_cleared=on_cleared, on_clicked=on_clicked, on_dismissed=on_dismissed, attachment=attachment, @@ -252,6 +256,10 @@ async def get_current_notifications(self) -> list[str]: """Returns identifiers of all currently displayed notifications for this app.""" return await self._backend.get_current_notifications() + def get_cached_notifications(self) -> dict[str, DispatchedNotification]: + """Returns the notifications known to this Desktop Notifier instance.""" + return self._backend.get_cached_notifications() + async def clear(self, identifier: str) -> None: """ Removes the given notification from the notification center. @@ -275,12 +283,48 @@ async def get_capabilities(self) -> frozenset[Capability]: self._capabilities = await self._backend.get_capabilities() return self._capabilities + @property + def on_dispatched(self) -> Callable[[str], Any] | None: + """ + A method to call when a notification is sent to the notifications server + + The method must take the notification identifier as a single argument. + + If the notification itself already specifies an on_dispatched handler, it will + be used instead of the class-level handler. + """ + return self._backend.on_dispatched + + @on_dispatched.setter + def on_dispatched(self, handler: Callable[[str], Any] | None) -> None: + self._backend.on_dispatched = handler + + @property + def on_cleared(self) -> Callable[[str], Any] | None: + """ + A method to call when a notification is cleared without user interaction + (e.g. after a timeout, or if cleared by another process) + + The method must take the notification identifier as a single argument. + + If the notification itself already specifies an on_cleared handler, it will be + used instead of the class-level handler. + """ + return self._backend.on_cleared + + @on_cleared.setter + def on_cleared(self, handler: Callable[[str], Any] | None) -> None: + self._backend.on_cleared = handler + @property def on_clicked(self) -> Callable[[str], Any] | None: """ A method to call when a notification is clicked - The method must take the notification identifier as a single argument. + The method must take the notification identifier as a single argument. You must + check whether the given identifier matches any of the notifications you care + about, because the notifications server might signal events of other + applications as well. If the notification itself already specifies an on_clicked handler, it will be used instead of the class-level handler. @@ -296,7 +340,10 @@ def on_dismissed(self) -> Callable[[str], Any] | None: """ A method to call when a notification is dismissed - The method must take the notification identifier as a single argument. + The method must take the notification identifier as a single argument. You must + check whether the given identifier matches any of the notifications you care + about, because the notifications server might signal events of other + applications as well. If the notification itself already specifies an on_dismissed handler, it will be used instead of the class-level handler. @@ -310,10 +357,12 @@ def on_dismissed(self, handler: Callable[[str], Any] | None) -> None: @property def on_button_pressed(self) -> Callable[[str, str], Any] | None: """ - A method to call when a notification is dismissed + A method to call when one of the notification's buttons is clicked The method must take the notification identifier and the button identifier as - arguments. + arguments. You must check whether the given identifier matches any of the + notifications you care about, because the notifications server might signal + events of other applications as well. If the notification button itself already specifies an on_pressed handler, it will be used instead of the class-level handler. @@ -330,6 +379,9 @@ def on_replied(self) -> Callable[[str, str], Any] | None: A method to call when a user responds through the reply field of a notification The method must take the notification identifier and input text as arguments. + You must check whether the given identifier matches any of the notifications + you care about, because the notifications server might signal events of other + applications as well. If the notification's reply field itself already specifies an on_replied handler, it will be used instead of the class-level handler. diff --git a/src/desktop_notifier/sync.py b/src/desktop_notifier/sync.py index 447f240f..46dfecec 100644 --- a/src/desktop_notifier/sync.py +++ b/src/desktop_notifier/sync.py @@ -12,6 +12,7 @@ Attachment, Button, Capability, + DispatchedNotification, Icon, Notification, ReplyField, @@ -41,7 +42,7 @@ def __init__( app_icon: Icon | None = DEFAULT_ICON, notification_limit: int | None = None, ) -> None: - self._async_api = DesktopNotifier(app_name, app_icon) + self._async_api = DesktopNotifier(app_name, app_icon, notification_limit) self._loop = asyncio.new_event_loop() def _run_coro_sync(self, coro: Coroutine[None, None, T]) -> T: @@ -62,9 +63,17 @@ def app_name(self) -> str: @app_name.setter def app_name(self, value: str) -> None: - """Setter: app_name""" self._async_api.app_name = value + @property + def app_icon(self) -> Icon | None: + """The application icon""" + return self._async_api.app_icon + + @app_icon.setter + def app_icon(self, value: Icon | None) -> None: + self._async_api.app_icon = value + def request_authorisation(self) -> bool: """See :meth:`desktop_notifier.main.DesktopNotifier.request_authorisation`""" coro = self._async_api.request_authorisation() @@ -75,7 +84,9 @@ def has_authorisation(self) -> bool: coro = self._async_api.has_authorisation() return self._run_coro_sync(coro) - def send_notification(self, notification: Notification) -> str: + def send_notification( + self, notification: Notification | DispatchedNotification + ) -> DispatchedNotification | None: """See :meth:`desktop_notifier.main.DesktopNotifier.send_notification`""" coro = self._async_api.send_notification(notification) return self._run_coro_sync(coro) @@ -88,13 +99,15 @@ def send( icon: Icon | None = None, buttons: Sequence[Button] = (), reply_field: ReplyField | None = None, + on_dispatched: Callable[[], Any] | None = None, + on_cleared: Callable[[], Any] | None = None, on_clicked: Callable[[], Any] | None = None, on_dismissed: Callable[[], Any] | None = None, attachment: Attachment | None = None, sound: Sound | None = None, thread: str | None = None, timeout: int = -1, # in seconds - ) -> str: + ) -> DispatchedNotification | None: """See :meth:`desktop_notifier.main.DesktopNotifier.send`""" notification = Notification( title, @@ -103,6 +116,8 @@ def send( icon=icon, buttons=tuple(buttons), reply_field=reply_field, + on_dispatched=on_dispatched, + on_cleared=on_cleared, on_clicked=on_clicked, on_dismissed=on_dismissed, attachment=attachment, @@ -132,3 +147,101 @@ def get_capabilities(self) -> frozenset[Capability]: """See :meth:`desktop_notifier.main.DesktopNotifier.get_capabilities`""" coro = self._async_api.get_capabilities() return self._run_coro_sync(coro) + + @property + def on_dispatched(self) -> Callable[[str], Any] | None: + """ + A method to call when a notification is sent to the notifications server + + The method must take the notification identifier as a single argument. + + If the notification itself already specifies an on_dispatched handler, it will + be used instead of the class-level handler. + """ + return self._async_api.on_dispatched + + @on_dispatched.setter + def on_dispatched(self, handler: Callable[[str], Any] | None) -> None: + self._async_api.on_dispatched = handler + + @property + def on_cleared(self) -> Callable[[str], Any] | None: + """ + A method to call when a notification is cleared without user interaction + (e.g. after a timeout, or if cleared by another process) + + The method must take the notification identifier as a single argument. + + If the notification itself already specifies an on_cleared handler, it will be + used instead of the class-level handler. + """ + return self._async_api.on_cleared + + @on_cleared.setter + def on_cleared(self, handler: Callable[[str], Any] | None) -> None: + self._async_api.on_cleared = handler + + @property + def on_clicked(self) -> Callable[[str], Any] | None: + """ + A method to call when a notification is clicked + + The method must take the notification identifier as a single argument. + + If the notification itself already specifies an on_clicked handler, it will be + used instead of the class-level handler. + """ + return self._async_api.on_clicked + + @on_clicked.setter + def on_clicked(self, handler: Callable[[str], Any] | None) -> None: + self._async_api.on_clicked = handler + + @property + def on_dismissed(self) -> Callable[[str], Any] | None: + """ + A method to call when a notification is dismissed + + The method must take the notification identifier as a single argument. + + If the notification itself already specifies an on_dismissed handler, it will be + used instead of the class-level handler. + """ + return self._async_api.on_dismissed + + @on_dismissed.setter + def on_dismissed(self, handler: Callable[[str], Any] | None) -> None: + self._async_api.on_dismissed = handler + + @property + def on_button_pressed(self) -> Callable[[str, str], Any] | None: + """ + A method to call when a notification is dismissed + + The method must take the notification identifier and the button identifier as + arguments. + + If the notification button itself already specifies an on_pressed handler, it + will be used instead of the class-level handler. + """ + return self._async_api.on_button_pressed + + @on_button_pressed.setter + def on_button_pressed(self, handler: Callable[[str, str], Any] | None) -> None: + self._async_api.on_button_pressed = handler + + @property + def on_replied(self) -> Callable[[str, str], Any] | None: + """ + A method to call when a user responds through the reply field of a notification + + The method must take the notification identifier and input text as arguments. + + If the notification's reply field itself already specifies an on_replied + handler, it will be used instead of the class-level handler. + """ + return self._async_api.on_replied + + @on_replied.setter + def on_replied(self, handler: Callable[[str, str], Any] | None) -> None: + self._async_api.on_replied = handler diff --git a/tests/backends/__init__.py b/tests/backends/__init__.py index ac5c7a36..c9f9dc43 100644 --- a/tests/backends/__init__.py +++ b/tests/backends/__init__.py @@ -1,6 +1,6 @@ import platform -from desktop_notifier import DesktopNotifier +from desktop_notifier import DesktopNotifier, DispatchedNotification __all__ = [ "simulate_button_pressed", @@ -25,18 +25,26 @@ ) else: - def simulate_clicked(notifier: DesktopNotifier, identifier: str) -> None: + def simulate_clicked( + notifier: DesktopNotifier, dispatched_notification: DispatchedNotification + ) -> None: raise NotImplementedError(f"{platform.system()} is not supported") - def simulate_dismissed(notifier: DesktopNotifier, identifier: str) -> None: + def simulate_dismissed( + notifier: DesktopNotifier, dispatched_notification: DispatchedNotification + ) -> None: raise NotImplementedError(f"{platform.system()} is not supported") def simulate_button_pressed( - notifier: DesktopNotifier, identifier: str, button_identifier: str + notifier: DesktopNotifier, + dispatched_notification: DispatchedNotification, + button_identifier: str, ) -> None: raise NotImplementedError(f"{platform.system()} is not supported") def simulate_replied( - notifier: DesktopNotifier, identifier: str, reply: str + notifier: DesktopNotifier, + dispatched_notification: DispatchedNotification, + reply: str, ) -> None: raise NotImplementedError(f"{platform.system()} is not supported") diff --git a/tests/backends/dbus.py b/tests/backends/dbus.py index 638013b9..76fef7c0 100644 --- a/tests/backends/dbus.py +++ b/tests/backends/dbus.py @@ -1,46 +1,53 @@ from __future__ import annotations -from desktop_notifier import DesktopNotifier +from desktop_notifier import DesktopNotifier, DispatchedNotification from desktop_notifier.backends.dbus import ( NOTIFICATION_CLOSED_DISMISSED, DBusDesktopNotifier, ) -def simulate_clicked(notifier: DesktopNotifier, identifier: str) -> None: +def simulate_clicked( + notifier: DesktopNotifier, dispatched_notification: DispatchedNotification +) -> None: assert isinstance(notifier._backend, DBusDesktopNotifier) + assert isinstance(dispatched_notification, DispatchedNotification) - nid = notifier._backend._platform_to_interface_notification_identifier.inverse[ - identifier - ] + nid = int(dispatched_notification.identifier) notifier._backend._on_action(nid, "default") # Any closing of the notification also triggers a dismissed event. notifier._backend._on_closed(nid, NOTIFICATION_CLOSED_DISMISSED) -def simulate_dismissed(notifier: DesktopNotifier, identifier: str) -> None: +def simulate_dismissed( + notifier: DesktopNotifier, dispatched_notification: DispatchedNotification +) -> None: assert isinstance(notifier._backend, DBusDesktopNotifier) + assert isinstance(dispatched_notification, DispatchedNotification) - nid = notifier._backend._platform_to_interface_notification_identifier.inverse[ - identifier - ] + nid = int(dispatched_notification.identifier) notifier._backend._on_closed(nid, NOTIFICATION_CLOSED_DISMISSED) def simulate_button_pressed( - notifier: DesktopNotifier, identifier: str, button_identifier: str + notifier: DesktopNotifier, + dispatched_notification: DispatchedNotification, + button_identifier: str, ) -> None: assert isinstance(notifier._backend, DBusDesktopNotifier) + assert isinstance(dispatched_notification, DispatchedNotification) - nid = notifier._backend._platform_to_interface_notification_identifier.inverse[ - identifier - ] + nid = int(dispatched_notification.identifier) notifier._backend._on_action(nid, button_identifier) # Any closing of the notification also triggers a dismissed event. notifier._backend._on_closed(nid, NOTIFICATION_CLOSED_DISMISSED) -def simulate_replied(notifier: DesktopNotifier, identifier: str, reply: str) -> None: +def simulate_replied( + notifier: DesktopNotifier, + dispatched_notification: DispatchedNotification, + reply: str, +) -> None: raise NotImplementedError("Relied callbacks on supported on Linux") diff --git a/tests/backends/macos.py b/tests/backends/macos.py index 581c5f44..3c658437 100644 --- a/tests/backends/macos.py +++ b/tests/backends/macos.py @@ -5,7 +5,7 @@ from rubicon.objc.api import Block, ObjCClass from rubicon.objc.runtime import load_library -from desktop_notifier import DesktopNotifier +from desktop_notifier import DesktopNotifier, DispatchedNotification from desktop_notifier.backends.macos import ( CocoaNotificationCenter, ReplyActionIdentifier, @@ -28,34 +28,60 @@ UNMutableNotificationContent = ObjCClass("UNMutableNotificationContent") -def simulate_clicked(notifier: DesktopNotifier, identifier: str) -> None: +def simulate_clicked( + notifier: DesktopNotifier, dispatched_notification: DispatchedNotification +) -> None: assert isinstance(notifier._backend, CocoaNotificationCenter) - _send_response(notifier._backend, identifier, UNNotificationDefaultActionIdentifier) + assert isinstance(dispatched_notification, DispatchedNotification) + _send_response( + notifier._backend, + dispatched_notification, + UNNotificationDefaultActionIdentifier, + ) -def simulate_dismissed(notifier: DesktopNotifier, identifier: str) -> None: +def simulate_dismissed( + notifier: DesktopNotifier, dispatched_notification: DispatchedNotification +) -> None: assert isinstance(notifier._backend, CocoaNotificationCenter) - _send_response(notifier._backend, identifier, UNNotificationDismissActionIdentifier) + assert isinstance(dispatched_notification, DispatchedNotification) + _send_response( + notifier._backend, + dispatched_notification, + UNNotificationDismissActionIdentifier, + ) def simulate_button_pressed( - notifier: DesktopNotifier, identifier: str, button_identifier: str + notifier: DesktopNotifier, + dispatched_notification: DispatchedNotification, + button_identifier: str, ) -> None: assert isinstance(notifier._backend, CocoaNotificationCenter) - _send_response(notifier._backend, identifier, button_identifier) + assert isinstance(dispatched_notification, DispatchedNotification) + _send_response(notifier._backend, dispatched_notification, button_identifier) -def simulate_replied(notifier: DesktopNotifier, identifier: str, reply: str) -> None: +def simulate_replied( + notifier: DesktopNotifier, + dispatched_notification: DispatchedNotification, + reply: str, +) -> None: assert isinstance(notifier._backend, CocoaNotificationCenter) - _send_response(notifier._backend, identifier, ReplyActionIdentifier, reply) + assert isinstance(dispatched_notification, DispatchedNotification) + _send_response( + notifier._backend, dispatched_notification, ReplyActionIdentifier, reply + ) def _send_response( backend: CocoaNotificationCenter, - identifier: str, + dispatched_notification: DispatchedNotification, action_identifier: str, response_text: str | None = None, ) -> None: + identifier = dispatched_notification.identifier + content = UNMutableNotificationContent.alloc().init() request = UNNotificationRequest.requestWithIdentifier( identifier, content=content, trigger=None diff --git a/tests/conftest.py b/tests/conftest.py index fe8d12fb..6c136345 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,14 +1,22 @@ from __future__ import annotations import asyncio +import os import platform +import sys import time from typing import AsyncGenerator, Generator import pytest import pytest_asyncio -from desktop_notifier import DesktopNotifier, DesktopNotifierSync +try: + from desktop_notifier import DesktopNotifier, DesktopNotifierSync +except ModuleNotFoundError: + import_dir = os.path.dirname(os.path.realpath(__file__)) + sys.path.append(import_dir + "/../src") + + from desktop_notifier import DesktopNotifier, DesktopNotifierSync if platform.system() == "Darwin": from rubicon.objc.eventloop import EventLoopPolicy diff --git a/tests/test_api.py b/tests/test_api.py index ae60f3f0..00cd18fb 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -10,7 +10,9 @@ Attachment, Button, DesktopNotifier, + DispatchedNotification, Icon, + Notification, ReplyField, Sound, Urgency, @@ -45,7 +47,7 @@ async def test_request_authorisation(notifier: DesktopNotifier) -> None: @pytest.mark.asyncio async def test_send(notifier: DesktopNotifier) -> None: - notification = await notifier.send( + dispatched_notification = await notifier.send( title="Julius Caesar", message="Et tu, Brute?", urgency=Urgency.Critical, @@ -57,14 +59,68 @@ async def test_send(notifier: DesktopNotifier) -> None: button_title="Send", on_replied=lambda text: print("Brutus replied:", text), ), + on_dispatched=lambda: print("Notification showing"), on_clicked=lambda: print("Notification clicked"), on_dismissed=lambda: print("Notification dismissed"), sound=DEFAULT_SOUND, thread="test_notifications", timeout=5, ) + assert isinstance(dispatched_notification, DispatchedNotification) + await wait_for_notifications(notifier) - assert notification in await notifier.get_current_notifications() + + current_notifications = await notifier.get_current_notifications() + assert len(current_notifications) == 1 + assert dispatched_notification.identifier in current_notifications + + +@pytest.mark.asyncio +async def test_send_twice(notifier: DesktopNotifier) -> None: + notification = Notification( + title="Julius Caesar", + message="Et tu, Brute?", + ) + + n0 = await notifier.send_notification(notification) + assert isinstance(n0, DispatchedNotification) + + n1 = await notifier.send_notification(notification) + assert isinstance(n1, DispatchedNotification) + + await wait_for_notifications(notifier, 2) + + current_notifications = await notifier.get_current_notifications() + assert len(current_notifications) == 2 + assert n0.identifier in current_notifications + assert n1.identifier in current_notifications + + +@pytest.mark.asyncio +@pytest.mark.skipif( + sys.platform.startswith("win"), + reason="Clearing individual notifications is broken on Windows, so is resending notifications", +) +async def test_resend(notifier: DesktopNotifier) -> None: + notification = Notification( + title="Julius Caesar", + message="Et tu, Brute?", + ) + + n0 = await notifier.send_notification(notification) + assert isinstance(n0, DispatchedNotification) + + await wait_for_notifications(notifier) + + n1 = await notifier.send_notification(n0) + assert isinstance(n1, DispatchedNotification) + assert n1.identifier == n0.identifier + + await wait_for_notifications(notifier) + + current_notifications = await notifier.get_current_notifications() + assert len(current_notifications) == 1 + assert n1.identifier in current_notifications @pytest.mark.asyncio @@ -141,14 +197,25 @@ async def test_clear(notifier: DesktopNotifier) -> None: title="Julius Caesar", message="Et tu, Brute?", ) + + assert isinstance(n0, DispatchedNotification) + assert isinstance(n1, DispatchedNotification) + await wait_for_notifications(notifier, 2) - current_notifications = await notifier.get_current_notifications() - assert n0 in current_notifications - assert n1 in current_notifications + nlist0 = await notifier.get_current_notifications() + assert len(nlist0) == 2 + assert n0.identifier in nlist0 + assert n1.identifier in nlist0 + + await notifier.clear(n0.identifier) - await notifier.clear(n0) - assert n0 not in await notifier.get_current_notifications() + await wait_for_notifications(notifier, 1) + + nlist1 = await notifier.get_current_notifications() + assert len(nlist1) == 1 + assert n0.identifier not in nlist1 + assert n1.identifier in nlist1 @pytest.mark.asyncio @@ -161,11 +228,19 @@ async def test_clear_all(notifier: DesktopNotifier) -> None: title="Julius Caesar", message="Et tu, Brute?", ) + + assert isinstance(n0, DispatchedNotification) + assert isinstance(n1, DispatchedNotification) + await wait_for_notifications(notifier, 2) - current_notifications = await notifier.get_current_notifications() - assert n0 in current_notifications - assert n1 in current_notifications + current_notifications = await notifier.get_current_notifications() + assert len(current_notifications) == 2 + assert n0.identifier in current_notifications + assert n1.identifier in current_notifications await notifier.clear_all() + + await wait_for_notifications(notifier, 0) + assert len(await notifier.get_current_notifications()) == 0 diff --git a/tests/test_callbacks.py b/tests/test_callbacks.py index 2347e6e0..618a9dae 100644 --- a/tests/test_callbacks.py +++ b/tests/test_callbacks.py @@ -7,6 +7,7 @@ Button, Capability, DesktopNotifier, + DispatchedNotification, Notification, ReplyField, ) @@ -25,7 +26,68 @@ async def check_supported(notifier: DesktopNotifier, capability: Capability) -> None: capabilities = await notifier.get_capabilities() if capability not in capabilities: - pytest.skip(f"{notifier} not supported by backend") + pytest.skip(f"{capability} not supported by {notifier} backend") + + +@pytest.mark.asyncio +async def test_cleared_callback_called(notifier: DesktopNotifier) -> None: + await check_supported(notifier, Capability.ON_CLEARED) + + class_handler = Mock() + notification_handler = Mock() + notifier.on_cleared = class_handler + notification = Notification( + title="Julius Caesar", + message="Et tu, Brute?", + on_cleared=notification_handler, + ) + + dispatched_notification = await notifier.send_notification(notification) + assert isinstance(dispatched_notification, DispatchedNotification) + + await notifier.clear(dispatched_notification.identifier) + + class_handler.assert_not_called() + notification_handler.assert_called_once() + + +@pytest.mark.asyncio +async def test_cleared_callback_not_dismissed(notifier: DesktopNotifier) -> None: + await check_supported(notifier, Capability.ON_CLEARED) + + on_cleared = Mock() + on_dismissed = Mock() + notification = Notification( + title="Julius Caesar", + message="Et tu, Brute?", + on_cleared=on_cleared, + on_dismissed=on_dismissed, + ) + + dispatched_notification = await notifier.send_notification(notification) + assert isinstance(dispatched_notification, DispatchedNotification) + + await notifier.clear(dispatched_notification.identifier) + + on_dismissed.assert_not_called() + on_cleared.assert_called_once() + + +@pytest.mark.asyncio +async def test_dispatched_callback_called(notifier: DesktopNotifier) -> None: + class_handler = Mock() + notification_handler = Mock() + notifier.on_dispatched = class_handler + notification = Notification( + title="Julius Caesar", + message="Et tu, Brute?", + on_dispatched=notification_handler, + ) + + await notifier.send_notification(notification) + + class_handler.assert_not_called() + notification_handler.assert_called_once() @pytest.mark.asyncio @@ -41,8 +103,10 @@ async def test_clicked_callback_called(notifier: DesktopNotifier) -> None: on_clicked=notification_handler, ) - identifier = await notifier.send_notification(notification) - simulate_clicked(notifier, identifier) + dispatched_notification = await notifier.send_notification(notification) + assert isinstance(dispatched_notification, DispatchedNotification) + + simulate_clicked(notifier, dispatched_notification) class_handler.assert_not_called() notification_handler.assert_called_once() @@ -65,8 +129,10 @@ async def test_clicked_callback_dismissed_not_called(notifier: DesktopNotifier) on_dismissed=on_dismissed, ) - identifier = await notifier.send_notification(notification) - simulate_clicked(notifier, identifier) + dispatched_notification = await notifier.send_notification(notification) + assert isinstance(dispatched_notification, DispatchedNotification) + + simulate_clicked(notifier, dispatched_notification) on_dismissed.assert_not_called() on_clicked.assert_called_once() @@ -85,8 +151,10 @@ async def test_dismissed_callback_called(notifier: DesktopNotifier) -> None: on_dismissed=notification_handler, ) - identifier = await notifier.send_notification(notification) - simulate_dismissed(notifier, identifier) + dispatched_notification = await notifier.send_notification(notification) + assert isinstance(dispatched_notification, DispatchedNotification) + + simulate_dismissed(notifier, dispatched_notification) class_handler.assert_not_called() notification_handler.assert_called_once() @@ -108,8 +176,10 @@ async def test_button_pressed_callback_called(notifier: DesktopNotifier) -> None buttons=(button0, button1), ) - identifier = await notifier.send_notification(notification) - simulate_button_pressed(notifier, identifier, button1.identifier) + dispatched_notification = await notifier.send_notification(notification) + assert isinstance(dispatched_notification, DispatchedNotification) + + simulate_button_pressed(notifier, dispatched_notification, button1.identifier) class_handler.assert_not_called() notification_b1_handler.assert_called_once() @@ -128,13 +198,43 @@ async def test_replied_callback_called(notifier: DesktopNotifier) -> None: reply_field=ReplyField(on_replied=notification_handler), ) - identifier = await notifier.send_notification(notification) - simulate_replied(notifier, identifier, "A notification response") + dispatched_notification = await notifier.send_notification(notification) + assert isinstance(dispatched_notification, DispatchedNotification) + + simulate_replied(notifier, dispatched_notification, "A notification response") class_handler.assert_not_called() notification_handler.assert_called_with("A notification response") +@pytest.mark.asyncio +async def test_dispatched_fallback_handler_called(notifier: DesktopNotifier) -> None: + class_handler = Mock() + notifier.on_dispatched = class_handler + notification = Notification(title="Julius Caesar", message="Et tu, Brute?") + + dispatched_notification = await notifier.send_notification(notification) + assert isinstance(dispatched_notification, DispatchedNotification) + + class_handler.assert_called_with(dispatched_notification.identifier) + + +@pytest.mark.asyncio +async def test_cleared_fallback_handler_called(notifier: DesktopNotifier) -> None: + await check_supported(notifier, Capability.ON_CLEARED) + + class_handler = Mock() + notifier.on_cleared = class_handler + notification = Notification(title="Julius Caesar", message="Et tu, Brute?") + + dispatched_notification = await notifier.send_notification(notification) + assert isinstance(dispatched_notification, DispatchedNotification) + + await notifier.clear(dispatched_notification.identifier) + + class_handler.assert_called_with(dispatched_notification.identifier) + + @pytest.mark.asyncio async def test_clicked_fallback_handler_called(notifier: DesktopNotifier) -> None: await check_supported(notifier, Capability.ON_CLICKED) @@ -143,10 +243,12 @@ async def test_clicked_fallback_handler_called(notifier: DesktopNotifier) -> Non notifier.on_clicked = class_handler notification = Notification(title="Julius Caesar", message="Et tu, Brute?") - identifier = await notifier.send_notification(notification) - simulate_clicked(notifier, identifier) + dispatched_notification = await notifier.send_notification(notification) + assert isinstance(dispatched_notification, DispatchedNotification) - class_handler.assert_called_with(identifier) + simulate_clicked(notifier, dispatched_notification) + + class_handler.assert_called_with(dispatched_notification.identifier) @pytest.mark.asyncio @@ -157,10 +259,12 @@ async def test_dismissed_fallback_handler_called(notifier: DesktopNotifier) -> N notifier.on_dismissed = class_handler notification = Notification(title="Julius Caesar", message="Et tu, Brute?") - identifier = await notifier.send_notification(notification) - simulate_dismissed(notifier, identifier) + dispatched_notification = await notifier.send_notification(notification) + assert isinstance(dispatched_notification, DispatchedNotification) + + simulate_dismissed(notifier, dispatched_notification) - class_handler.assert_called_with(identifier) + class_handler.assert_called_with(dispatched_notification.identifier) @pytest.mark.asyncio @@ -179,10 +283,14 @@ async def test_button_pressed_fallback_handler_called( buttons=(button0, button1), ) - identifier = await notifier.send_notification(notification) - simulate_button_pressed(notifier, identifier, button1.identifier) + dispatched_notification = await notifier.send_notification(notification) + assert isinstance(dispatched_notification, DispatchedNotification) - class_handler.assert_called_once_with(identifier, button1.identifier) + simulate_button_pressed(notifier, dispatched_notification, button1.identifier) + + class_handler.assert_called_once_with( + dispatched_notification.identifier, button1.identifier + ) @pytest.mark.asyncio @@ -197,7 +305,11 @@ async def test_replied_fallback_handler_called(notifier: DesktopNotifier) -> Non reply_field=ReplyField(), ) - identifier = await notifier.send_notification(notification) - simulate_replied(notifier, identifier, "A notification response") + dispatched_notification = await notifier.send_notification(notification) + assert isinstance(dispatched_notification, DispatchedNotification) - class_handler.assert_called_with(identifier, "A notification response") + simulate_replied(notifier, dispatched_notification, "A notification response") + + class_handler.assert_called_with( + dispatched_notification.identifier, "A notification response" + ) diff --git a/tests/test_sync_api.py b/tests/test_sync_api.py index 76b86b57..cfcf397b 100644 --- a/tests/test_sync_api.py +++ b/tests/test_sync_api.py @@ -7,6 +7,7 @@ DEFAULT_SOUND, Button, DesktopNotifierSync, + DispatchedNotification, ReplyField, Urgency, ) @@ -26,7 +27,7 @@ def wait_for_notifications( def test_send(notifier_sync: DesktopNotifierSync) -> None: - notification = notifier_sync.send( + dispatched_notification = notifier_sync.send( title="Julius Caesar", message="Et tu, Brute?", urgency=Urgency.Critical, @@ -47,8 +48,12 @@ def test_send(notifier_sync: DesktopNotifierSync) -> None: thread="test_notifications", timeout=5, ) + assert isinstance(dispatched_notification, DispatchedNotification) + wait_for_notifications(notifier_sync) - assert notification in notifier_sync.get_current_notifications() + + current_notifications = notifier_sync.get_current_notifications() + assert dispatched_notification.identifier in current_notifications @pytest.mark.skipif( @@ -64,14 +69,25 @@ def test_clear(notifier_sync: DesktopNotifierSync) -> None: title="Julius Caesar", message="Et tu, Brute?", ) + + assert isinstance(n0, DispatchedNotification) + assert isinstance(n1, DispatchedNotification) + wait_for_notifications(notifier_sync, 2) - current_notifications = notifier_sync.get_current_notifications() - assert n0 in current_notifications - assert n1 in current_notifications + nlist0 = notifier_sync.get_current_notifications() + assert len(nlist0) == 2 + assert n0.identifier in nlist0 + assert n1.identifier in nlist0 + + notifier_sync.clear(n0.identifier) - notifier_sync.clear(n0) - assert n0 not in notifier_sync.get_current_notifications() + wait_for_notifications(notifier_sync, 1) + + nlist1 = notifier_sync.get_current_notifications() + assert len(nlist1) == 1 + assert n0.identifier not in nlist1 + assert n1.identifier in nlist1 def test_clear_all(notifier_sync: DesktopNotifierSync) -> None: @@ -83,11 +99,19 @@ def test_clear_all(notifier_sync: DesktopNotifierSync) -> None: title="Julius Caesar", message="Et tu, Brute?", ) + + assert isinstance(n0, DispatchedNotification) + assert isinstance(n1, DispatchedNotification) + wait_for_notifications(notifier_sync, 2) - current_notifications = notifier_sync.get_current_notifications() - assert n0 in current_notifications - assert n1 in current_notifications + current_notifications = notifier_sync.get_current_notifications() + assert len(current_notifications) == 2 + assert n0.identifier in current_notifications + assert n1.identifier in current_notifications notifier_sync.clear_all() + + wait_for_notifications(notifier_sync, 0) + assert len(notifier_sync.get_current_notifications()) == 0