CERBERUS/kitty
1
0
Fork 0
mirror of https://github.com/kovidgoyal/kitty.git synced 2026-09-05 17:28:36 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Fix notification close mode

Browse source
This commit is contained in:
Hassan Raza 2026-08-28 11:20:05 +05:00
parent da90724f54
commit 99fb1b8aca
No known key found for this signature in database
2 changed files with 73 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

56
kittens/notify/main.go
View file

@ -40,7 +40,6 @@ type parsed_data struct {
expire_time time.Duration
title, body, identifier string
image_data []byte
initial_msg string
}
func (p *parsed_data) create_metadata() string {
@ -86,6 +85,10 @@ func (p *parsed_data) generate_chunks(callback func(string)) {
write_chunk := func(middle string) {
callback(prefix + middle + ESC_CODE_SUFFIX)
}
if p.title == "" {
write_chunk(":p=close;")
return
}
add_payload := func(payload_type, payload string) {
if payload == "" {
@ -127,10 +130,11 @@ func (p *parsed_data) run_loop() (err error) {
})
}
lp.OnInitialize = func() (string, error) {
if p.initial_msg != "" {
return p.initial_msg, nil
}
p.generate_chunks(func(x string) { lp.QueueWriteString(x) })
if p.title == "" {
// Closing an existing notification does not guarantee a close response.
poll_for_close()
}
return "", nil
}
lp.OnEscapeCode = func(ect loop.EscapeCodeType, data []byte) error {
@ -262,17 +266,19 @@ func (p *parsed_data) load_image_data() (err error) {
}
func main(_ *cli.Command, opts *Options, args []string) (rc int, err error) {
if len(args) == 0 {
return 1, fmt.Errorf("Must specify a TITLE for the notification")
}
var p parsed_data
p.opts = opts
p.title = args[0]
if len(args) > 0 {
p.title = args[0]
}
if len(args) > 1 {
p.body = strings.Join(args[1:], " ")
}
ident := opts.Identifier
if ident == "" {
if p.title == "" {
return 1, fmt.Errorf("Must specify a non-empty TITLE for the notification or specify an identifier to close a notification.")
}
if ident, err = random_ident(); err != nil {
return 1, fmt.Errorf("Failed to generate a random identifier with error: %w", err)
}
@ -287,34 +293,14 @@ func main(_ *cli.Command, opts *Options, args []string) (rc int, err error) {
if !check_id_valid(opts.IconCacheId) {
return 1, bad_ident(opts.IconCacheId)
}
if len(p.title) == 0 {
if ident == "" {
return 1, fmt.Errorf("Must specify a non-empty TITLE for the notification or specify an identifier to close a notification.")
}
msg := ESC_CODE_PREFIX + "i=" + ident + ":p=close;" + ESC_CODE_SUFFIX
if opts.OnlyPrintEscapeCode {
_, err = os.Stdout.WriteString(msg)
} else if p.wait_till_closed {
p.initial_msg = msg
err = p.run_loop()
} else {
var term *tty.Term
if term, err = tty.OpenControllingTerm(); err != nil {
return 1, fmt.Errorf("Failed to open controlling terminal with error: %w", err)
}
if _, err = term.WriteString(msg); err != nil {
term.RestoreAndClose()
return 1, err
}
term.RestoreAndClose()
}
}
if p.expire_time, err = parse_duration(opts.ExpireAfter); err != nil {
return 1, fmt.Errorf("Invalid expire time: %s with error: %w", opts.ExpireAfter, err)
}
p.wait_till_closed = opts.WaitTillClosed
if err = p.load_image_data(); err != nil {
return 1, fmt.Errorf("Failed to load image data from %s with error %w", opts.IconPath, err)
if p.title != "" {
if p.expire_time, err = parse_duration(opts.ExpireAfter); err != nil {
return 1, fmt.Errorf("Invalid expire time: %s with error: %w", opts.ExpireAfter, err)
}
if err = p.load_image_data(); err != nil {
return 1, fmt.Errorf("Failed to load image data from %s with error %w", opts.IconPath, err)
}
}
if opts.OnlyPrintEscapeCode {
p.generate_chunks(func(x string) {

52
kitty_tests/notify.py Normal file
View file

@ -0,0 +1,52 @@
#!/usr/bin/env python
# License: GPLv3 Copyright: 2026, Kovid Goyal <kovid at kovidgoyal.net>
import subprocess
from kitty.constants import kitten_exe
from .base import PTY, BaseTest
class TestNotify(BaseTest):
def test_notify_close_escape_code(self):
for title in ([], ['']):
for wait in ([], ['--wait-till-closed']):
with self.subTest(title=title, wait=wait):
p = subprocess.run(
[kitten_exe(), 'notify', '--only-print-escape-code', '--identifier=test', *wait, *title],
capture_output=True,
timeout=10,
)
self.ae(p.returncode, 0, p.stderr)
self.ae(p.stdout, b'\x1b]99;i=test:p=close;\x1b\\')
def test_notify_close_requires_identifier(self):
for args, error in (([], b'non-empty TITLE'), ([''], b'non-empty TITLE'), (['--identifier=invalid:id'], b'Invalid identifier')):
with self.subTest(args=args):
p = subprocess.run([kitten_exe(), 'notify', '--only-print-escape-code', *args], capture_output=True, timeout=10)
self.ae(p.returncode, 1)
self.ae(p.stdout, b'')
self.assertIn(error, p.stderr)
def test_notify_close_terminal(self):
with PTY([kitten_exe(), 'notify', '--identifier=test']) as pty:
pty.wait_till(lambda: bool(pty.callbacks.notifications))
pty.wait_till_child_exits(require_exit_code=0)
self.ae(pty.callbacks.notifications, [(99, 'i=test:p=close;')])
def test_notify_close_wait(self):
for title, response in (([], 'alive'), ([''], 'close')):
with self.subTest(title=title), PTY([kitten_exe(), 'notify', '--identifier=test', '--wait-till-closed', *title]) as pty:
messages = pty.callbacks.notifications
pty.wait_till(lambda: bool(messages))
self.ae(messages[0], (99, 'i=test:p=close;'))
pty.wait_till(lambda: (99, 'i=test:p=alive;') in messages)
self.ae(messages, [(99, 'i=test:p=close;'), (99, 'i=test:p=alive;')])
messages.clear()
pty.write_to_child('\x1b]99;i=test:p=alive;test\x1b\\')
pty.wait_till(lambda: bool(messages))
self.ae(messages, [(99, 'i=test:p=alive;')])
pty.write_to_child(f'\x1b]99;i=test:p={response};\x1b\\')
pty.wait_till_child_exits(require_exit_code=0)
self.assertTrue(all(m == (99, 'i=test:p=alive;') for m in messages))