kuddev
bcbae7c5da
Keep hook authority separate from screen observations and shell completion. Install owned native and remote bridges, preserve durable session recovery, and support Shift+Enter for byte-stream clients without changing native keys. Addresses #199.
208 lines
9.8 KiB
Python
208 lines
9.8 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
import select
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from urllib.parse import quote
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[2] / "nebula_app/res/shell"
|
|
|
|
|
|
class ShellSession:
|
|
def __init__(self, program: str, home: Path, args: list[str], env: dict[str, str]) -> None:
|
|
import pty
|
|
|
|
self.master, slave = pty.openpty()
|
|
self.output = b""
|
|
self.process = subprocess.Popen(
|
|
[program, *args], stdin=slave, stdout=slave, stderr=slave, cwd=home,
|
|
env={"PATH": os.environ["PATH"], "HOME": str(home), "TERM": "xterm-256color",
|
|
"LC_ALL": "en_US.UTF-8" if sys.platform == "darwin" else "C.UTF-8", "PS1": "nebula-test> ",
|
|
**{name: os.environ[name] for name in ("FPATH", "NEBULA_TEST_ZSH_MODULE_DIR") if name in os.environ},
|
|
**env},
|
|
start_new_session=True,
|
|
)
|
|
os.close(slave)
|
|
|
|
def wait(self, marker: bytes) -> bytes:
|
|
collected = b""
|
|
deadline = time.monotonic() + 8
|
|
while time.monotonic() < deadline:
|
|
if select.select([self.master], [], [], 0.1)[0]:
|
|
try:
|
|
block = os.read(self.master, 65536)
|
|
except OSError:
|
|
break
|
|
collected += block
|
|
self.output += block
|
|
if marker in collected:
|
|
return collected
|
|
raise AssertionError(f"shell never emitted {marker!r}: {self.output!r}")
|
|
|
|
def command(self, command: str, marker: bytes) -> bytes:
|
|
os.write(self.master, command.encode("utf-8") + b"\n")
|
|
return self.wait(marker)
|
|
|
|
def close(self) -> None:
|
|
if self.process.poll() is None:
|
|
self.process.kill()
|
|
self.process.wait(timeout=5)
|
|
os.close(self.master)
|
|
|
|
|
|
@unittest.skipUnless(os.name == "posix", "requires Unix PTYs")
|
|
class ShellIntegrationTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temporary = tempfile.TemporaryDirectory(prefix="nebula-shell-test-")
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.home = Path(self.temporary.name)
|
|
|
|
def start(self, shell: str, rc: str = "", original_zdotdir: Path | None = None) -> ShellSession:
|
|
program = shutil.which(shell)
|
|
if not program:
|
|
self.skipTest(f"{shell} is not installed; native CI must run this case")
|
|
if shell == "bash":
|
|
if sys.platform == "darwin":
|
|
self.skipTest("macOS Bash keeps native login startup; this rc wrapper is Linux-only")
|
|
(self.home / ".bashrc").write_text(rc, encoding="utf-8")
|
|
args = ["--noprofile", "--rcfile", str(SCRIPTS / "bashrc"), "-i"]
|
|
env = {}
|
|
else:
|
|
dotfiles = original_zdotdir or self.home
|
|
dotfiles.mkdir(exist_ok=True)
|
|
(dotfiles / ".zshenv").write_text(
|
|
'[[ -n $NEBULA_TEST_ZSH_MODULE_DIR ]] && module_path=("$NEBULA_TEST_ZSH_MODULE_DIR" $module_path)\n'
|
|
"export NEBULA_PROFILE_TEST=env\n", encoding="utf-8")
|
|
(dotfiles / ".zprofile").write_text("NEBULA_PROFILE_TEST+=:profile\n", encoding="utf-8")
|
|
(dotfiles / ".zshrc").write_text("NEBULA_PROFILE_TEST+=:rc\n" + rc, encoding="utf-8")
|
|
(dotfiles / ".zlogin").write_text("NEBULA_PROFILE_TEST+=:login\n", encoding="utf-8")
|
|
wrapper = self.home / "integration"
|
|
wrapper.mkdir()
|
|
for source, target in [("zshenv", ".zshenv"), ("zprofile", ".zprofile"), ("zshrc", ".zshrc")]:
|
|
shutil.copyfile(SCRIPTS / source, wrapper / target)
|
|
args = ["-d", "-l", "-i"]
|
|
env = {"ZDOTDIR": str(wrapper), "NEBULA_ZSH_INTEGRATION": str(wrapper),
|
|
"NEBULA_ZDOTDIR_WAS_SET": "1" if original_zdotdir else "0"}
|
|
if original_zdotdir:
|
|
env["NEBULA_ORIGINAL_ZDOTDIR"] = str(original_zdotdir)
|
|
session = ShellSession(program, self.home, args, env)
|
|
self.addCleanup(session.close)
|
|
session.wait(b"\x1b]133;A\x07")
|
|
return session
|
|
|
|
def check_protocol(self, shell: str) -> None:
|
|
session = self.start(shell)
|
|
output = session.command("(exit 7)", b"\x1b]133;D;7\x07")
|
|
self.assertIn(b"\x1b]133;C\x07", output)
|
|
directory = self.home / "目录 % # space"
|
|
directory.mkdir()
|
|
encoded = quote(str(directory), safe="/._~-").encode("ascii")
|
|
session.command(f"cd '{directory}'", b"\x1b]7;file://localhost" + encoded + b"\x07")
|
|
|
|
def test_bash_command_status_and_utf8_cwd(self) -> None:
|
|
self.check_protocol("bash")
|
|
|
|
def test_zsh_command_status_and_utf8_cwd(self) -> None:
|
|
self.check_protocol("zsh")
|
|
|
|
def test_zsh_loads_user_rcs_without_host_global_rcs(self) -> None:
|
|
session = self.start("zsh")
|
|
session.command('print -r -- "RCS=$options[rcs] GLOBAL_RCS=$options[globalrcs]"',
|
|
b"RCS=on GLOBAL_RCS=off")
|
|
|
|
def test_bash_preserves_prompt_command(self) -> None:
|
|
session = self.start("bash", "PROMPT_COMMAND=\"printf 'USER_PROMPT\\\\n'\"\n")
|
|
session.command("(exit 7)", b"\x1b]133;D;7\x07")
|
|
self.assertIn(b"USER_PROMPT", session.output)
|
|
|
|
def test_bash_does_not_replace_user_debug_trap(self) -> None:
|
|
session = self.start("bash", "trap 'printf USER_DEBUG' DEBUG\n")
|
|
session.command("trap -p DEBUG", b"trap -- 'printf USER_DEBUG' DEBUG")
|
|
self.assertIn(b"USER_DEBUG", session.output)
|
|
|
|
def test_bash_preserves_array_prompt_command(self) -> None:
|
|
session = self.start("bash", "PROMPT_COMMAND=('printf FIRST_PROMPT' 'printf SECOND_PROMPT')\n")
|
|
session.command("(exit 7)", b"\x1b]133;D;7\x07")
|
|
self.assertIn(b"FIRST_PROMPT", session.output)
|
|
self.assertIn(b"SECOND_PROMPT", session.output)
|
|
|
|
def test_bash_preserves_exported_array_and_failure_status(self) -> None:
|
|
session = self.start("bash", "declare -ax PROMPT_COMMAND=('printf ARRAY_STATUS=%s_END\\n \"$?\"')\n")
|
|
output = session.command("(exit 7)", b"\x1b]133;D;7\x07")
|
|
self.assertIn(b"ARRAY_STATUS=7_END", output)
|
|
|
|
def test_bash_repeat_source_does_not_duplicate_startup_or_prompt_hooks(self) -> None:
|
|
session = self.start("bash", "(( user_rc_count += 1 ))\nPROMPT_COMMAND=('printf USER_PROMPT')\n")
|
|
session.command(f"source '{SCRIPTS / 'bashrc'}'", b"\x1b]133;D;0\x07")
|
|
output = session.command('printf "RC_COUNT=%s_END\\n" "$user_rc_count"', b"\x1b]133;D;0\x07")
|
|
self.assertIn(b"RC_COUNT=1_END", output)
|
|
self.assertEqual(output.count(b"\x1b]133;D;0\x07"), 1)
|
|
|
|
def test_bash_inherited_guard_does_not_call_missing_functions(self) -> None:
|
|
session = self.start("bash", "export PROMPT_COMMAND='printf INHERITED_USER_PROMPT'\n")
|
|
output = session.command("bash --noprofile --norc -i", b"INHERITED_USER_PROMPT")
|
|
output += session.command("(exit 7)", b"INHERITED_USER_PROMPT")
|
|
self.assertNotIn(b"command not found", output)
|
|
|
|
def test_bash_remote_login_sources_profile_and_rc_once(self) -> None:
|
|
if not shutil.which("bash") or sys.platform == "darwin":
|
|
self.skipTest("requires modern Bash")
|
|
for forwards in [False, True]:
|
|
with self.subTest(profile_forwards_to_rc=forwards):
|
|
(self.home / ".bash_profile").write_text(
|
|
"(( profile_count += 1 ))\n" + ('source "$HOME/.bashrc"\n' if forwards else ""), encoding="utf-8")
|
|
(self.home / ".bashrc").write_text("(( user_rc_count += 1 ))\n", encoding="utf-8")
|
|
session = ShellSession(shutil.which("bash"), self.home,
|
|
["--rcfile", str(SCRIPTS / "bashrc"), "-i"],
|
|
{"PEBREL_REMOTE_LOGIN": "1"})
|
|
self.addCleanup(session.close)
|
|
session.wait(b"\x1b]133;A\x07")
|
|
output = session.command('printf "STARTUP=%s:%s_END\\n" "$profile_count" "$user_rc_count"', b"\x1b]133;D;0\x07")
|
|
self.assertIn(b"STARTUP=1:1_END", output)
|
|
|
|
def test_readonly_prompt_does_not_emit_unmatched_command_start(self) -> None:
|
|
if not shutil.which("bash"):
|
|
self.skipTest("requires Bash")
|
|
(self.home / ".bashrc").write_text("readonly PROMPT_COMMAND='printf READONLY_PROMPT'\n", encoding="utf-8")
|
|
session = ShellSession(shutil.which("bash"), self.home,
|
|
["--noprofile", "--rcfile", str(SCRIPTS / "bashrc"), "-i"], {})
|
|
self.addCleanup(session.close)
|
|
session.wait(b"READONLY_PROMPT")
|
|
output = session.command("true", b"READONLY_PROMPT")
|
|
self.assertNotIn(b"\x1b]133;C", output)
|
|
|
|
def test_zsh_sources_login_files_and_restores_zdotdir(self) -> None:
|
|
session = self.start("zsh")
|
|
session.command('printf "PROFILE=%s ZDOTDIR=%s\\n" "$NEBULA_PROFILE_TEST" "${ZDOTDIR-unset}"',
|
|
b"PROFILE=env:profile:rc:login ZDOTDIR=unset")
|
|
|
|
def test_zsh_preserves_custom_zdotdir(self) -> None:
|
|
directory = self.home / "custom dotfiles"
|
|
session = self.start("zsh", original_zdotdir=directory)
|
|
session.command('printf "PROFILE=%s ZDOTDIR=%s_END\\n" "$NEBULA_PROFILE_TEST" "$ZDOTDIR"',
|
|
f"PROFILE=env:profile:rc:login ZDOTDIR={directory}_END".encode())
|
|
|
|
def test_zsh_preserves_precmd_hooks_after_failure(self) -> None:
|
|
session = self.start("zsh", """
|
|
typeset -gi user_prompt_count=0
|
|
_user_precmd() {
|
|
local status_code=$?
|
|
(( ++user_prompt_count ))
|
|
printf 'USER_PRECMD_%s_STATUS=%s_END\\n' "$user_prompt_count" "$status_code"
|
|
}
|
|
precmd_functions=(_user_precmd)
|
|
""")
|
|
output = session.command("(exit 7)", b"USER_PRECMD_2_STATUS=7_END")
|
|
self.assertIn(b"\x1b]133;D;7\x07", output)
|
|
session.command("true", b"USER_PRECMD_3_STATUS=0_END")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|