mirrors/kitty
1
0
Fork 0
mirror of https://github.com/kovidgoyal/kitty synced 2026-06-08 14:18:26 +02:00
Code Issues Packages Projects Releases Wiki Activity

unlink SHM file at end of each individual test

Browse Source
This commit is contained in:
Kovid Goyal
2022-03-10 10:57:36 +05:30
parent a1311a2332
commit 920086ae88
2 changed files with 32 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
kittens/ssh/main.py
View File

@@ -216,7 +216,7 @@ def bootstrap_script(
script_type: str = 'sh', remote_args: Sequence[str] = (),
ssh_opts_dict: Dict[str, Dict[str, Any]] = {},
test_script: str = '', request_id: Optional[str] = None, cli_hostname: str = '', cli_uname: str = ''
) -> Tuple[str, Dict[str, str]]:
) -> Tuple[str, Dict[str, str], SharedMemory]:
if request_id is None:
request_id = os.environ['KITTY_PID'] + '-' + os.environ['KITTY_WINDOW_ID']
exec_cmd = prepare_exec_cmd(remote_args, script_type == 'py') if remote_args else ''
@@ -236,7 +236,7 @@ def bootstrap_script(
'DATA_PASSWORD': pw, 'PASSWORD_FILENAME': shm.name, 'EXEC_CMD': exec_cmd, 'TEST_SCRIPT': test_script,
'REQUEST_ID': request_id
}
return prepare_script(ans, replacements), replacements
return prepare_script(ans, replacements), replacements, shm
def get_ssh_cli() -> Tuple[Set[str], Set[str]]:
@@ -435,7 +435,7 @@ def get_remote_command(
) -> Tuple[List[str], Dict[str, str]]:
q = os.path.basename(interpreter).lower()
is_python = 'python' in q
sh_script, replacements = bootstrap_script(
sh_script, replacements, shm = bootstrap_script(
script_type='py' if is_python else 'sh', remote_args=remote_args, ssh_opts_dict=ssh_opts_dict,
cli_hostname=cli_hostname, cli_uname=cli_uname)
return wrap_bootstrap_script(sh_script, interpreter), replacements

55
kitty_tests/ssh.py
View File

@@ -239,32 +239,35 @@ copy --exclude */w.* d1
test_script = f'print("UNTAR_DONE", flush=True); {test_script}'
else:
test_script = f'echo "UNTAR_DONE"; {test_script}'
script = bootstrap_script(
script, replacements, shm = bootstrap_script(
script_type='py' if 'python' in sh else 'sh', request_id="testing",
test_script=test_script, ssh_opts_dict={'*': ssh_opts},
)[0]
env = basic_shell_env(home_dir)
# Avoid generating unneeded completion scripts
os.makedirs(os.path.join(home_dir, '.local', 'share', 'fish', 'generated_completions'), exist_ok=True)
# prevent newuser-install from running
open(os.path.join(home_dir, '.zshrc'), 'w').close()
options = {'shell_integration': shell_integration(SHELL_INTEGRATION_VALUE or 'disabled')}
cmd = wrap_bootstrap_script(script, sh)
pty = self.create_pty([launcher, '-c', ' '.join(cmd)], cwd=home_dir, env=env, options=options)
if pre_data:
pty.write_buf = pre_data.encode('utf-8')
del script
)
try:
env = basic_shell_env(home_dir)
# Avoid generating unneeded completion scripts
os.makedirs(os.path.join(home_dir, '.local', 'share', 'fish', 'generated_completions'), exist_ok=True)
# prevent newuser-install from running
open(os.path.join(home_dir, '.zshrc'), 'w').close()
options = {'shell_integration': shell_integration(SHELL_INTEGRATION_VALUE or 'disabled')}
cmd = wrap_bootstrap_script(script, sh)
pty = self.create_pty([launcher, '-c', ' '.join(cmd)], cwd=home_dir, env=env, options=options)
if pre_data:
pty.write_buf = pre_data.encode('utf-8')
del script
def check_untar_or_fail():
q = pty.screen_contents()
if 'bzip2' in q:
raise ValueError('Untarring failed with screen contents:\n' + q)
return 'UNTAR_DONE' in q
pty.wait_till(check_untar_or_fail)
self.assertTrue(os.path.exists(os.path.join(home_dir, '.terminfo/kitty.terminfo')))
if SHELL_INTEGRATION_VALUE != 'enabled':
pty.wait_till(lambda: len(pty.screen_contents().splitlines()) > 1)
self.assertEqual(pty.screen.cursor.shape, 0)
else:
pty.wait_till(lambda: pty.screen.cursor.shape == CURSOR_BEAM)
return pty
def check_untar_or_fail():
q = pty.screen_contents()
if 'bzip2' in q:
raise ValueError('Untarring failed with screen contents:\n' + q)
return 'UNTAR_DONE' in q
pty.wait_till(check_untar_or_fail)
self.assertTrue(os.path.exists(os.path.join(home_dir, '.terminfo/kitty.terminfo')))
if SHELL_INTEGRATION_VALUE != 'enabled':
pty.wait_till(lambda: len(pty.screen_contents().splitlines()) > 1)
self.assertEqual(pty.screen.cursor.shape, 0)
else:
pty.wait_till(lambda: pty.screen.cursor.shape == CURSOR_BEAM)
return pty
finally:
shm.unlink()