Port the tests
This commit is contained in:
Kovid Goyal
parent
7dc30c7897
commit
095e1917c1
2 changed files with 93 additions and 58 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@
|
|||
# License: GPLv3 Copyright: 2024, Kovid Goyal <kovid at kovidgoyal.net>
|
||||
|
||||
|
||||
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):
|
||||
|
|
|
|||
Loading…
Reference in a new issue