misc parser and test fixes

This commit is contained in:
Kovid Goyal
2023-10-29 14:39:36 +05:30
parent 5168e0b576
commit 52025ff030
10 changed files with 55 additions and 76 deletions

View File

@@ -37,21 +37,21 @@ class Callbacks:
self.set_pointer_shape = lambda data: None
def write(self, data) -> None:
self.wtcbuf += data
self.wtcbuf += bytes(data)
def title_changed(self, data, is_base64=False) -> None:
self.titlebuf.append(process_title_from_child(data, is_base64))
self.titlebuf.append(process_title_from_child(data, is_base64, ''))
def icon_changed(self, data) -> None:
self.iconbuf += data
self.iconbuf += str(data, 'utf-8')
def set_dynamic_color(self, code, data) -> None:
def set_dynamic_color(self, code, data='') -> None:
if code == 22:
self.set_pointer_shape(data)
else:
self.colorbuf += data or ''
self.colorbuf += str(data or b'', 'utf-8')
def set_color_table_color(self, code, data) -> None:
def set_color_table_color(self, code, data='') -> None:
self.ctbuf += ''
def color_profile_popped(self, x) -> None:
@@ -65,23 +65,19 @@ class Callbacks:
for c in get_capabilities(q, None):
self.write(c.encode('ascii'))
def use_utf8(self, on) -> None:
self.iutf8 = on
def desktop_notify(self, osc_code: int, raw_data: str) -> None:
self.notifications.append((osc_code, raw_data))
def desktop_notify(self, osc_code: int, raw_data: memoryview) -> None:
self.notifications.append((osc_code, str(raw_data, 'utf-8')))
def open_url(self, url: str, hyperlink_id: int) -> None:
self.open_urls.append((url, hyperlink_id))
def clipboard_control(self, data: str, is_partial: bool = False) -> None:
self.cc_buf.append((data, is_partial))
def clipboard_control(self, data: memoryview, is_partial: bool = False) -> None:
self.cc_buf.append((str(data, 'utf-8'), is_partial))
def clear(self) -> None:
self.wtcbuf = b''
self.iconbuf = self.colorbuf = self.ctbuf = ''
self.titlebuf = []
self.iutf8 = True
self.notifications = []
self.open_urls = []
self.cc_buf = []
@@ -112,6 +108,7 @@ class Callbacks:
print(text, file=sys.__stdout__, end='', flush=True)
def handle_remote_clone(self, msg):
msg = str(msg, 'utf-8')
if not msg:
if self.current_clone_data:
cdata, self.current_clone_data = self.current_clone_data, ''

View File

@@ -9,13 +9,19 @@ from functools import partial
from kitty.fast_data_types import CURSOR_BLOCK, base64_decode, base64_encode
from kitty.notify import NotificationCommand, handle_notification_cmd, notification_activated, reset_registry
from . import BaseTest
from . import BaseTest, parse_bytes
def cnv(x):
if isinstance(x, memoryview):
x = str(x, 'utf-8')
return x
class CmdDump(list):
def __call__(self, *a):
self.append(a)
self.append(tuple(map(cnv, a)))
class TestParser(BaseTest):
@@ -24,7 +30,7 @@ class TestParser(BaseTest):
cd = CmdDump()
if isinstance(x, str):
x = x.encode('utf-8')
cmds = tuple(('draw', x) if isinstance(x, str) else x for x in cmds)
cmds = tuple(('draw', x) if isinstance(x, str) else tuple(map(cnv, x)) for x in cmds)
s.vt_parser.parse_bytes(s, x, cd)
current = ''
q = []
@@ -82,23 +88,6 @@ class TestParser(BaseTest):
pb('\033c123', ('screen_reset', ), '123')
self.ae(str(s.line(0)), '123')
def test_charsets(self):
s = self.create_screen()
pb = partial(self.parse_bytes_dump, s)
pb(b'\xc3')
pb(b'\xa1', ('draw', b'\xc3\xa1'.decode('utf-8')))
s = self.create_screen()
pb = partial(self.parse_bytes_dump, s)
pb('\033)0\x0e/_', ('screen_designate_charset', 1, ord('0')), ('screen_change_charset', 1), '/_')
self.ae(str(s.line(0)), '/\xa0')
self.assertTrue(s.callbacks.iutf8)
pb('\033%@_', ('screen_use_latin1', 1), '_')
self.assertFalse(s.callbacks.iutf8)
s = self.create_screen()
pb = partial(self.parse_bytes_dump, s)
pb('\033(0/_', ('screen_designate_charset', 0, ord('0')), '/_')
self.ae(str(s.line(0)), '/\xa0')
def test_csi_codes(self):
s = self.create_screen()
pb = partial(self.parse_bytes_dump, s)
@@ -218,7 +207,7 @@ class TestParser(BaseTest):
s = self.create_screen()
pb = partial(self.parse_bytes_dump, s)
c = s.callbacks
pb('a\033]2;x\\ryz\x9cbcde', 'a', ('set_title', 'x\\ryz'), 'bcde')
pb('a\033]2;x\\ryz\033\\bcde', 'a', ('set_title', 'x\\ryz'), 'bcde')
self.ae(str(s.line(0)), 'abcde')
self.ae(c.titlebuf, ['x\\ryz'])
c.clear()
@@ -319,7 +308,7 @@ class TestParser(BaseTest):
c = s.callbacks
pb = partial(self.parse_bytes_dump, s)
q = hexlify(b'kind').decode('ascii')
pb(f'a\033P+q{q}\x9cbcde', 'a', ('screen_request_capabilities', 43, q), 'bcde')
pb(f'a\033P+q{q}\033\\bcde', 'a', ('screen_request_capabilities', 43, q), 'bcde')
self.ae(str(s.line(0)), 'abcde')
self.ae(c.wtcbuf, '1+r{}={}'.format(q, '1b5b313b3242').encode('ascii'))
c.clear()
@@ -331,24 +320,13 @@ class TestParser(BaseTest):
for sgr in '0;34;102;1;2;3;4 0;38:5:200;58:2:10:11:12'.split():
expected = set(sgr.split(';')) - {'0'}
c.clear()
s.vte_parser.parse_bytes(s, f'\033[{sgr}m\033P$qm\033\\'.encode('ascii'))
parse_bytes(s, f'\033[{sgr}m\033P$qm\033\\'.encode('ascii'))
r = c.wtcbuf.decode('ascii').partition('r')[2].partition('m')[0]
self.ae(expected, set(r.split(';')))
c.clear()
pb('\033P$qr\033\\', ('screen_request_capabilities', ord('$'), 'r'))
self.ae(c.wtcbuf, f'\033P1$r{s.margin_top + 1};{s.margin_bottom + 1}r\033\\'.encode('ascii'))
def test_sc81t(self):
s = self.create_screen()
pb = partial(self.parse_bytes_dump, s)
pb('\033 G', ('screen_set_8bit_controls', 1))
c = s.callbacks
pb('\033P$qm\033\\', ('screen_request_capabilities', ord('$'), 'm'))
self.ae(c.wtcbuf, b'\x901$rm\x9c')
c.clear()
pb('\033[0c', ('report_device_attributes', 0, 0))
self.ae(c.wtcbuf, b'\x9b?62;c')
def test_pending(self):
s = self.create_screen()
timeout = 0.1
@@ -400,12 +378,8 @@ class TestParser(BaseTest):
def test_oth_codes(self):
s = self.create_screen()
pb = partial(self.parse_bytes_dump, s)
for prefix in '\033_', '\u009f':
for suffix in '\u009c', '\033\\':
pb(f'a{prefix}+\\++{suffix}bcde', ('draw', 'a'), ('Unrecognized APC code: 0x2b',), ('draw', 'bcde'))
for prefix in '\033^', '\u009e':
for suffix in '\u009c', '\033\\':
pb(f'a{prefix}+\\++{suffix}bcde', ('draw', 'a'), ('Unrecognized PM code: 0x2b',), ('draw', 'bcde'))
pb('a\033_+\\+\033\\bcde', ('draw', 'a'), ('Unrecognized APC code: 0x2b',), ('draw', 'bcde'))
pb('a\033^+\\+\033\\bcde', ('draw', 'a'), ('Unrecognized PM code: 0x2b',), ('draw', 'bcde'))
def test_graphics_command(self):
from base64 import standard_b64encode

View File

@@ -1188,7 +1188,7 @@ class TestScreen(BaseTest):
def cb(data):
nonlocal response
response = set_pointer_shape(s, data)
response = set_pointer_shape(s, str(data, 'utf-8'))
c.set_pointer_shape = cb
def send(a):

View File

@@ -152,12 +152,12 @@ RPS1="{rps1}"
pty.callbacks.clear()
pty.send_cmd_to_child('printf "%s\x16\a%s" "a" "b"')
pty.wait_till(lambda: 'ab' in pty.screen_contents())
self.assertTrue(pty.screen.last_reported_cwd.endswith(self.home_dir))
self.assertTrue(pty.screen.last_reported_cwd.decode().endswith(self.home_dir))
self.assertIn('%s^G%s', pty.screen_contents())
q = os.path.join(self.home_dir, 'testing-cwd-notification-🐱')
os.mkdir(q)
pty.send_cmd_to_child(f'cd {q}')
pty.wait_till(lambda: pty.screen.last_reported_cwd.endswith(q))
pty.wait_till(lambda: pty.screen.last_reported_cwd.decode().endswith(q))
with self.run_shell(rc=f'''PS1="{ps1}"\nexport ES="a\n b c\nd"''') as pty:
pty.callbacks.clear()
pty.send_cmd_to_child('clone-in-kitty')
@@ -188,13 +188,13 @@ function _set_status_prompt; function fish_prompt; echo -n "$pipestatus $status
pty.wait_till(lambda: 'XDD_OK' in pty.screen_contents())
# CWD reporting
self.assertTrue(pty.screen.last_reported_cwd.endswith(self.home_dir))
self.assertTrue(pty.screen.last_reported_cwd.decode().endswith(self.home_dir))
q = os.path.join(self.home_dir, 'testing-cwd-notification-🐱')
os.mkdir(q)
pty.send_cmd_to_child(f'cd {q}')
pty.wait_till(lambda: pty.screen.last_reported_cwd.endswith(q))
pty.wait_till(lambda: pty.screen.last_reported_cwd.decode().endswith(q))
pty.send_cmd_to_child('cd -')
pty.wait_till(lambda: pty.screen.last_reported_cwd.endswith(self.home_dir))
pty.wait_till(lambda: pty.screen.last_reported_cwd.decode().endswith(self.home_dir))
# completion and prompt marking
pty.wait_till(lambda: 'cd -' not in pty.screen_contents().splitlines()[-1])
@@ -304,13 +304,13 @@ PS1="{ps1}"
pty.send_cmd_to_child('printf "%s\x16\a%s" "a" "b"')
pty.wait_till(lambda: pty.screen_contents().count(ps1) == 2)
self.ae(pty.screen_contents(), f'{ps1}printf "%s^G%s" "a" "b"\nab{ps1}')
self.assertTrue(pty.screen.last_reported_cwd.endswith(self.home_dir))
self.assertTrue(pty.screen.last_reported_cwd.decode().endswith(self.home_dir))
pty.send_cmd_to_child('echo $HISTFILE')
pty.wait_till(lambda: '.bash_history' in pty.screen_contents())
q = os.path.join(self.home_dir, 'testing-cwd-notification-🐱')
os.mkdir(q)
pty.send_cmd_to_child(f'cd {q}')
pty.wait_till(lambda: pty.screen.last_reported_cwd.endswith(q))
pty.wait_till(lambda: pty.screen.last_reported_cwd.decode().endswith(q))
for ps1 in ('line1\\nline\\2\\prompt> ', 'line1\nprompt> ', 'line1\\nprompt> ',):
with self.subTest(ps1=ps1), self.run_shell(