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()