mirror of
https://github.com/kovidgoyal/kitty
synced 2026-07-16 13:34:48 +02:00
Add tests for cmdline reporting in shell integration
This commit is contained in:
@@ -19,12 +19,12 @@ from typing import Optional
|
||||
from unittest import TestCase
|
||||
|
||||
from kitty.config import finalize_keys, finalize_mouse_mappings
|
||||
from kitty.fast_data_types import Cursor, HistoryBuf, LineBuf, Screen, get_options, set_options
|
||||
from kitty.fast_data_types import Cursor, HistoryBuf, LineBuf, Screen, get_options, monotonic, set_options
|
||||
from kitty.options.parse import merge_result_dicts
|
||||
from kitty.options.types import Options, defaults
|
||||
from kitty.types import MouseEvent
|
||||
from kitty.utils import read_screen_size
|
||||
from kitty.window import process_remote_print, process_title_from_child
|
||||
from kitty.window import decode_cmdline, process_remote_print, process_title_from_child
|
||||
|
||||
|
||||
def parse_bytes(screen, data, dump_callback=None):
|
||||
@@ -43,6 +43,9 @@ class Callbacks:
|
||||
self.pty = pty
|
||||
self.ftc = None
|
||||
self.set_pointer_shape = lambda data: None
|
||||
self.last_cmd_at = 0
|
||||
self.last_cmd_cmdline = ''
|
||||
self.last_cmd_exit_status = sys.maxsize
|
||||
|
||||
def write(self, data) -> None:
|
||||
self.wtcbuf += bytes(data)
|
||||
@@ -66,7 +69,13 @@ class Callbacks:
|
||||
pass
|
||||
|
||||
def cmd_output_marking(self, is_start: Optional[bool], data: str = '') -> None:
|
||||
pass
|
||||
if is_start:
|
||||
self.last_cmd_at = monotonic()
|
||||
self.last_cmd_cmdline = decode_cmdline(data) if data else data
|
||||
else:
|
||||
if self.last_cmd_at != 0:
|
||||
self.last_cmd_at = 0
|
||||
self.last_cmd_exit_status = int(data)
|
||||
|
||||
def request_capabilities(self, q) -> None:
|
||||
from kitty.terminfo import get_capabilities
|
||||
@@ -93,6 +102,9 @@ class Callbacks:
|
||||
self.bell_count = 0
|
||||
self.clone_cmds = []
|
||||
self.current_clone_data = ''
|
||||
self.last_cmd_exit_status = sys.maxsize
|
||||
self.last_cmd_cmdline = ''
|
||||
self.last_cmd_at = 0
|
||||
|
||||
def on_bell(self) -> None:
|
||||
self.bell_count += 1
|
||||
@@ -335,6 +347,8 @@ class PTY:
|
||||
self.process_input_from_child(0)
|
||||
|
||||
def send_cmd_to_child(self, cmd, flush=False):
|
||||
self.callbacks.last_cmd_exit_status = sys.maxsize
|
||||
self.last_cmd = cmd
|
||||
self.write_to_child(cmd + '\r', flush=flush)
|
||||
|
||||
def process_input_from_child(self, timeout=10):
|
||||
|
||||
@@ -121,6 +121,7 @@ RPS1="{rps1}"
|
||||
self.ae(pty.callbacks.titlebuf[-1], '~')
|
||||
pty.callbacks.clear()
|
||||
pty.send_cmd_to_child('mkdir test && ls -a')
|
||||
self.assert_command(pty)
|
||||
pty.wait_till(lambda: pty.screen_contents().count(rps1) == 2)
|
||||
self.ae(pty.callbacks.titlebuf[-2:], ['mkdir test && ls -a', '~'])
|
||||
q = '\n'.join(str(pty.screen.line(i)) for i in range(1, pty.screen.cursor.y))
|
||||
@@ -138,10 +139,12 @@ RPS1="{rps1}"
|
||||
pty.wait_till(redrawn)
|
||||
self.ae(q, str(pty.screen.line(pty.screen.cursor.y)))
|
||||
pty.write_to_child('\r')
|
||||
self.assert_command(pty, 'echo $COLUMNS')
|
||||
pty.wait_till(lambda: pty.screen_contents().count(rps1) == 3)
|
||||
self.ae('40', str(pty.screen.line(pty.screen.cursor.y - 1)))
|
||||
self.ae(q, str(pty.screen.line(pty.screen.cursor.y - 2)))
|
||||
pty.send_cmd_to_child('clear')
|
||||
self.assert_command(pty)
|
||||
q = ps1 + ' ' * (pty.screen.columns - len(ps1) - len(rps1)) + rps1
|
||||
pty.wait_till(lambda: pty.screen_contents() == q)
|
||||
pty.wait_till(lambda: pty.screen.cursor.shape == CURSOR_BEAM)
|
||||
@@ -149,6 +152,7 @@ RPS1="{rps1}"
|
||||
pty.wait_till(lambda: pty.screen.cursor.shape == 0)
|
||||
pty.write_to_child('\x04')
|
||||
pty.wait_till(lambda: pty.screen.cursor.shape == CURSOR_BEAM)
|
||||
self.assert_command(pty)
|
||||
with self.run_shell(rc=f'''PS1="{ps1}"''') as pty:
|
||||
pty.callbacks.clear()
|
||||
pty.send_cmd_to_child('printf "%s\x16\a%s" "a" "b"')
|
||||
@@ -159,10 +163,12 @@ RPS1="{rps1}"
|
||||
os.mkdir(q)
|
||||
pty.send_cmd_to_child(f'cd {q}')
|
||||
pty.wait_till(lambda: pty.screen.last_reported_cwd.decode().endswith(q))
|
||||
self.assert_command(pty)
|
||||
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')
|
||||
pty.wait_till(lambda: len(pty.callbacks.clone_cmds) == 1)
|
||||
self.assert_command(pty)
|
||||
env = pty.callbacks.clone_cmds[0].env
|
||||
self.ae(env.get('ES'), 'a\n b c\nd')
|
||||
|
||||
@@ -185,14 +191,17 @@ function _set_status_prompt; function fish_prompt; echo -n "$pipestatus $status
|
||||
self.ae(pty.screen_contents(), q)
|
||||
|
||||
# shell integration dir must no be in XDG_DATA_DIRS
|
||||
pty.send_cmd_to_child(f'string match -q -- "*{shell_integration_dir}*" "$XDG_DATA_DIRS" || echo "XDD_OK"')
|
||||
cmd = f'string match -q -- "*{shell_integration_dir}*" "$XDG_DATA_DIRS" || echo "XDD_OK"'
|
||||
pty.send_cmd_to_child(cmd)
|
||||
pty.wait_till(lambda: 'XDD_OK' in pty.screen_contents())
|
||||
self.assert_command(pty, cmd)
|
||||
|
||||
# CWD reporting
|
||||
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}')
|
||||
self.assert_command(pty)
|
||||
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.decode().endswith(self.home_dir))
|
||||
@@ -202,6 +211,7 @@ function _set_status_prompt; function fish_prompt; echo -n "$pipestatus $status
|
||||
pty.send_cmd_to_child('clear')
|
||||
pty.wait_till(lambda: pty.screen_contents().count(right_prompt) == 1)
|
||||
pty.send_cmd_to_child('_test_comp_path')
|
||||
self.assert_command(pty)
|
||||
pty.wait_till(lambda: pty.screen_contents().count(right_prompt) == 2)
|
||||
q = '\n'.join(str(pty.screen.line(i)) for i in range(1, pty.screen.cursor.y))
|
||||
self.ae(q, 'ok')
|
||||
@@ -221,6 +231,7 @@ function _set_status_prompt; function fish_prompt; echo -n "$pipestatus $status
|
||||
self.ae(q, str(pty.screen.line(pty.screen.cursor.y)))
|
||||
pty.write_to_child('\r')
|
||||
pty.wait_till(lambda: pty.screen_contents().count(right_prompt) == 3)
|
||||
self.assert_command(pty, 'echo $COLUMNS')
|
||||
self.ae('40', str(pty.screen.line(pty.screen.cursor.y - 1)))
|
||||
self.ae(q, str(pty.screen.line(pty.screen.cursor.y - 2)))
|
||||
|
||||
@@ -245,11 +256,17 @@ function _set_status_prompt; function fish_prompt; echo -n "$pipestatus $status
|
||||
pty.write_to_child('i')
|
||||
pty.wait_till(lambda: pty.screen.cursor.shape == CURSOR_BEAM)
|
||||
pty.send_cmd_to_child('_set_key default')
|
||||
self.assert_command(pty)
|
||||
pty.wait_till(lambda: pty.screen_contents().count(right_prompt) == 4)
|
||||
pty.wait_till(lambda: pty.screen.cursor.shape == CURSOR_BEAM)
|
||||
|
||||
pty.send_cmd_to_child('exit')
|
||||
|
||||
def assert_command(self, pty, cmd='', exit_status=0):
|
||||
cmd = cmd or pty.last_cmd
|
||||
pty.wait_till(lambda: pty.callbacks.last_cmd_exit_status == 0)
|
||||
pty.wait_till(lambda: pty.callbacks.last_cmd_cmdline == cmd)
|
||||
|
||||
@unittest.skipUnless(bash_ok(), 'bash not installed, too old, or debug build')
|
||||
def test_bash_integration(self):
|
||||
ps1 = 'prompt> '
|
||||
@@ -266,9 +283,11 @@ PS1="{ps1}"
|
||||
pty.wait_till(lambda: pty.callbacks.titlebuf[-1:] == ['~'])
|
||||
self.ae(pty.callbacks.titlebuf[-1], '~')
|
||||
pty.callbacks.clear()
|
||||
pty.send_cmd_to_child('mkdir test && ls -a')
|
||||
pty.wait_till(lambda: pty.callbacks.titlebuf[-2:] == ['mkdir test && ls -a', '~'])
|
||||
cmd = 'mkdir test && ls -a'
|
||||
pty.send_cmd_to_child(cmd)
|
||||
pty.wait_till(lambda: pty.callbacks.titlebuf[-2:] == [cmd, '~'])
|
||||
pty.wait_till(lambda: pty.screen_contents().count(ps1) == 2)
|
||||
self.assert_command(pty, cmd)
|
||||
q = '\n'.join(str(pty.screen.line(i)) for i in range(1, pty.screen.cursor.y))
|
||||
self.ae(pty.last_cmd_output(), q)
|
||||
# shrink the screen
|
||||
@@ -300,6 +319,7 @@ PS1="{ps1}"
|
||||
pty.callbacks.clear()
|
||||
pty.send_cmd_to_child('declare')
|
||||
pty.wait_till(lambda: 'LOCAL_KSI_VAR' in pty.screen_contents())
|
||||
self.assert_command(pty, 'declare')
|
||||
with self.run_shell(shell='bash', rc=f'''PS1="{ps1}"''') as pty:
|
||||
pty.callbacks.clear()
|
||||
pty.send_cmd_to_child('printf "%s\x16\a%s" "a" "b"')
|
||||
@@ -332,6 +352,7 @@ PS1="{ps1}"
|
||||
pty.wait_till(lambda: pty.screen_contents().count(ps1) == 3)
|
||||
self.ae('40', str(pty.screen.line(pty.screen.cursor.y - len(ps1.splitlines()))))
|
||||
self.ae(ps1.splitlines()[-1] + 'echo $COLUMNS', str(pty.screen.line(pty.screen.cursor.y - 1 - len(ps1.splitlines()))))
|
||||
self.assert_command(pty, 'echo $COLUMNS')
|
||||
|
||||
# test startup file sourcing
|
||||
|
||||
|
||||
Reference in New Issue
Block a user