From f3fabd6d225c528e3a8ec4a506dccb022ac1280f Mon Sep 17 00:00:00 2001 From: Kovid Goyal Date: Fri, 24 Jul 2026 11:11:30 +0530 Subject: [PATCH] ... --- kitty_tests/base.py | 75 ++++++++++++++++++++++++++++++++++----------- 1 file changed, 57 insertions(+), 18 deletions(-) diff --git a/kitty_tests/base.py b/kitty_tests/base.py index 61e624260..f85d6565e 100644 --- a/kitty_tests/base.py +++ b/kitty_tests/base.py @@ -40,13 +40,12 @@ def parse_bytes(screen, data, dump_callback=None): def draw_multicell( screen: Screen, text: str, width: int = 0, scale: int = 1, subscale_n: int = 0, subscale_d: int = 0, vertical_align: int = 0, horizontal_align: int = 0 - ) -> None: +) -> None: cmd = f'\x1b]{TEXT_SIZE_CODE};w={width}:s={scale}:n={subscale_n}:d={subscale_d}:v={vertical_align}:h={horizontal_align};{text}\a' parse_bytes(screen, cmd.encode()) class Callbacks: - def __init__(self, pty=None) -> None: self.clear() self.pty = pty @@ -62,14 +61,16 @@ class Callbacks: def notify_child_of_resize(self): self.num_of_resize_events += 1 - def on_reset(self) -> None: + def on_reset(self, is_hard_reset: bool = True) -> None: if self.pty is not None: self.pty.reset_termios_state() def color_control(self, code, data) -> None: from kitty.window import color_control + response = color_control(self.color_profile, code, data) if response: + def p(x): if '@' in x: return (to_color(x.partition('@')[0]), int(255 * float(x.partition('@')[2]))) @@ -77,7 +78,8 @@ class Callbacks: if ans is None: ans = x return ans - parts = {x.partition('=')[0]:p(x.partition('=')[2]) for x in response.split(';')[1:]} + + parts = {x.partition('=')[0]: p(x.partition('=')[2]) for x in response.split(';')[1:]} self.color_control_responses.append(parts) def title_changed(self, data, is_base64=False) -> None: @@ -113,6 +115,7 @@ class Callbacks: def request_capabilities(self, q) -> None: from kitty.terminfo import get_capabilities + for c in get_capabilities(q, None): self.write(c.encode('ascii')) @@ -188,6 +191,7 @@ class Callbacks: if self.current_clone_data: cdata, self.current_clone_data = self.current_clone_data, '' from kitty.launch import CloneCmd + self.clone_cmds.append(CloneCmd(cdata)) self.current_clone_data = '' return @@ -198,12 +202,14 @@ class Callbacks: def handle_remote_ssh(self, msg): from kittens.ssh.utils import get_ssh_data + if self.pty: - for line in get_ssh_data(msg, "testing"): + for line in get_ssh_data(msg, 'testing'): self.pty.write_to_child(line) def handle_remote_echo(self, msg): from base64 import standard_b64decode + if self.pty: data = standard_b64decode(msg) self.pty.write_to_child(data) @@ -251,17 +257,18 @@ def retry_on_failure(max_attempts=max_attempts, sleep_duration=sleep_duration): try: return func(*args, **kwargs) except Exception: - if attempt < max_attempts - 1: # Don't sleep on the last attempt + if attempt < max_attempts - 1: # Don't sleep on the last attempt time.sleep(sleep_duration) print(f'{func.__name__} failed, retrying in {sleep_duration} seconds', file=sys.stderr) else: - raise # Re-raise the last exception + raise # Re-raise the last exception + return wrapper + return decorator class BaseTest(TestCase): - ae = TestCase.assertEqual maxDiff = 2048 is_ci = is_ci @@ -287,6 +294,7 @@ class BaseTest(TestCase): def cmd_to_run_python_code(self, code): from kitty.constants import kitty_exe + return [kitty_exe(), '+runpy', code] def create_screen(self, cols=5, lines=5, scrollback=5, cell_width=10, cell_height=20, options=None): @@ -297,12 +305,32 @@ class BaseTest(TestCase): return s def create_pty( - self, argv=None, cols=80, lines=100, scrollback=100, cell_width=10, cell_height=20, - options=None, cwd=None, env=None, stdin_fd=None, stdout_fd=None, needs_da1=False, + self, + argv=None, + cols=80, + lines=100, + scrollback=100, + cell_width=10, + cell_height=20, + options=None, + cwd=None, + env=None, + stdin_fd=None, + stdout_fd=None, + needs_da1=False, ): self.set_options(options) return PTY( - argv, lines, cols, scrollback, cell_width, cell_height, cwd, env, stdin_fd=stdin_fd, stdout_fd=stdout_fd, + argv, + lines, + cols, + scrollback, + cell_width, + cell_height, + cwd, + env, + stdin_fd=stdin_fd, + stdout_fd=stdout_fd, needs_da1=needs_da1, ) @@ -334,10 +362,20 @@ def forwardable_stdio(): class PTY: - def __init__( - self, argv=None, rows=25, columns=80, scrollback=100, cell_width=10, cell_height=20, - cwd=None, env=None, stdin_fd=None, stdout_fd=None, needs_da1=True, window_id=0, + self, + argv=None, + rows=25, + columns=80, + scrollback=100, + cell_width=10, + cell_height=20, + cwd=None, + env=None, + stdin_fd=None, + stdout_fd=None, + needs_da1=True, + window_id=0, log_data_flow=False, ): self.is_child = False @@ -347,6 +385,7 @@ class PTY: self.write_buf = b'' if argv is None: from kitty.child import openpty + self.master_fd, self.slave_fd = openpty() self.child_pid = 0 self.initial_termios_state = termios.tcgetattr(self.master_fd) @@ -433,7 +472,7 @@ class PTY: self.process_input_from_child(0) def send_da1_response(self, data: str) -> None: - self.write_to_child('\x1b[' + data, flush=False) # ]]]]]] + self.write_to_child('\x1b[' + data, flush=False) # ]]]]]] def send_cmd_to_child(self, cmd, flush=False): self.callbacks.last_cmd_exit_status = sys.maxsize @@ -481,9 +520,7 @@ class PTY: ec = os.waitstatus_to_exitcode(status) if hasattr(os, 'waitstatus_to_exitcode') else require_exit_code self.child_waited_for = True if require_exit_code is not None and ec != require_exit_code: - raise AssertionError( - f'Child exited with exit status: {status} code: {ec} != {require_exit_code}.' - f' {self.screen_contents_for_error()}') + raise AssertionError(f'Child exited with exit status: {status} code: {ec} != {require_exit_code}. {self.screen_contents_for_error()}') return status with suppress(OSError): self.process_input_from_child(timeout=0.02) @@ -500,6 +537,7 @@ class PTY: def screen_contents_for_error(self): from kitty.window import as_text + ans = as_text(self.screen, add_history=True, as_ansi=False) return f'Screen contents as repr:\n{ans!r}\nScreen contents:\n{ans.rstrip()}' @@ -513,4 +551,5 @@ class PTY: def last_cmd_output(self, as_ansi=False, add_wrap_markers=False): from kitty.window import cmd_output + return cmd_output(self.screen, as_ansi=as_ansi, add_wrap_markers=add_wrap_markers)