From d68e49fe64dcc6dfc9e8cd532767ee15bc9ec2de Mon Sep 17 00:00:00 2001 From: Kovid Goyal Date: Wed, 24 Jul 2024 22:11:02 +0530 Subject: [PATCH] Start work on refactoring notifications handling Makes the code cleaner and easily mockable for testing. Also, add code to handle closing notifications on Linux. --- glfw/glfw.py | 2 +- glfw/linux_notify.c | 38 ++- glfw/linux_notify.h | 2 +- kitty/boss.py | 40 +-- kitty/clipboard.py | 2 +- kitty/fast_data_types.pyi | 3 + kitty/glfw-wrapper.h | 2 +- kitty/glfw.c | 25 +- kitty/notifications.py | 550 +++++++++++++++++++++++++++++++++++ kitty/notify.py | 328 --------------------- kitty/update_check.py | 7 +- kitty/window.py | 44 +-- kitty_tests/notifications.py | 164 +++++++++++ kitty_tests/parser.py | 88 ------ 14 files changed, 798 insertions(+), 497 deletions(-) create mode 100644 kitty/notifications.py delete mode 100644 kitty/notify.py create mode 100644 kitty_tests/notifications.py diff --git a/glfw/glfw.py b/glfw/glfw.py index afe47ed86..d2d14661a 100755 --- a/glfw/glfw.py +++ b/glfw/glfw.py @@ -364,7 +364,7 @@ typedef bool (* GLFWcocoatogglefullscreenfun)(GLFWwindow*); typedef void (* GLFWcocoarenderframefun)(GLFWwindow*); typedef void (*GLFWwaylandframecallbackfunc)(unsigned long long id); typedef void (*GLFWDBusnotificationcreatedfun)(unsigned long long, uint32_t, void*); -typedef void (*GLFWDBusnotificationactivatedfun)(uint32_t, const char*); +typedef void (*GLFWDBusnotificationactivatedfun)(uint32_t, int, const char*); {} const char* load_glfw(const char* path); diff --git a/glfw/linux_notify.c b/glfw/linux_notify.c index f52c00169..80319aeaa 100644 --- a/glfw/linux_notify.c +++ b/glfw/linux_notify.c @@ -50,26 +50,58 @@ message_handler(DBusConnection *conn UNUSED, DBusMessage *msg, void *user_data U /* printf("session_bus message_handler invoked interface: %s member: %s\n", dbus_message_get_interface(msg), dbus_message_get_member(msg)); */ if (dbus_message_is_signal(msg, NOTIFICATIONS_IFACE, "ActionInvoked")) { uint32_t id; - const char *action; + const char *action = NULL; if (glfw_dbus_get_args(msg, "Failed to get args from ActionInvoked notification signal", DBUS_TYPE_UINT32, &id, DBUS_TYPE_STRING, &action, DBUS_TYPE_INVALID)) { if (activated_handler) { - activated_handler(id, action); + activated_handler(id, 2, action); return DBUS_HANDLER_RESULT_HANDLED; } } } + + if (dbus_message_is_signal(msg, NOTIFICATIONS_IFACE, "ActivationToken")) { + uint32_t id; + const char *token = NULL; + if (glfw_dbus_get_args(msg, "Failed to get args from ActivationToken notification signal", + DBUS_TYPE_UINT32, &id, DBUS_TYPE_STRING, &token, DBUS_TYPE_INVALID)) { + if (activated_handler) { + activated_handler(id, 1, token); + return DBUS_HANDLER_RESULT_HANDLED; + } + } + + } + + if (dbus_message_is_signal(msg, NOTIFICATIONS_IFACE, "NotificationClosed")) { + uint32_t id; + if (glfw_dbus_get_args(msg, "Failed to get args from NotificationClosed notification signal", + DBUS_TYPE_UINT32, &id, DBUS_TYPE_INVALID)) { + if (activated_handler) { + activated_handler(id, 0, ""); + return DBUS_HANDLER_RESULT_HANDLED; + } + } + } return DBUS_HANDLER_RESULT_NOT_YET_HANDLED; } +static bool +cancel_user_notification(DBusConnection *session_bus, uint32_t *id) { + return glfw_dbus_call_method_no_reply(session_bus, NOTIFICATIONS_SERVICE, NOTIFICATIONS_PATH, NOTIFICATIONS_IFACE, "CloseNotification", DBUS_TYPE_UINT32, id, DBUS_TYPE_INVALID); +} + notification_id_type glfw_dbus_send_user_notification(const char *app_name, const char* icon, const char *summary, const char *body, const char* action_name, int32_t timeout, int urgency, GLFWDBusnotificationcreatedfun callback, void *user_data) { DBusConnection *session_bus = glfw_dbus_session_bus(); - static DBusConnection *added_signal_match = NULL; if (!session_bus) return 0; + if (timeout == -9999 && urgency == -9999) return cancel_user_notification(session_bus, user_data) ? 1 : 0; + static DBusConnection *added_signal_match = NULL; if (added_signal_match != session_bus) { dbus_bus_add_match(session_bus, "type='signal',interface='" NOTIFICATIONS_IFACE "',member='ActionInvoked'", NULL); + dbus_bus_add_match(session_bus, "type='signal',interface='" NOTIFICATIONS_IFACE "',member='NotificationClosed'", NULL); + dbus_bus_add_match(session_bus, "type='signal',interface='" NOTIFICATIONS_IFACE "',member='ActivationToken'", NULL); dbus_connection_add_filter(session_bus, message_handler, NULL, NULL); added_signal_match = session_bus; } diff --git a/glfw/linux_notify.h b/glfw/linux_notify.h index 7369a3e43..a60e7f9a1 100644 --- a/glfw/linux_notify.h +++ b/glfw/linux_notify.h @@ -12,7 +12,7 @@ typedef unsigned long long notification_id_type; typedef void (*GLFWDBusnotificationcreatedfun)(notification_id_type, uint32_t, void*); -typedef void (*GLFWDBusnotificationactivatedfun)(uint32_t, const char*); +typedef void (*GLFWDBusnotificationactivatedfun)(uint32_t, int, const char*); notification_id_type glfw_dbus_send_user_notification(const char *app_name, const char* icon, const char *summary, const char *body, const char *action_name, int32_t timeout, int urgency, GLFWDBusnotificationcreatedfun, void*); void diff --git a/kitty/boss.py b/kitty/boss.py index c0a01ec4b..acf6b04a3 100644 --- a/kitty/boss.py +++ b/kitty/boss.py @@ -121,7 +121,7 @@ from .fast_data_types import ( from .key_encoding import get_name_to_functional_number_map from .keys import Mappings from .layout.base import set_layout_options -from .notify import notification_activated +from .notifications import NotificationManager from .options.types import Options, nullable_colors from .options.utils import MINIMUM_FONT_SIZE, KeyboardMode, KeyDefinition from .os_window_size import initial_window_size_func @@ -372,13 +372,11 @@ class Boss: DumpCommands(args) if args.dump_commands or args.dump_bytes else None, talk_fd, listen_fd, ) - set_boss(self) self.args: CLIOptions = args self.mouse_handler: Optional[Callable[[WindowSystemMouseEvent], None]] = None + set_boss(self) self.mappings: Mappings = Mappings(global_shortcuts, self.refresh_active_tab_bar) - if is_macos: - from .fast_data_types import cocoa_set_notification_activated_callback - cocoa_set_notification_activated_callback(notification_activated) + self.notification_manager: NotificationManager = NotificationManager() def startup_first_child(self, os_window_id: Optional[int], startup_sessions: Iterable[Session] = ()) -> None: si = startup_sessions or create_sessions(get_options(), self.args, default_session=get_options().startup_session) @@ -538,7 +536,9 @@ class Boss: if q: yield q - def set_active_window(self, window: Window, switch_os_window_if_needed: bool = False, for_keep_focus: bool = False) -> Optional[int]: + def set_active_window( + self, window: Window, switch_os_window_if_needed: bool = False, for_keep_focus: bool = False, activation_token: str = '' + ) -> Optional[int]: for os_window_id, tm in self.os_window_map.items(): for tab in tm: for w in tab: @@ -546,8 +546,8 @@ class Boss: if tab is not self.active_tab: tm.set_active_tab(tab, for_keep_focus=window.tabref() if for_keep_focus else None) tab.set_active_window(w, for_keep_focus=window if for_keep_focus else None) - if switch_os_window_if_needed and current_focused_os_window_id() != os_window_id: - focus_os_window(os_window_id, True) + if activation_token or (switch_os_window_if_needed and current_focused_os_window_id() != os_window_id): + focus_os_window(os_window_id, True, activation_token) return os_window_id return None @@ -2750,14 +2750,8 @@ class Boss: except Exception as e: log_error(f'Failed to process update check data {raw!r}, with error: {e}') - def dbus_notification_callback(self, activated: bool, a: int, b: Union[int, str]) -> None: - from .notify import dbus_notification_activated, dbus_notification_created - if activated: - assert isinstance(b, str) - dbus_notification_activated(a, b) - else: - assert isinstance(b, int) - dbus_notification_created(a, b) + def dbus_notification_callback(self, event_type: str, a: int, b: Union[int, str]) -> None: + self.notification_manager.desktop_integration.dispatch_event_from_desktop(event_type, a, b) def show_bad_config_lines(self, bad_lines: Iterable[BadLine], misc_errors: Iterable[str] = ()) -> None: @@ -2984,19 +2978,7 @@ class Boss: # Can be called with kitty -o "map f1 send_test_notification" def send_test_notification(self) -> None: - from .notify import notify - now = monotonic() - ident = f'test-notify-{now}' - notify(f'Test {now}', f'At: {now}', identifier=ident, subtitle=f'Test subtitle {now}') - - def notification_activated(self, identifier: str, window_id: int, focus: bool, report: bool) -> None: - w = self.window_id_map.get(window_id) - if w is None: - return - if focus: - self.set_active_window(w, switch_os_window_if_needed=True) - if report: - w.report_notification_activated(identifier) + self.notification_manager.send_test_notification() @ac('debug', 'Show the environment variables that the kitty process sees') def show_kitty_env_vars(self) -> None: diff --git a/kitty/clipboard.py b/kitty/clipboard.py index f195eb87b..d9f94f2bd 100644 --- a/kitty/clipboard.py +++ b/kitty/clipboard.py @@ -339,7 +339,7 @@ class ClipboardRequestManager: def parse_osc_5522(self, data: memoryview) -> None: import base64 - from .notify import sanitize_id + from .notifications import sanitize_id idx = find_in_memoryview(data, ord(b';')) if idx > -1: metadata = str(data[:idx], "utf-8", "replace") diff --git a/kitty/fast_data_types.pyi b/kitty/fast_data_types.pyi index ed0152613..8c1df5f2d 100644 --- a/kitty/fast_data_types.pyi +++ b/kitty/fast_data_types.pyi @@ -555,6 +555,9 @@ def dbus_send_notification( pass +def dbus_close_notification(dbus_notification_id: int) -> bool: ... + + def cocoa_send_notification( identifier: Optional[str], title: str, diff --git a/kitty/glfw-wrapper.h b/kitty/glfw-wrapper.h index 35d6adba2..9acc19242 100644 --- a/kitty/glfw-wrapper.h +++ b/kitty/glfw-wrapper.h @@ -1699,7 +1699,7 @@ typedef bool (* GLFWcocoatogglefullscreenfun)(GLFWwindow*); typedef void (* GLFWcocoarenderframefun)(GLFWwindow*); typedef void (*GLFWwaylandframecallbackfunc)(unsigned long long id); typedef void (*GLFWDBusnotificationcreatedfun)(unsigned long long, uint32_t, void*); -typedef void (*GLFWDBusnotificationactivatedfun)(uint32_t, const char*); +typedef void (*GLFWDBusnotificationactivatedfun)(uint32_t, int, const char*); typedef int (*glfwInit_func)(monotonic_t); GFW_EXTERN glfwInit_func glfwInit_impl; #define glfwInit glfwInit_impl diff --git a/kitty/glfw.c b/kitty/glfw.c index 7319820ff..f9b5f702a 100644 --- a/kitty/glfw.c +++ b/kitty/glfw.c @@ -1420,9 +1420,14 @@ error_callback(int error, const char* description) { #ifndef __APPLE__ static void -dbus_user_notification_activated(uint32_t notification_id, const char* action) { +dbus_user_notification_activated(uint32_t notification_id, int type, const char* action) { unsigned long nid = notification_id; - call_boss(dbus_notification_callback, "Oks", Py_True, nid, action); + const char *stype = "activated"; + switch (type) { + case 0: stype = "closed"; break; + case 1: stype = "activation_token"; break; + } + call_boss(dbus_notification_callback, "sks", stype, nid, action); } #endif @@ -2064,7 +2069,7 @@ request_frame_render(OSWindow *w) { void dbus_notification_created_callback(unsigned long long notification_id, uint32_t new_notification_id, void* data UNUSED) { unsigned long new_id = new_notification_id; - call_boss(dbus_notification_callback, "OKk", Py_False, notification_id, new_id); + call_boss(dbus_notification_callback, "sKk", "created", notification_id, new_id); } static PyObject* @@ -2080,6 +2085,19 @@ dbus_send_notification(PyObject *self UNUSED, PyObject *args) { return PyLong_FromUnsignedLongLong(notification_id); } +static PyObject* +dbus_close_notification(PyObject *self UNUSED, PyObject *args) { + unsigned int id; + if (!PyArg_ParseTuple(args, "I", &id)) return NULL; + if (!glfwDBusUserNotify) { + PyErr_SetString(PyExc_RuntimeError, "Failed to load glfwDBusUserNotify, did you call glfw_init?"); + return NULL; + } + if (glfwDBusUserNotify("", "", "", "", "", -9999, -9999, NULL, &id)) Py_RETURN_TRUE; + Py_RETURN_FALSE; +} + + #endif static PyObject* @@ -2244,6 +2262,7 @@ static PyMethodDef module_methods[] = { METHODB(strip_csi, METH_O), #ifndef __APPLE__ METHODB(dbus_send_notification, METH_VARARGS), + METHODB(dbus_close_notification, METH_VARARGS), #else {"cocoa_recreate_global_menu", (PyCFunction)py_recreate_global_menu, METH_NOARGS, ""}, {"cocoa_clear_global_shortcuts", (PyCFunction)py_clear_global_shortcuts, METH_NOARGS, ""}, diff --git a/kitty/notifications.py b/kitty/notifications.py new file mode 100644 index 000000000..ab5f40a65 --- /dev/null +++ b/kitty/notifications.py @@ -0,0 +1,550 @@ +#!/usr/bin/env python +# License: GPLv3 Copyright: 2024, Kovid Goyal + +import re +from collections import OrderedDict +from contextlib import suppress +from enum import Enum +from itertools import count +from typing import Any, Dict, FrozenSet, List, NamedTuple, Optional, Tuple, Union + +from .constants import is_macos, logo_png_file +from .fast_data_types import ESC_OSC, current_focused_os_window_id, get_boss +from .types import run_once +from .typing import WindowType +from .utils import get_custom_window_icon, log_error, sanitize_control_codes + + +class Urgency(Enum): + Low: int = 0 + Normal: int = 1 + Critical: int = 2 + + +class PayloadType(Enum): + unknown = '' + title = 'title' + body = 'body' + query = '?' + close = 'close' + + @property + def is_text(self) -> bool: + return self in (PayloadType.title, PayloadType.body, PayloadType.close, PayloadType.query) + + +class OnlyWhen(Enum): + unset = '' + always = 'always' + unfocused = 'unfocused' + invisible = 'invisible' + + +class Action(Enum): + focus = 'focus' + report = 'report' + + +class DataStore: + + def __init__(self) -> None: + self.buf: List[bytes] = [] + + def __call__(self, data: bytes) -> None: + self.buf.append(data) + + def finalise(self) -> bytes: + return b''.join(self.buf) + + +class EncodedDataStore: + + def __init__(self, data_store: DataStore) -> None: + self.current_leftover_bytes = memoryview(b'') + self.data_store = data_store + + def add_unencoded_data(self, data: Union[str, bytes]) -> None: + if isinstance(data, str): + data = data.encode('utf-8') + self.flush_encoded_data() + self.data_store(data) + + def add_base64_data(self, data: Union[str, bytes]) -> None: + if isinstance(data, str): + data = data.encode('ascii') + + def write_saving_leftover_bytes(data: bytes) -> None: + if len(data) == 0: + return + extra = len(data) % 4 + if extra > 0: + mv = memoryview(data) + self.current_leftover_bytes = memoryview(bytes(mv[-extra:])) + mv = mv[:-extra] + if len(mv) > 0: + self._write_base64_data(mv) + else: + self._write_base64_data(data) + + if len(self.current_leftover_bytes) > 0: + extra = 4 - len(self.current_leftover_bytes) + if len(data) >= extra: + self._write_base64_data(memoryview(bytes(self.current_leftover_bytes) + data[:extra])) + self.current_leftover_bytes = memoryview(b'') + data = memoryview(data)[extra:] + write_saving_leftover_bytes(data) + else: + self.current_leftover_bytes = memoryview(bytes(self.current_leftover_bytes) + data) + else: + write_saving_leftover_bytes(data) + + def _write_base64_data(self, b: bytes) -> None: + from base64 import standard_b64decode + d = standard_b64decode(b) + self.data_store(d) + + def flush_encoded_data(self) -> None: + b = self.current_leftover_bytes + self.current_leftover_bytes = memoryview(b'') + padding = 4 - len(b) + if padding in (1, 2): + self._write_base64_data(memoryview(bytes(b) + b'=' * padding)) + + def finalise(self) -> bytes: + self.flush_encoded_data() + return self.data_store.finalise() + + +def limit_size(x: str) -> str: + if len(x) > 1024: + x = x[:1024] + return x + + +class NotificationCommand: + + done: bool = True + identifier: str = '' + channel_id: int = 0 + desktop_notification_id: int = -1 + title: str = '' + body: str = '' + actions: FrozenSet[Action] = frozenset((Action.focus,)) + only_when: OnlyWhen = OnlyWhen.unset + urgency: Optional[Urgency] = None + close_response_requested: bool = False + + # payload handling + current_payload_type: PayloadType = PayloadType.title + current_payload_buffer: Optional[EncodedDataStore] = None + + # desktop integration specific fields + created_by_desktop: bool = False + activation_token: str = '' + + @property + def report_requested(self) -> bool: + return Action.report in self.actions + + @property + def focus_requested(self) -> bool: + return Action.focus in self.actions + + def __repr__(self) -> str: + return ( + f'NotificationCommand(identifier={self.identifier!r}, title={self.title!r}, body={self.body!r},' + f'actions={self.actions}, done={self.done!r}, urgency={self.urgency})') + + def parse_metadata(self, metadata: str, prev: 'NotificationCommand') -> Tuple[PayloadType, bool]: + payload_type = PayloadType.title + payload_is_encoded = False + for part in metadata.split(':'): + k, v = part.split('=', 1) + if k == 'p': + try: + payload_type = PayloadType(v) + except ValueError: + payload_type = PayloadType.unknown + elif k == 'i': + self.identifier = sanitize_id(v) + elif k == 'e': + payload_is_encoded = v == '1' + elif k == 'd': + self.done = v != '0' + elif k == 'a': + for ax in v.split(','): + if remove := ax.startswith('-'): + ax = ax.lstrip('+-') + try: + ac = Action(ax) + except ValueError: + pass + else: + if remove: + self.actions -= {ac} + else: + self.actions = self.actions.union({ac}) + elif k == 'o': + with suppress(ValueError): + self.only_when = OnlyWhen(v) + elif k == 'u': + with suppress(Exception): + self.urgency = Urgency(int(v)) + if not prev.done and prev.identifier == self.identifier: + self.actions = prev.actions.union(self.actions) + self.title = prev.title + self.body = prev.body + if self.only_when is OnlyWhen.unset: + self.only_when = prev.only_when + if self.urgency is None: + self.urgency = prev.urgency + + return payload_type, payload_is_encoded + + def create_payload_buffer(self, payload_type: PayloadType) -> EncodedDataStore: + self.current_payload_type = payload_type + return EncodedDataStore(DataStore()) + + def set_payload(self, payload_type: PayloadType, payload_is_encoded: bool, payload: str, prev_cmd: 'NotificationCommand') -> None: + if prev_cmd.current_payload_type is payload_type: + self.current_payload_type = payload_type + self.current_payload_buffer = prev_cmd.current_payload_buffer + prev_cmd.current_payload_buffer = None + else: + if prev_cmd.current_payload_buffer: + self.current_payload_type = prev_cmd.current_payload_type + self.commit_data(prev_cmd.current_payload_buffer.finalise()) + if self.current_payload_buffer is None: + self.current_payload_buffer = self.create_payload_buffer(payload_type) + if payload_is_encoded: + self.current_payload_buffer.add_base64_data(payload) + else: + self.current_payload_buffer.add_unencoded_data(payload) + + def commit_data(self, data: bytes) -> None: + if not data: + return + if self.current_payload_type.is_text: + text = data.decode('utf-8', 'replace') + if self.current_payload_type is PayloadType.title: + self.title = limit_size(self.title + text) + elif self.current_payload_type is PayloadType.body: + self.body = limit_size(self.body + text) + + def finalise(self) -> None: + if self.current_payload_buffer: + self.commit_data(self.current_payload_buffer.finalise()) + self.current_payload_buffer = None + + +class DesktopIntegration: + + def __init__(self, notification_manager: 'NotificationManager'): + self.notification_manager = notification_manager + self.initialize() + + def initialize(self) -> None: + pass + + def dispatch_event_from_desktop(self, *a: Any) -> None: + raise NotImplementedError('Implement me in subclass') + + def close_notification(self, desktop_notification_id: int) -> bool: + raise NotImplementedError('Implement me in subclass') + + def notify(self, + title: str, + body: str, + timeout: int = -1, + application: str = 'kitty', + icon: bool = True, + subtitle: Optional[str] = None, + urgency: Urgency = Urgency.Normal, + ) -> int: + raise NotImplementedError('Implement me in subclass') + + +class MacOSIntegration(DesktopIntegration): + + def initialize(self) -> None: + from .fast_data_types import cocoa_set_notification_activated_callback + self.id_counter = count() + cocoa_set_notification_activated_callback(self.notification_activated) + + def notify(self, + title: str, + body: str, + timeout: int = -1, + application: str = 'kitty', + icon: bool = True, + subtitle: Optional[str] = None, + urgency: Urgency = Urgency.Normal, + ) -> int: + desktop_notification_id = next(self.id_counter) + from .fast_data_types import cocoa_send_notification + cocoa_send_notification(str(desktop_notification_id), title, body, subtitle, urgency.value) + return desktop_notification_id + + def notification_activated(self, ident: str) -> None: + try: + desktop_notification_id = int(ident) + except Exception: + log_error(f'Got unexpected notification activated event with id: {ident!r} from cocoa') + else: + self.notification_manager.notification_activated(desktop_notification_id) + + +class FreeDesktopIntegration(DesktopIntegration): + + def close_notification(self, desktop_notification_id: int) -> bool: + from .fast_data_types import dbus_close_notification + return dbus_close_notification(desktop_notification_id) + + def dispatch_event_from_desktop(self, *args: Any) -> None: + event_type: str = args[0] + dbus_notification_id: int = args[1] + if event_type == 'created': + self.notification_manager.notification_created(dbus_notification_id) + elif event_type == 'activation_token': + token: str = args[2] + self.notification_manager.notification_activation_token_received(dbus_notification_id, token) + elif event_type == 'activated': + self.notification_manager.notification_activated(dbus_notification_id) + elif event_type == 'closed': + self.notification_manager.notification_closed(dbus_notification_id) + + def notify(self, + title: str, + body: str, + timeout: int = -1, + application: str = 'kitty', + icon: bool = True, + subtitle: Optional[str] = None, + urgency: Urgency = Urgency.Normal, + ) -> int: + icf = '' + if icon is True: + icf = get_custom_window_icon()[1] or logo_png_file + from .fast_data_types import dbus_send_notification + return dbus_send_notification(application, icf, title, body, 'Click to see changes', timeout, urgency.value) + + +class UIState(NamedTuple): + has_keyboard_focus: bool + is_visible: bool + + +class Channel: + + def window_for_id(self, channel_id: int) -> Optional[WindowType]: + boss = get_boss() + if channel_id: + return boss.window_id_map.get(channel_id) + return boss.active_window + + def ui_state(self, channel_id: int) -> UIState: + has_focus = is_visible = False + boss = get_boss() + if w := self.window_for_id(channel_id): + has_focus = w.is_active and w.os_window_id == current_focused_os_window_id() + # window is in the active OS window and the active tab and is visible in the tab layout + is_visible = w.os_window_id == current_focused_os_window_id() and w.tabref() is boss.active_tab and w.is_visible_in_layout + return UIState(has_focus, is_visible) + + def send(self, channel_id: int, osc_escape_code: str) -> bool: + if w := self.window_for_id(channel_id): + if not w.destroyed: + w.screen.send_escape_code_to_child(ESC_OSC, osc_escape_code) + return True + return False + + def focus(self, channel_id: int, activation_token: str) -> None: + boss = get_boss() + if w := self.window_for_id(channel_id): + boss.set_active_window(w, switch_os_window_if_needed=True, activation_token=activation_token) + + +def sanitize_text(x: str) -> str: + return sanitize_control_codes(x) + +@run_once +def sanitize_identifier_pat() -> 're.Pattern[str]': + return re.compile(r'[^a-zA-Z0-9-_+.]+') + + +def sanitize_id(v: str) -> str: + return sanitize_identifier_pat().sub('', v) + + +class Log: + def __call__(self, *a: Any, **kw: str) -> None: + log_error(*a, **kw) + + +class NotificationManager: + + def __init__( + self, + desktop_integration: Optional[DesktopIntegration] = None, + channel: Channel = Channel(), + log: Log = Log(), + ): + if desktop_integration is None: + self.desktop_integration = MacOSIntegration(self) if is_macos else FreeDesktopIntegration(self) + else: + self.desktop_integration = desktop_integration + self.channel = channel + self.log = log + self.reset() + + def reset(self) -> None: + self.new_version_notification_id = '' + self.in_progress_notification_commands: 'OrderedDict[int, NotificationCommand]' = OrderedDict() + self.in_progress_notification_commands_by_client_id: Dict[str, NotificationCommand] = {} + self.pending_commands: Dict[int, NotificationCommand] = {} + + def dispatch_event_from_desktop(self, *args: Any) -> None: + self.desktop_integration.dispatch_event_from_desktop(*args) + + def notification_created(self, desktop_notification_id: int) -> None: + if n := self.in_progress_notification_commands.get(desktop_notification_id): + n.created_by_desktop = True + + def notification_activation_token_received(self, desktop_notification_id: int, token: str) -> None: + if n := self.in_progress_notification_commands.get(desktop_notification_id): + n.activation_token = token + + def notification_activated(self, desktop_notification_id: int) -> None: + if n := self.in_progress_notification_commands.get(desktop_notification_id): + if n.focus_requested: + self.channel.focus(n.channel_id, n.activation_token) + if n.report_requested: + if n.identifier: + self.channel.send(n.channel_id, f'99;i={n.identifier}') + if self.new_version_notification_id and n.identifier == self.new_version_notification_id: + self.new_version_notification_id = '' + from .update_check import notification_activated + notification_activated() + + def notification_closed(self, desktop_notification_id: int) -> None: + if n := self.in_progress_notification_commands.get(desktop_notification_id): + self.purge_notification(n) + if n.close_response_requested: + self.send_closed_response(n.channel_id, n.identifier) + + def send_test_notification(self) -> None: + boss = get_boss() + if w := boss.active_window: + from time import monotonic + cmd = NotificationCommand() + now = monotonic() + cmd.title = f'Test {now}' + cmd.body = f'At: {now}' + self.notify_with_command(cmd, w.id) + + def send_new_version_notification(self, version: str) -> None: + from .short_uuid import uuid4 + cmd = NotificationCommand() + cmd.title = 'kitty update available!' + cmd.body = f'kitty version {version} released' + self.new_version_notification_id = cmd.identifier = uuid4() + self.notify_with_command(cmd, 0) + + def is_notification_allowed(self, cmd: NotificationCommand, channel_id: int) -> bool: + if cmd.only_when is not OnlyWhen.always and cmd.only_when is not OnlyWhen.unset: + ui_state = self.channel.ui_state(channel_id) + if ui_state.has_keyboard_focus: + return False + if cmd.only_when is OnlyWhen.invisible and ui_state.is_visible: + return False + return True + + def notify_with_command(self, cmd: NotificationCommand, channel_id: int) -> Optional[int]: + cmd.channel_id = channel_id + cmd.finalise() + title = cmd.title or cmd.body + body = cmd.body if cmd.title else '' + if not title or not self.is_notification_allowed(cmd, channel_id): + return None + urgency = Urgency.Normal if cmd.urgency is None else cmd.urgency + desktop_notification_id = self.desktop_integration.notify(title=sanitize_text(title), body=sanitize_text(body), urgency=urgency) + self.register_in_progress_notification(cmd, desktop_notification_id) + return desktop_notification_id + + def register_in_progress_notification(self, cmd: NotificationCommand, desktop_notification_id: int) -> None: + cmd.desktop_notification_id = desktop_notification_id + self.in_progress_notification_commands[desktop_notification_id] = cmd + if cmd.identifier: + self.in_progress_notification_commands_by_client_id[cmd.identifier] = cmd + if len(self.in_progress_notification_commands) > 128: + _, cmd = self.in_progress_notification_commands.popitem(False) + self.in_progress_notification_commands_by_client_id.pop(cmd.identifier, None) + + def parse_notification_cmd(self, prev_cmd: NotificationCommand, channel_id: int, raw: str) -> Optional[NotificationCommand]: + metadata, payload = raw.partition(';')[::2] + cmd = NotificationCommand() + try: + payload_type, payload_is_encoded = cmd.parse_metadata(metadata, prev_cmd) + except Exception: + self.log('Malformed metadata section in OSC 99: ' + metadata) + return None + if payload_type is PayloadType.query: + actions = ','.join(x.value for x in Action) + when = ','.join(x.value for x in OnlyWhen if x.value) + urgency = ','.join(str(x.value) for x in Urgency) + i = f'i={cmd.identifier or "0"}:' + p = ','.join(x.value for x in PayloadType if x.value) + self.channel.send(channel_id, f'99;{i}p=?;a={actions}:o={when}:u={urgency}:p={p}') + return None + if payload_type is PayloadType.close: + if payload_is_encoded: + from base64 import standard_b64decode + try: + payload = standard_b64decode(payload).decode('utf-8') + except Exception: + self.log('Malformed OSC 99 close command: payload is not base64 encoded UTF-8 text') + if cmd.identifier: + to_close = self.in_progress_notification_commands_by_client_id.get(cmd.identifier) + if to_close: + if payload == 'notify': + to_close.close_response_requested = True + if not self.desktop_integration.close_notification(to_close.desktop_notification_id): + if to_close.close_response_requested: + self.send_closed_response(to_close.channel_id, to_close.identifier) + self.purge_notification(to_close) + else: + self.send_closed_response(channel_id, cmd.identifier) + return None + + if payload_type is PayloadType.unknown: + self.log(f'OSC 99: unknown payload type: {payload_type}, ignoring payload') + payload = '' + + cmd.set_payload(payload_type, payload_is_encoded, payload, prev_cmd) + return cmd + + def send_closed_response(self, channel_id: int, client_id: str) -> None: + self.channel.send(channel_id, f'99;i={client_id}:p=close;') + + def purge_notification(self, cmd: NotificationCommand) -> None: + self.in_progress_notification_commands_by_client_id.pop(cmd.identifier, None) + self.in_progress_notification_commands.pop(cmd.desktop_notification_id, None) + + def handle_notification_cmd(self, channel_id: int, osc_code: int, raw: str) -> None: + if osc_code == 99: + cmd = self.pending_commands.pop(channel_id, None) or NotificationCommand() + q = self.parse_notification_cmd(cmd, channel_id, raw) + if q is not None: + if q.done: + self.notify_with_command(q, channel_id) + else: + self.pending_commands[channel_id] = q + elif osc_code == 9: + n = NotificationCommand() + n.title = raw + self.notify_with_command(n, channel_id) + elif osc_code == 777: + n = NotificationCommand() + parts = raw.split(';', 1) + n.title, n.body = parts[0], (parts[1] if len(parts) > 1 else '') + self.notify_with_command(n, channel_id) diff --git a/kitty/notify.py b/kitty/notify.py deleted file mode 100644 index 8270a7f04..000000000 --- a/kitty/notify.py +++ /dev/null @@ -1,328 +0,0 @@ -#!/usr/bin/env python -# License: GPLv3 Copyright: 2019, Kovid Goyal - -import re -from base64 import standard_b64decode -from collections import OrderedDict -from contextlib import suppress -from enum import Enum -from itertools import count -from typing import Callable, Dict, FrozenSet, Optional - -from .constants import is_macos, logo_png_file -from .fast_data_types import current_focused_os_window_id, get_boss -from .types import run_once -from .utils import get_custom_window_icon, log_error - - -class Urgency(Enum): - Low: int = 0 - Normal: int = 1 - Critical: int = 2 - - -if is_macos: - from .fast_data_types import cocoa_send_notification - - def notify( - title: str, - body: str, - timeout: int = 5000, - application: str = 'kitty', - icon: bool = True, - identifier: Optional[str] = None, - subtitle: Optional[str] = None, - urgency: Urgency = Urgency.Normal, - ) -> None: - cocoa_send_notification(identifier, title, body, subtitle, urgency.value) - -else: - - from .fast_data_types import dbus_send_notification - - alloc_map: Dict[int, str] = {} - identifier_map: Dict[str, int] = {} - - def dbus_notification_created(alloc_id: int, notification_id: int) -> None: - identifier = alloc_map.pop(alloc_id, None) - if identifier is not None: - identifier_map[identifier] = notification_id - - def dbus_notification_activated(notification_id: int, action: str) -> None: - rmap = {v: k for k, v in identifier_map.items()} - identifier = rmap.get(notification_id) - if identifier is not None: - notification_activated(identifier) - - def notify( - title: str, - body: str, - timeout: int = -1, - application: str = 'kitty', - icon: bool = True, - identifier: Optional[str] = None, - subtitle: Optional[str] = None, - urgency: Urgency = Urgency.Normal, - ) -> None: - icf = '' - if icon is True: - icf = get_custom_window_icon()[1] or logo_png_file - alloc_id = dbus_send_notification(application, icf, title, body, 'Click to see changes', timeout, urgency.value) - if alloc_id and identifier is not None: - alloc_map[alloc_id] = identifier - -class NotifyImplementation: - def __call__(self, title: str, body: str, identifier: str, urgency: Urgency = Urgency.Normal) -> None: - notify(title, body, identifier=identifier, urgency=urgency) - -notify_implementation = NotifyImplementation() - - -class PayloadType(Enum): - unknown = '' - title = 'title' - body = 'body' - query = '?' - close = 'close' - - -class OnlyWhen(Enum): - unset = '' - always = 'always' - unfocused = 'unfocused' - invisible = 'invisible' - - -class Action(Enum): - focus = 'focus' - report = 'report' - - -class NotificationCommand: - - done: bool = True - identifier: str = '0' - title: str = '' - body: str = '' - actions: FrozenSet[Action] = frozenset((Action.focus,)) - only_when: OnlyWhen = OnlyWhen.unset - urgency: Optional[Urgency] = None - - def __repr__(self) -> str: - return ( - f'NotificationCommand(identifier={self.identifier!r}, title={self.title!r}, body={self.body!r},' - f'actions={self.actions}, done={self.done!r}, urgency={self.urgency})') - - -def parse_osc_9(raw: str) -> NotificationCommand: - ans = NotificationCommand() - ans.title = raw - return ans - - -def parse_osc_777(raw: str) -> NotificationCommand: - parts = raw.split(';', 1) - ans = NotificationCommand() - ans.title = parts[0] - if len(parts) > 1: - ans.body = parts[1] - return ans - - -@run_once -def sanitize_identifier_pat() -> 're.Pattern[str]': - return re.compile(r'[^a-zA-Z0-9-_+.]+') - - -def sanitize_id(v: str) -> str: - return sanitize_identifier_pat().sub('', v) - - -class QueryResponse(Exception): - - def __init__(self, response_string: str): - super().__init__('not an error') - self.response_string = response_string - - -def parse_osc_99(raw: str, log_warnings: bool = True) -> NotificationCommand: - cmd = NotificationCommand() - metadata, payload = raw.partition(';')[::2] - payload_is_encoded = False - payload_type = PayloadType.title - if metadata: - for part in metadata.split(':'): - try: - k, v = part.split('=', 1) - except Exception: - if log_warnings: - log_error('Malformed OSC 99: metadata is not key=value pairs') - return cmd - if k == 'p': - try: - payload_type = PayloadType(v) - except ValueError: - payload_type = PayloadType.unknown - elif k == 'i': - cmd.identifier = sanitize_id(v) - elif k == 'e': - payload_is_encoded = v == '1' - elif k == 'd': - cmd.done = v != '0' - elif k == 'a': - for ax in v.split(','): - if remove := ax.startswith('-'): - ax = ax.lstrip('+-') - try: - ac = Action(ax) - except ValueError: - pass - else: - if remove: - cmd.actions -= {ac} - else: - cmd.actions = cmd.actions.union({ac}) - elif k == 'o': - with suppress(ValueError): - cmd.only_when = OnlyWhen(v) - elif k == 'u': - with suppress(Exception): - cmd.urgency = Urgency(int(v)) - if payload_type is PayloadType.query: - actions = ','.join(x.value for x in Action) - when = ','.join(x.value for x in OnlyWhen if x.value) - urgency = ','.join(str(x.value) for x in Urgency) - i = f'i={sanitize_id(cmd.identifier or "0")}:' - p = ','.join(x.value for x in PayloadType if x.value) - raise QueryResponse(f'99;{i}p=?;a={actions}:o={when}:u={urgency}:p={p}') - - if payload_type in (PayloadType.title, PayloadType.body): - if payload_is_encoded: - try: - payload = standard_b64decode(payload).decode('utf-8') - except Exception: - if log_warnings: - log_error('Malformed OSC 99: payload is not base64 encoded UTF-8 text') - return NotificationCommand() - if payload_type is PayloadType.title: - cmd.title = payload - else: - cmd.body = payload - else: - if log_warnings: - log_error(f'OSC 99: unknown payload type: {payload_type}, ignoring payload') - return cmd - - -def limit_size(x: str) -> str: - if len(x) > 1024: - x = x[:1024] - return x - - -def merge_osc_99(prev: NotificationCommand, cmd: NotificationCommand) -> NotificationCommand: - if prev.done or prev.identifier != cmd.identifier: - return cmd - cmd.actions = prev.actions.union(cmd.actions) - cmd.title = limit_size(prev.title + cmd.title) - cmd.body = limit_size(prev.body + cmd.body) - if cmd.only_when is OnlyWhen.unset: - cmd.only_when = prev.only_when - if cmd.urgency is None: - cmd.urgency = prev.urgency - return cmd - - -identifier_registry: "OrderedDict[str, RegisteredNotification]" = OrderedDict() -id_counter = count() - - -class RegisteredNotification: - identifier: str - window_id: int - focus: bool = False - report: bool = False - - def __init__(self, cmd: NotificationCommand, window_id: int): - self.window_id = window_id - for x in cmd.actions: - if x is Action.focus: - self.focus = True - elif x is Action.report: - self.report = True - self.identifier = cmd.identifier - - -def register_identifier(identifier: str, cmd: NotificationCommand, window_id: int) -> None: - identifier_registry[identifier] = RegisteredNotification(cmd, window_id) - if len(identifier_registry) > 100: - identifier_registry.popitem(False) - - -def notification_activated(identifier: str, activated_implementation: Optional[Callable[[str, int, bool, bool], None]] = None) -> None: - if identifier == 'new-version': - from .update_check import notification_activated as do - do() - elif identifier.startswith('test-notify-'): - log_error(f'Test notification {identifier} activated') - else: - r = identifier_registry.pop(identifier, None) - if r is not None and (r.focus or r.report): - if activated_implementation is None: - get_boss().notification_activated(r.identifier, r.window_id, r.focus, r.report) - else: - activated_implementation(r.identifier, r.window_id, r.focus, r.report) - - -def reset_registry() -> None: - global id_counter - identifier_registry.clear() - id_counter = count() - - -def notify_with_command(cmd: NotificationCommand, window_id: int, notify_implementation: NotifyImplementation = notify_implementation) -> None: - title = cmd.title or cmd.body - body = cmd.body if cmd.title else '' - if not title: - return - urgency = Urgency.Normal if cmd.urgency is None else cmd.urgency - if cmd.only_when is not OnlyWhen.always and cmd.only_when is not OnlyWhen.unset: - w = get_boss().window_id_map.get(window_id) - if w is None: - return - boss = get_boss() - window_has_keyboard_focus = w.is_active and w.os_window_id == current_focused_os_window_id() - if window_has_keyboard_focus: - return - if cmd.only_when is OnlyWhen.invisible: - if w.os_window_id == current_focused_os_window_id() and w.tabref() is boss.active_tab and w.is_visible_in_layout: - return # window is in the active OS window and the active tab and is visible in the tab layout - if title: - identifier = f'i{next(id_counter)}' - notify_implementation(title, body, identifier, urgency) - register_identifier(identifier, cmd, window_id) - - -def handle_notification_cmd( - osc_code: int, - raw_data: str, - window_id: int, - prev_cmd: NotificationCommand, - notify_implementation: NotifyImplementation = notify_implementation, - log_warnings: bool = True, -) -> Optional[NotificationCommand]: - if osc_code == 99: - cmd = merge_osc_99(prev_cmd, parse_osc_99(raw_data, log_warnings=log_warnings)) - if cmd.done: - notify_with_command(cmd, window_id, notify_implementation) - cmd = NotificationCommand() - return cmd - if osc_code == 9: - cmd = parse_osc_9(raw_data) - notify_with_command(cmd, window_id, notify_implementation) - return cmd - if osc_code == 777: - cmd = parse_osc_777(raw_data) - notify_with_command(cmd, window_id, notify_implementation) - return cmd - return None diff --git a/kitty/update_check.py b/kitty/update_check.py index 379f4c712..b3386a07b 100644 --- a/kitty/update_check.py +++ b/kitty/update_check.py @@ -11,7 +11,6 @@ from urllib.request import urlopen from .config import atomic_save from .constants import Version, cache_dir, clear_handled_signals, kitty_exe, version, website_url from .fast_data_types import add_timer, get_boss, monitor_pid -from .notify import notify from .utils import log_error, open_url CHANGELOG_URL = website_url('changelog') @@ -37,11 +36,7 @@ def version_notification_log() -> str: def notify_new_version(release_version: Version) -> None: - notify( - 'kitty update available!', - 'kitty version {} released'.format('.'.join(map(str, release_version))), - identifier='new-version', - ) + get_boss().notification_manager.send_new_version_notification('.'.join(map(str, release_version))) def get_released_version() -> str: diff --git a/kitty/window.py b/kitty/window.py index 4e2174926..7e3bb89dd 100644 --- a/kitty/window.py +++ b/kitty/window.py @@ -90,16 +90,6 @@ from .fast_data_types import ( wakeup_main_loop, ) from .keys import keyboard_mode_name, mod_mask -from .notify import ( - NotificationCommand, - NotifyImplementation, - OnlyWhen, - QueryResponse, - Urgency, - handle_notification_cmd, - notify_with_command, - sanitize_identifier_pat, -) from .rgb import to_color from .terminfo import get_capabilities from .types import MouseEvent, OverlayType, WindowGeometry, ac, run_once @@ -611,7 +601,6 @@ class Window: self.current_remote_data: List[str] = [] self.current_mouse_event_button = 0 self.current_clipboard_read_ask: Optional[bool] = None - self.prev_osc99_cmd = NotificationCommand() self.last_cmd_output_start_time = 0. self.open_url_handler: 'OpenUrlHandler' = None self.last_cmd_cmdline = '' @@ -1078,12 +1067,7 @@ class Window: log_error(f'Ignoring unknown OSC 777: {raw_data}') return # unknown OSC 777 raw_data = raw_data[len('notify;'):] - try: - cmd = handle_notification_cmd(osc_code, raw_data, self.id, self.prev_osc99_cmd) - if cmd is not None and osc_code == 99: - self.prev_osc99_cmd = cmd - except QueryResponse as err: - self.screen.send_escape_code_to_child(ESC_OSC, err.response_string) + get_boss().notification_manager.handle_notification_cmd(self.id, osc_code, raw_data) def on_mouse_event(self, event: Dict[str, Any]) -> bool: event['mods'] = event.get('mods', 0) & mod_mask @@ -1244,10 +1228,6 @@ class Window: b |= b << 8 self.screen.send_escape_code_to_child(ESC_OSC, f'{code};rgb:{r:04x}/{g:04x}/{b:04x}') - def report_notification_activated(self, identifier: str) -> None: - identifier = sanitize_identifier_pat().sub('', identifier) - self.screen.send_escape_code_to_child(ESC_OSC, f'99;i={identifier};') - def notify_child_of_resize(self) -> None: pty_size = self.last_reported_pty_size if pty_size[0] > -1 and self.screen.in_band_resize_notification: @@ -1492,29 +1472,21 @@ class Window: when, duration, action, notify_cmdline = opts.notify_on_cmd_finish if last_cmd_output_duration >= duration and when != 'never': + from .notifications import NotificationCommand, OnlyWhen cmd = NotificationCommand() cmd.title = 'kitty' s = self.last_cmd_cmdline.replace('\\\n', ' ') cmd.body = f'Command {s} finished with status: {exit_status}.\nClick to focus.' cmd.only_when = OnlyWhen(when) + nm = get_boss().notification_manager + if not nm.is_notification_allowed(cmd, self.id): + return if action == 'notify': - notify_with_command(cmd, self.id) + nm.notify_with_command(cmd, self.id) elif action == 'bell': - class Bell(NotifyImplementation): - def __init__(self, window_id: int): - self.window_id = window_id - def __call__(self, title: str, body: str, identifier: str, urgency: Urgency = Urgency.Normal) -> None: - w = get_boss().window_id_map.get(self.window_id) - if w: - w.screen.bell() - notify_with_command(cmd, self.id, notify_implementation=Bell(self.id)) + self.screen.bell() elif action == 'command': - class Run(NotifyImplementation): - def __init__(self, last_cmd_cmdline: str): - self.last_cmd_cmdline = last_cmd_cmdline - def __call__(self, title: str, body: str, identifier: str, urgency: Urgency = Urgency.Normal) -> None: - open_cmd([x.replace('%c', self.last_cmd_cmdline).replace('%s', exit_status) for x in notify_cmdline]) - notify_with_command(cmd, self.id, notify_implementation=Run(self.last_cmd_cmdline)) + open_cmd([x.replace('%c', self.last_cmd_cmdline).replace('%s', exit_status) for x in notify_cmdline]) else: raise ValueError(f'Unknown action in option `notify_on_cmd_finish`: {action}') diff --git a/kitty_tests/notifications.py b/kitty_tests/notifications.py new file mode 100644 index 000000000..38e62990a --- /dev/null +++ b/kitty_tests/notifications.py @@ -0,0 +1,164 @@ +#!/usr/bin/env python +# License: GPLv3 Copyright: 2024, Kovid Goyal + + +from base64 import standard_b64encode +from typing import Optional + +from kitty.notifications import Channel, DesktopIntegration, NotificationManager, UIState, Urgency + +from . import BaseTest + + +def n(title='title', body='', urgency=Urgency.Normal, desktop_notification_id=1): + return {'title': title, 'body': body, 'urgency': urgency, 'id': desktop_notification_id} + + +class DesktopIntegration(DesktopIntegration): + def initialize(self): + self.reset() + + def reset(self): + self.notifications = [] + self.close_events = [] + self.counter = 0 + + def close_notification(self, desktop_notification_id: int) -> bool: + self.close_events.append(desktop_notification_id) + + def notify(self, + title: str, + body: str, + timeout: int = -1, + application: str = 'kitty', + icon: bool = True, + subtitle: Optional[str] = None, + urgency: Urgency = Urgency.Normal, + ) -> int: + self.counter += 1 + self.notifications.append(n(title, body, urgency, self.counter)) + return self.counter + + +class Channel(Channel): + + focused = visible = True + + def __init__(self, *a): + super().__init__(*a) + self.reset() + + def reset(self): + self.responses = [] + self.focus_events = [] + + def ui_state(self, channel_id): + return UIState(self.focused, self.visible) + + def focus(self, channel_id: int, activation_token: str) -> None: + self.focus_events.append(activation_token) + + def send(self, channel_id: int, osc_escape_code: str) -> bool: + self.responses.append(osc_escape_code) + + +def do_test(self: 'TestNotifications') -> None: + di = DesktopIntegration(None) + ch = Channel() + nm = NotificationManager(di, ch, lambda *a, **kw: None) + di.notification_manager = nm + + def reset(): + di.reset() + ch.reset() + nm.reset() + + def h(raw_data, osc_code=99, channel_id=1): + nm.handle_notification_cmd(channel_id, osc_code, raw_data) + + def activate(which=0): + n = di.notifications[which] + nm.notification_activated(n['id']) + + h('test it', osc_code=9) + self.ae(di.notifications, [n(title='test it')]) + activate() + assert_events() + reset() + + h('d=0:u=2:i=x;title') + h('d=1:i=x:p=body;body') + self.ae(notifications, [n(client_id='x', body='body', urgency=Urgency.Critical)]) + activate() + assert_events('x') + reset() + + h('i=x:p=body:a=-focus;body') + self.ae(notifications, [n(client_id='x', title='body')]) + activate() + assert_events('x', focus=False) + reset() + + h('i=x:e=1;' + standard_b64encode(b'title').decode('ascii')) + self.ae(notifications, [n(client_id='x', )]) + activate() + assert_events('x') + reset() + + h('e=1;' + standard_b64encode(b'title').decode('ascii')) + self.ae(notifications, [n()]) + activate() + assert_events() + reset() + + h('d=0:i=x:a=-report;title') + h('d=1:i=x:a=report;body') + self.ae(notifications, [n(client_id='x', title='titlebody')]) + activate() + assert_events('x', report=True) + reset() + + h('d=0:i=y;title') + h('d=1:i=y:p=xxx;title') + self.ae(notifications, [n(client_id='y')]) + reset() + + # test closing interactions with reporting and activation + h('i=c;title') + self.ae(notifications, [n(client_id='c')]) + close() + assert_events('c', focus=False, close=True) + reset() + h('i=c;title') + self.ae(notifications, [n(client_id='c')]) + h('i=c:p=close') + self.ae(notifications, [n(client_id='c')]) + assert_events('c', focus=False, close=True) + reset() + h('i=c;title') + h('i=c:p=close;notify') + assert_events('c', focus=False, close=True, close_response=True) + reset() + + h(';title') + self.ae(notifications, [n()]) + activate() + assert_events() + reset() + + # Test querying + h('i=xyz:p=?') + self.assertFalse(notifications) + qr = 'a=focus,report:o=always,unfocused,invisible:u=0,1,2:p=title,body,?,close' + self.ae(query_responses, [f'99;i=xyz:p=?;{qr}']) + reset() + h('p=?') + self.assertFalse(notifications) + self.ae(query_responses, [f'99;i=0:p=?;{qr}']) + + +class TestNotifications(BaseTest): + + def test_desktop_notify(self): + do_test(self) + diff --git a/kitty_tests/parser.py b/kitty_tests/parser.py index c0a15dd11..3b581e0dc 100644 --- a/kitty_tests/parser.py +++ b/kitty_tests/parser.py @@ -1,7 +1,6 @@ #!/usr/bin/env python # License: GPL v3 Copyright: 2016, Kovid Goyal -from base64 import standard_b64encode from binascii import hexlify from functools import partial @@ -15,7 +14,6 @@ from kitty.fast_data_types import ( test_find_either_of_two_bytes, test_utf8_decode_to_sentinel, ) -from kitty.notify import NotificationCommand, QueryResponse, Urgency, handle_notification_cmd, notification_activated, reset_registry from . import BaseTest, parse_bytes @@ -521,92 +519,6 @@ class TestParser(BaseTest): c.clear() pb('\033]52;p;xyz\x07', ('clipboard_control', 52, 'p;xyz')) - def test_desktop_notify(self): - reset_registry() - notifications = [] - activations = [] - query_responses = [] - prev_cmd = NotificationCommand() - - def reset(): - nonlocal prev_cmd - reset_registry() - del notifications[:] - del activations[:] - del query_responses[:] - prev_cmd = NotificationCommand() - - def notify(title, body, identifier, urgency=Urgency.Normal): - notifications.append((title, body, identifier, urgency)) - - def h(raw_data, osc_code=99, window_id=1): - nonlocal prev_cmd - try: - x = handle_notification_cmd(osc_code, raw_data, window_id, prev_cmd, notify, log_warnings=False) - if x is not None and osc_code == 99: - prev_cmd = x - except QueryResponse as err: - query_responses.append(err.response_string) - - def activated(identifier, window_id, focus, report): - activations.append((identifier, window_id, focus, report)) - - h('test it', osc_code=9) - self.ae(notifications, [('test it', '', 'i0', Urgency.Normal)]) - notification_activated(notifications[-1][-2], activated) - self.ae(activations, [('0', 1, True, False)]) - reset() - - h('d=0:u=2:i=x;title') - h('d=1:i=x:p=body;body') - self.ae(notifications, [('title', 'body', 'i0', Urgency.Critical)]) - notification_activated(notifications[-1][-2], activated) - self.ae(activations, [('x', 1, True, False)]) - reset() - - h('i=x:p=body:a=-focus;body') - self.ae(notifications, [('body', '', 'i0', Urgency.Normal)]) - notification_activated(notifications[-1][-2], activated) - self.ae(activations, []) - reset() - - h('i=x:e=1;' + standard_b64encode(b'title').decode('ascii')) - self.ae(notifications, [('title', '', 'i0', Urgency.Normal)]) - notification_activated(notifications[-1][-2], activated) - self.ae(activations, [('x', 1, True, False)]) - reset() - - h('e=1;' + standard_b64encode(b'title').decode('ascii')) - self.ae(notifications, [('title', '', 'i0', Urgency.Normal)]) - notification_activated(notifications[-1][-2], activated) - self.ae(activations, [('0', 1, True, False)]) - reset() - - h('d=0:i=x:a=-report;title') - h('d=1:i=x:a=report;body') - self.ae(notifications, [('titlebody', '', 'i0', Urgency.Normal)]) - notification_activated(notifications[-1][-2], activated) - self.ae(activations, [('x', 1, True, True)]) - reset() - - h('d=0:i=y;title') - h('d=1:i=y:p=xxx;title') - self.ae(notifications, [('title', '', 'i0', Urgency.Normal)]) - reset() - - h(';title') - self.ae(notifications, [('title', '', 'i0', Urgency.Normal)]) - notification_activated(notifications[-1][-2], activated) - self.ae(activations, [('0', 1, True, False)]) - reset() - h('i=xyz:p=?') - self.assertFalse(notifications) - self.ae(query_responses, ['99;i=xyz:p=?;a=focus,report:o=always,unfocused,invisible:u=0,1,2:p=title,body,?,close']) - reset() - h('p=?') - self.assertFalse(notifications) - self.ae(query_responses, ['99;i=0:p=?;a=focus,report:o=always,unfocused,invisible:u=0,1,2:p=title,body,?,close']) - def test_dcs_codes(self): s = self.create_screen() c = s.callbacks