mirror of
https://github.com/kovidgoyal/kitty.git
synced 2026-08-04 14:46:03 +00:00
...
This commit is contained in:
Kovid Goyal
parent
2bd68cf8f9
commit
e574422fa4
1 changed files with 57 additions and 18 deletions
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue