From 095e1917c143210fa734dc7a4f17b53f351a8fb2 Mon Sep 17 00:00:00 2001 From: Kovid Goyal Date: Thu, 25 Jul 2024 07:39:29 +0530 Subject: [PATCH] Port the tests --- kitty/notifications.py | 74 +++++++++++++++++----------------- kitty_tests/notifications.py | 77 ++++++++++++++++++++++++++---------- 2 files changed, 93 insertions(+), 58 deletions(-) diff --git a/kitty/notifications.py b/kitty/notifications.py index 91e6a63bc..dd5d36b8b 100644 --- a/kitty/notifications.py +++ b/kitty/notifications.py @@ -161,38 +161,39 @@ def __repr__(self) -> str: 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('+-') + if metadata: + for part in metadata.split(':'): + k, v = part.split('=', 1) + if k == 'p': try: - ac = Action(ax) + payload_type = PayloadType(v) except ValueError: - pass - else: - if remove: - self.actions -= {ac} + 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: - 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 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 @@ -266,6 +267,10 @@ def notify(self, ) -> int: raise NotImplementedError('Implement me in subclass') + def on_new_version_notification_activation(self, cmd: NotificationCommand) -> None: + from .update_check import notification_activated + notification_activated() + class MacOSIntegration(DesktopIntegration): @@ -367,8 +372,7 @@ def focus(self, channel_id: int, activation_token: str) -> None: 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) +sanitize_text = sanitize_control_codes @run_once def sanitize_identifier_pat() -> 're.Pattern[str]': @@ -423,7 +427,7 @@ def notification_activated(self, desktop_notification_id: int) -> None: 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}') + self.channel.send(n.channel_id, f'99;i={n.identifier};') if n.on_activation: try: n.on_activation(n) @@ -451,13 +455,9 @@ def send_new_version_notification(self, version: str) -> None: cmd = NotificationCommand() cmd.title = 'kitty update available!' cmd.body = f'kitty version {version} released' - cmd.on_activation = self.on_new_version_notification_activation + cmd.on_activation = self.desktop_integration.on_new_version_notification_activation self.notify_with_command(cmd, 0) - def on_new_version_notification_activation(self, cmd: NotificationCommand) -> None: - from .update_check import notification_activated - notification_activated() - 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) diff --git a/kitty_tests/notifications.py b/kitty_tests/notifications.py index 38e62990a..2b4784f8f 100644 --- a/kitty_tests/notifications.py +++ b/kitty_tests/notifications.py @@ -2,6 +2,7 @@ # License: GPLv3 Copyright: 2024, Kovid Goyal +import re from base64 import standard_b64encode from typing import Optional @@ -21,10 +22,18 @@ def initialize(self): def reset(self): self.notifications = [] self.close_events = [] + self.new_version_activated = False + self.close_succeeds = True self.counter = 0 + def on_new_version_notification_activation(self, cmd) -> None: + self.new_version_activated = True + def close_notification(self, desktop_notification_id: int) -> bool: self.close_events.append(desktop_notification_id) + if self.close_succeeds: + self.notification_manager.notification_closed(desktop_notification_id) + return self.close_succeeds def notify(self, title: str, @@ -80,6 +89,26 @@ def activate(which=0): n = di.notifications[which] nm.notification_activated(n['id']) + def close(which=0): + n = di.notifications[which] + di.close_notification(n['id']) + + def assert_events(focus=True, close=0, report='', close_response=''): + self.ae(ch.focus_events, [''] if focus else []) + if report: + self.assertIn(f'99;i={report};', ch.responses) + else: + for r in ch.responses: + m = re.match(r'99;i=[a-z0-9]+;', r) + self.assertIsNone(m, f'Unexpectedly found report response: {r}') + if close_response: + self.assertIn(f'99;i={close_response}:p=close;', ch.responses) + else: + for r in ch.responses: + m = re.match(r'99;i=[a-z0-9]+:p=close;', r) + self.assertIsNone(m, f'Unexpectedly found close response: {r}') + self.ae(di.close_events, [close] if close else []) + h('test it', osc_code=9) self.ae(di.notifications, [n(title='test it')]) activate() @@ -88,73 +117,79 @@ def activate(which=0): 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)]) + self.ae(di.notifications, [n(body='body', urgency=Urgency.Critical)]) activate() - assert_events('x') + assert_events() reset() h('i=x:p=body:a=-focus;body') - self.ae(notifications, [n(client_id='x', title='body')]) + self.ae(di.notifications, [n(title='body')]) activate() - assert_events('x', focus=False) + assert_events(focus=False) + reset() + + nm.send_new_version_notification('moose') + self.ae(di.notifications, [n('kitty update available!', 'kitty version moose released')]) + activate() + self.assertTrue(di.new_version_activated) reset() h('i=x:e=1;' + standard_b64encode(b'title').decode('ascii')) - self.ae(notifications, [n(client_id='x', )]) + self.ae(di.notifications, [n()]) activate() - assert_events('x') + assert_events() reset() h('e=1;' + standard_b64encode(b'title').decode('ascii')) - self.ae(notifications, [n()]) + self.ae(di.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')]) + self.ae(di.notifications, [n(title='titlebody')]) activate() - assert_events('x', report=True) + assert_events(report='x') reset() h('d=0:i=y;title') h('d=1:i=y:p=xxx;title') - self.ae(notifications, [n(client_id='y')]) + self.ae(di.notifications, [n()]) reset() # test closing interactions with reporting and activation h('i=c;title') - self.ae(notifications, [n(client_id='c')]) + self.ae(di.notifications, [n()]) close() - assert_events('c', focus=False, close=True) + assert_events(focus=False, close=True) reset() h('i=c;title') - self.ae(notifications, [n(client_id='c')]) + self.ae(di.notifications, [n()]) h('i=c:p=close') - self.ae(notifications, [n(client_id='c')]) - assert_events('c', focus=False, close=True) + self.ae(di.notifications, [n()]) + assert_events(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) + assert_events(focus=False, close=True, close_response='c') reset() h(';title') - self.ae(notifications, [n()]) + self.ae(di.notifications, [n()]) activate() assert_events() reset() # Test querying h('i=xyz:p=?') - self.assertFalse(notifications) + self.assertFalse(di.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}']) + self.ae(ch.responses, [f'99;i=xyz:p=?;{qr}']) reset() h('p=?') - self.assertFalse(notifications) - self.ae(query_responses, [f'99;i=0:p=?;{qr}']) + self.assertFalse(di.notifications) + self.ae(ch.responses, [f'99;i=0:p=?;{qr}']) class TestNotifications(BaseTest):