From 468d0c69a645f2fec58dabd1d5f3c3365708e433 Mon Sep 17 00:00:00 2001 From: Kovid Goyal Date: Thu, 9 May 2024 13:46:19 +0530 Subject: [PATCH] Add tests for cmdline reporting in shell integration --- kitty_tests/__init__.py | 20 +++++++++++++++++--- kitty_tests/shell_integration.py | 27 ++++++++++++++++++++++++--- 2 files changed, 41 insertions(+), 6 deletions(-) diff --git a/kitty_tests/__init__.py b/kitty_tests/__init__.py index 40ffb17d0..0fabfdc41 100644 --- a/kitty_tests/__init__.py +++ b/kitty_tests/__init__.py @@ -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): diff --git a/kitty_tests/shell_integration.py b/kitty_tests/shell_integration.py index acf43e330..faa264c25 100644 --- a/kitty_tests/shell_integration.py +++ b/kitty_tests/shell_integration.py @@ -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