2022-02-21 15:27:25 +03:00
|
|
|
#!/usr/bin/env python
|
|
|
|
# License: GPLv3 Copyright: 2022, Kovid Goyal <kovid at kovidgoyal.net>
|
|
|
|
|
|
|
|
|
|
|
|
import os
|
2022-02-21 15:30:23 +03:00
|
|
|
import shutil
|
2022-02-21 15:35:36 +03:00
|
|
|
import tempfile
|
2022-02-21 15:30:23 +03:00
|
|
|
import unittest
|
2022-02-21 15:27:25 +03:00
|
|
|
from contextlib import contextmanager
|
|
|
|
|
2022-02-21 16:59:35 +03:00
|
|
|
from kitty.constants import kitty_base_dir, terminfo_dir
|
2022-02-21 15:27:25 +03:00
|
|
|
from kitty.fast_data_types import CURSOR_BEAM
|
2022-02-22 09:53:19 +03:00
|
|
|
from kitty.shell_integration import rc_inset, setup_zsh_env
|
2022-02-21 15:27:25 +03:00
|
|
|
|
|
|
|
from . import BaseTest
|
|
|
|
|
|
|
|
|
|
|
|
def safe_env_for_running_shell(home_dir, rc='', shell='zsh'):
|
|
|
|
ans = {
|
|
|
|
'PATH': os.environ['PATH'],
|
|
|
|
'HOME': home_dir,
|
|
|
|
'TERM': 'xterm-kitty',
|
|
|
|
'TERMINFO': terminfo_dir,
|
|
|
|
'KITTY_SHELL_INTEGRATION': 'enabled',
|
2022-02-21 16:59:35 +03:00
|
|
|
'KITTY_INSTALLATION_DIR': kitty_base_dir,
|
2022-02-21 15:27:25 +03:00
|
|
|
}
|
2022-02-21 16:59:35 +03:00
|
|
|
for x in ('USER', 'LANG'):
|
|
|
|
if os.environ.get(x):
|
|
|
|
ans[x] = os.environ[x]
|
2022-02-21 15:27:25 +03:00
|
|
|
if shell == 'zsh':
|
|
|
|
ans['ZLE_RPROMPT_INDENT'] = '0'
|
|
|
|
with open(os.path.join(home_dir, '.zshenv'), 'w') as f:
|
|
|
|
print('unset GLOBAL_RCS', file=f)
|
|
|
|
with open(os.path.join(home_dir, '.zshrc'), 'w') as f:
|
2022-02-21 16:59:35 +03:00
|
|
|
print(rc + '\n', file=f)
|
2022-02-21 15:27:25 +03:00
|
|
|
setup_zsh_env(ans)
|
2022-02-22 09:53:19 +03:00
|
|
|
elif shell == 'bash':
|
|
|
|
with open(os.path.join(home_dir, '.bash_profile'), 'w') as f:
|
|
|
|
print('if [ -f ~/.bashrc ]; then . ~/.bashrc; fi', file=f)
|
|
|
|
with open(os.path.join(home_dir, '.bashrc'), 'w') as f:
|
|
|
|
# there is no way to prevent bash from reading the system
|
|
|
|
# /etc/profile and some distros set a PROMPT_COMMAND in it.
|
|
|
|
print('unset PROMPT_COMMAND', file=f)
|
|
|
|
# ensure LINES and COLUMNS are kept up to date
|
|
|
|
print('shopt -s checkwinsize', file=f)
|
|
|
|
# Not sure why bash turns off echo in this scenario
|
|
|
|
print('stty echo', file=f)
|
|
|
|
if rc:
|
|
|
|
print(rc, file=f)
|
|
|
|
print(rc_inset('bash'), file=f)
|
2022-02-21 15:27:25 +03:00
|
|
|
return ans
|
|
|
|
|
|
|
|
|
|
|
|
class ShellIntegration(BaseTest):
|
|
|
|
|
|
|
|
@contextmanager
|
2022-02-21 16:59:35 +03:00
|
|
|
def run_shell(self, shell='zsh', rc='', cmd='{shell} -il'):
|
2022-02-21 15:35:36 +03:00
|
|
|
home_dir = os.path.realpath(tempfile.mkdtemp())
|
|
|
|
try:
|
2022-02-22 09:53:19 +03:00
|
|
|
pty = self.create_pty(cmd.format(**locals()), cwd=home_dir, env=safe_env_for_running_shell(home_dir, rc=rc, shell=shell))
|
2022-02-21 15:27:25 +03:00
|
|
|
i = 10
|
|
|
|
while i > 0 and not pty.screen_contents().strip():
|
|
|
|
pty.process_input_from_child()
|
|
|
|
i -= 1
|
|
|
|
yield pty
|
2022-02-21 15:35:36 +03:00
|
|
|
finally:
|
|
|
|
if os.path.exists(home_dir):
|
|
|
|
shutil.rmtree(home_dir)
|
2022-02-21 15:27:25 +03:00
|
|
|
|
2022-02-21 15:30:23 +03:00
|
|
|
@unittest.skipUnless(shutil.which('zsh'), 'zsh not installed')
|
2022-02-21 15:27:25 +03:00
|
|
|
def test_zsh_integration(self):
|
|
|
|
ps1, rps1 = 'left>', '<right'
|
|
|
|
with self.run_shell(
|
|
|
|
rc=f'''
|
|
|
|
PS1="{ps1}"
|
|
|
|
RPS1="{rps1}"
|
|
|
|
''') as pty:
|
|
|
|
q = ps1 + ' ' * (pty.screen.columns - len(ps1) - len(rps1)) + rps1
|
2022-02-21 15:48:31 +03:00
|
|
|
try:
|
|
|
|
pty.wait_till(lambda: pty.screen.cursor.shape == CURSOR_BEAM)
|
|
|
|
except TimeoutError:
|
|
|
|
raise AssertionError(f'Cursor was not changed to beam. Screen contents: {repr(pty.screen_contents())}')
|
2022-02-21 15:37:54 +03:00
|
|
|
self.ae(pty.screen_contents(), q)
|
2022-02-22 09:53:19 +03:00
|
|
|
self.ae(pty.callbacks.titlebuf[-1], '~')
|
2022-02-21 18:14:10 +03:00
|
|
|
pty.callbacks.clear()
|
2022-02-21 15:27:25 +03:00
|
|
|
pty.send_cmd_to_child('mkdir test && ls -a')
|
2022-02-21 18:31:16 +03:00
|
|
|
pty.wait_till(lambda: pty.screen_contents().count(rps1) == 2)
|
2022-02-22 09:53:19 +03:00
|
|
|
self.ae(pty.callbacks.titlebuf[-2:], ['mkdir test && ls -a', '~'])
|
2022-02-21 16:59:35 +03:00
|
|
|
q = '\n'.join(str(pty.screen.line(i)) for i in range(1, pty.screen.cursor.y))
|
|
|
|
self.ae(pty.last_cmd_output(), q)
|
2022-02-21 18:31:16 +03:00
|
|
|
# shrink the screen
|
|
|
|
pty.write_to_child(r'echo $COLUMNS')
|
|
|
|
pty.set_window_size(rows=20, columns=40)
|
|
|
|
q = ps1 + 'echo $COLUMNS' + ' ' * (40 - len(ps1) - len(rps1) - len('echo $COLUMNS')) + rps1
|
|
|
|
pty.process_input_from_child()
|
|
|
|
|
|
|
|
def redrawn():
|
|
|
|
q = pty.screen_contents()
|
|
|
|
return '$COLUMNS' in q and q.count(rps1) == 2 and q.count(ps1) == 2
|
|
|
|
|
|
|
|
pty.wait_till(redrawn)
|
|
|
|
self.ae(q, str(pty.screen.line(pty.screen.cursor.y)))
|
|
|
|
pty.write_to_child('\r')
|
|
|
|
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)))
|
2022-02-22 09:53:19 +03:00
|
|
|
|
|
|
|
@unittest.skipUnless(shutil.which('bash'), 'bash not installed')
|
|
|
|
def test_bash_integration(self):
|
|
|
|
ps1 = 'prompt> '
|
|
|
|
with self.run_shell(
|
|
|
|
shell='bash', rc=f'''
|
|
|
|
PS1="{ps1}"
|
|
|
|
''') as pty:
|
|
|
|
try:
|
|
|
|
pty.wait_till(lambda: pty.screen.cursor.shape == CURSOR_BEAM)
|
|
|
|
except TimeoutError:
|
|
|
|
raise AssertionError(f'Cursor was not changed to beam. Screen contents: {repr(pty.screen_contents())}')
|
|
|
|
pty.wait_till(lambda: pty.screen_contents().count(ps1) == 1)
|
|
|
|
self.ae(pty.screen_contents(), 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', '~'])
|
|
|
|
pty.wait_till(lambda: pty.screen_contents().count(ps1) == 2)
|
|
|
|
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
|
|
|
|
pty.write_to_child(r'echo $COLUMNS')
|
|
|
|
pty.set_window_size(rows=20, columns=40)
|
|
|
|
pty.process_input_from_child()
|
|
|
|
|
|
|
|
def redrawn():
|
|
|
|
q = pty.screen_contents()
|
|
|
|
return '$COLUMNS' in q and q.count(ps1) == 2
|
|
|
|
|
|
|
|
pty.wait_till(redrawn)
|
|
|
|
self.ae(ps1 + 'echo $COLUMNS', str(pty.screen.line(pty.screen.cursor.y)))
|
|
|
|
pty.write_to_child('\r')
|
|
|
|
pty.wait_till(lambda: pty.screen_contents().count(ps1) == 3)
|
|
|
|
self.ae('40', str(pty.screen.line(pty.screen.cursor.y - 1)))
|
|
|
|
self.ae(ps1 + 'echo $COLUMNS', str(pty.screen.line(pty.screen.cursor.y - 2)))
|