Some research
This commit is contained in:
+415
-1
@@ -52,6 +52,11 @@ class _BashrcHookTestBase(TestCase):
|
||||
|
||||
mode = "pty"
|
||||
timeout = 30.0
|
||||
# Class-level defaults so a setup() that raises before these are assigned
|
||||
# (e.g. the TAI_CONTAINER_TEST guard below) doesn't also crash teardown()
|
||||
# with an unrelated AttributeError that masks the real failure reason.
|
||||
tmp_home = None
|
||||
_drain_task = None
|
||||
|
||||
async def setup(self):
|
||||
if os.environ.get("TAI_CONTAINER_TEST") != "1":
|
||||
@@ -88,7 +93,323 @@ class _BashrcHookTestBase(TestCase):
|
||||
async def teardown(self):
|
||||
if self._drain_task is not None:
|
||||
self._drain_task.cancel()
|
||||
shutil.rmtree(self.tmp_home, ignore_errors=True)
|
||||
if self.tmp_home is not None:
|
||||
shutil.rmtree(self.tmp_home, ignore_errors=True)
|
||||
|
||||
|
||||
class _ShellPlexerHookTestBase(TestCase):
|
||||
"""Same fixture shape as _BashrcHookTestBase, generalized over shell:
|
||||
installs the real plexer hook into a real $HOME/<rc file> for zsh or
|
||||
fish and spawns the real shell binary. Subclasses set `shell` ("zsh" or
|
||||
"fish") and `install_fn` (tai.op_zsh_plexer_hook / tai.op_fish_plexer_hook).
|
||||
Needs that shell actually installed -- present in tests/docker/Dockerfile,
|
||||
so setup() fails loud rather than silently skipping if it's missing,
|
||||
same posture as TmuxControllerLiveTest's missing-tmux check."""
|
||||
|
||||
mode = "pty"
|
||||
timeout = 30.0
|
||||
shell = None
|
||||
install_fn = None
|
||||
tmp_home = None
|
||||
_drain_task = None
|
||||
|
||||
async def setup(self):
|
||||
if os.environ.get("TAI_CONTAINER_TEST") != "1":
|
||||
raise TestAssertionFailed(
|
||||
"refusing to run (TAI_CONTAINER_TEST is not set): this test installs a real "
|
||||
"hook into a real shell rc file and must only ever run inside the disposable "
|
||||
"test container (see tests/docker/Dockerfile), never against a real machine's home"
|
||||
)
|
||||
if not shutil.which(self.shell):
|
||||
raise TestAssertionFailed("%s is not installed -- required for the %s plexer-hook tests" % (self.shell, self.shell))
|
||||
self.tmp_home = tempfile.mkdtemp(prefix="tai-test-home-%s-" % self.shell)
|
||||
os.makedirs(os.path.join(self.tmp_home, ".local", "bin"), exist_ok=True)
|
||||
real_home = os.environ.get("HOME")
|
||||
os.environ["HOME"] = self.tmp_home
|
||||
try:
|
||||
tai.op_binary("install", self.tmp_home)
|
||||
self.install_fn("install", self.tmp_home)
|
||||
finally:
|
||||
if real_home is not None:
|
||||
os.environ["HOME"] = real_home
|
||||
else:
|
||||
os.environ.pop("HOME", None)
|
||||
self._drain_task = None
|
||||
|
||||
async def _spawn_shell(self, args, mode=None, env_overrides=None):
|
||||
env = self.build_env()
|
||||
env.update(env_overrides or {})
|
||||
proc = ManagedProcess([self.shell] + list(args), mode=mode or self.mode, env=env, on_line=self._on_line)
|
||||
await proc.start()
|
||||
self._processes.append(proc)
|
||||
return proc
|
||||
|
||||
def _sock_matches(self):
|
||||
return glob.glob(os.path.join(self.tmp_home, ".tai", "plexer", "*.sock"))
|
||||
|
||||
async def teardown(self):
|
||||
if self._drain_task is not None:
|
||||
self._drain_task.cancel()
|
||||
if self.tmp_home is not None:
|
||||
shutil.rmtree(self.tmp_home, ignore_errors=True)
|
||||
|
||||
|
||||
class ZshInstallAndAutoWrapTest(_ShellPlexerHookTestBase):
|
||||
"""Real `zsh -i`, with the real hook installed into a real $HOME/.zshrc,
|
||||
auto-wraps into a real NativePlexerBackend -- the zsh counterpart of
|
||||
BashrcInstallAndAutoWrapTest. The hook content is byte-identical to the
|
||||
bash block (plain POSIX `[`/`case`, no bashism), only the target file
|
||||
differs; this is the test that actually proves that claim rather than
|
||||
just asserting it in a docstring."""
|
||||
|
||||
name = "zshrc-install-and-auto-wrap"
|
||||
shell = "zsh"
|
||||
install_fn = staticmethod(tai.op_zsh_plexer_hook)
|
||||
|
||||
async def run(self):
|
||||
proc = await self._spawn_shell(["-i"], env_overrides={"HOME": self.tmp_home})
|
||||
self._drain_task = asyncio.ensure_future(_drain_forever(proc))
|
||||
|
||||
sock_path = None
|
||||
for _ in range(50):
|
||||
matches = self._sock_matches()
|
||||
if matches:
|
||||
sock_path = matches[0]
|
||||
break
|
||||
await asyncio.sleep(0.2)
|
||||
if not sock_path:
|
||||
raise TestAssertionFailed(
|
||||
"zsh -i never auto-wrapped into a plexer backend via the installed .zshrc hook"
|
||||
)
|
||||
|
||||
controller = tai.NativePlexerController(sock_path)
|
||||
await asyncio.sleep(0.3)
|
||||
status = controller.status()
|
||||
|
||||
controller.send_keys("echo $TAI_PLEXER_SOCKET", enter=True)
|
||||
await asyncio.sleep(0.4)
|
||||
pane = controller.read_pane(50)
|
||||
|
||||
return {"sock_path": sock_path, "pane": pane, "status": status}
|
||||
|
||||
async def verify(self, result):
|
||||
if result["sock_path"] not in result["pane"]:
|
||||
raise TestAssertionFailed(
|
||||
"the zsh auto-wrapped inner shell's $TAI_PLEXER_SOCKET did not match the "
|
||||
"backend's own socket path: %r" % result["pane"][-300:]
|
||||
)
|
||||
if not result["status"].get("pid") or result["status"]["pid"] <= 0:
|
||||
raise TestAssertionFailed("status() did not report a live child pid after zsh auto-wrap")
|
||||
|
||||
|
||||
class ZshAntiRecursionTest(_ShellPlexerHookTestBase):
|
||||
name = "zshrc-hook-anti-recursion"
|
||||
shell = "zsh"
|
||||
install_fn = staticmethod(tai.op_zsh_plexer_hook)
|
||||
|
||||
async def run(self):
|
||||
proc = await self._spawn_shell(
|
||||
["-i"], env_overrides={"HOME": self.tmp_home, "TAI_PLEXER_ACTIVE": "1"}
|
||||
)
|
||||
await proc.send_line("echo anti_recursion_marker_ok")
|
||||
await proc.wait_for_pattern(re.compile(r"anti_recursion_marker_ok"), timeout=8.0)
|
||||
await asyncio.sleep(0.3)
|
||||
return {"sock_matches": self._sock_matches()}
|
||||
|
||||
async def verify(self, result):
|
||||
if result["sock_matches"]:
|
||||
raise TestAssertionFailed(
|
||||
"a plexer socket was created even though TAI_PLEXER_ACTIVE was already set under zsh"
|
||||
)
|
||||
|
||||
|
||||
class ZshTmuxActiveSkipTest(_ShellPlexerHookTestBase):
|
||||
name = "zshrc-hook-tmux-active-skip"
|
||||
shell = "zsh"
|
||||
install_fn = staticmethod(tai.op_zsh_plexer_hook)
|
||||
|
||||
async def run(self):
|
||||
proc = await self._spawn_shell(
|
||||
["-i"], env_overrides={"HOME": self.tmp_home, "TMUX": "dummy-session:0.0"}
|
||||
)
|
||||
await proc.send_line("echo tmux_skip_marker_ok")
|
||||
await proc.wait_for_pattern(re.compile(r"tmux_skip_marker_ok"), timeout=8.0)
|
||||
await asyncio.sleep(0.3)
|
||||
return {"sock_matches": self._sock_matches()}
|
||||
|
||||
async def verify(self, result):
|
||||
if result["sock_matches"]:
|
||||
raise TestAssertionFailed(
|
||||
"a plexer socket was created even though $TMUX was already set under zsh"
|
||||
)
|
||||
|
||||
|
||||
class ZshNonInteractiveSkipTest(_ShellPlexerHookTestBase):
|
||||
"""zsh counterpart of NonInteractiveSkipTest. Unlike bash, zsh's `-c`
|
||||
does not source .zshrc at all by default either, so this explicitly
|
||||
sources it, same technique as the bash version, to isolate the hook's
|
||||
own `case $- in *i*)` guard (identical text to the bash block) from
|
||||
zsh's separate default of not reading .zshrc for -c."""
|
||||
|
||||
name = "zshrc-hook-non-interactive-skip"
|
||||
shell = "zsh"
|
||||
install_fn = staticmethod(tai.op_zsh_plexer_hook)
|
||||
mode = "pipe"
|
||||
|
||||
async def run(self):
|
||||
proc = await self._spawn_shell(
|
||||
["-c", 'source "$HOME/.zshrc" 2>&1; echo marker_non_interactive_ok'],
|
||||
mode="pipe",
|
||||
env_overrides={"HOME": self.tmp_home},
|
||||
)
|
||||
output = await proc.drain_until_eof(timeout=10.0)
|
||||
returncode = await proc.wait(timeout=5.0)
|
||||
return {"output": output, "returncode": returncode}
|
||||
|
||||
async def verify(self, result):
|
||||
if "marker_non_interactive_ok" not in result["output"]:
|
||||
raise TestAssertionFailed(
|
||||
"a non-interactive `zsh -c` that explicitly sourced the hook never reached past "
|
||||
"it: %r" % result["output"][-500:]
|
||||
)
|
||||
if result["returncode"] != 0:
|
||||
raise TestAssertionFailed(
|
||||
"non-interactive zsh exited %r after sourcing the hook" % result["returncode"]
|
||||
)
|
||||
if self._sock_matches():
|
||||
raise TestAssertionFailed(
|
||||
"a plexer socket was created from a non-interactive zsh -- the "
|
||||
"`case $- in *i*)` guard failed"
|
||||
)
|
||||
|
||||
|
||||
class FishInstallAndAutoWrapTest(_ShellPlexerHookTestBase):
|
||||
"""Real `fish -i`, with the real hook installed into a real
|
||||
$HOME/.config/fish/config.fish, auto-wraps into a real NativePlexerBackend.
|
||||
fish needs its own syntax entirely (no `[`/`case`/`$-`): `status
|
||||
is-interactive`, `test`, `and`, and the `isatty` builtin instead of
|
||||
`test -t`, which fish's `test` does not support -- this is the test that
|
||||
proves that syntax actually works, not just that it looks plausible."""
|
||||
|
||||
name = "fish-config-install-and-auto-wrap"
|
||||
shell = "fish"
|
||||
install_fn = staticmethod(tai.op_fish_plexer_hook)
|
||||
|
||||
async def run(self):
|
||||
proc = await self._spawn_shell(["-i"], env_overrides={"HOME": self.tmp_home})
|
||||
self._drain_task = asyncio.ensure_future(_drain_forever(proc))
|
||||
|
||||
sock_path = None
|
||||
for _ in range(50):
|
||||
matches = self._sock_matches()
|
||||
if matches:
|
||||
sock_path = matches[0]
|
||||
break
|
||||
await asyncio.sleep(0.2)
|
||||
if not sock_path:
|
||||
raise TestAssertionFailed(
|
||||
"fish -i never auto-wrapped into a plexer backend via the installed config.fish hook"
|
||||
)
|
||||
|
||||
controller = tai.NativePlexerController(sock_path)
|
||||
await asyncio.sleep(0.3)
|
||||
status = controller.status()
|
||||
|
||||
controller.send_keys("echo $TAI_PLEXER_SOCKET", enter=True)
|
||||
await asyncio.sleep(0.4)
|
||||
pane = controller.read_pane(50)
|
||||
|
||||
return {"sock_path": sock_path, "pane": pane, "status": status}
|
||||
|
||||
async def verify(self, result):
|
||||
if result["sock_path"] not in result["pane"]:
|
||||
raise TestAssertionFailed(
|
||||
"the fish auto-wrapped inner shell's $TAI_PLEXER_SOCKET did not match the "
|
||||
"backend's own socket path: %r" % result["pane"][-300:]
|
||||
)
|
||||
if not result["status"].get("pid") or result["status"]["pid"] <= 0:
|
||||
raise TestAssertionFailed("status() did not report a live child pid after fish auto-wrap")
|
||||
|
||||
|
||||
class FishAntiRecursionTest(_ShellPlexerHookTestBase):
|
||||
name = "fish-config-hook-anti-recursion"
|
||||
shell = "fish"
|
||||
install_fn = staticmethod(tai.op_fish_plexer_hook)
|
||||
|
||||
async def run(self):
|
||||
proc = await self._spawn_shell(
|
||||
["-i"], env_overrides={"HOME": self.tmp_home, "TAI_PLEXER_ACTIVE": "1"}
|
||||
)
|
||||
await proc.send_line("echo anti_recursion_marker_ok")
|
||||
await proc.wait_for_pattern(re.compile(r"anti_recursion_marker_ok"), timeout=8.0)
|
||||
await asyncio.sleep(0.3)
|
||||
return {"sock_matches": self._sock_matches()}
|
||||
|
||||
async def verify(self, result):
|
||||
if result["sock_matches"]:
|
||||
raise TestAssertionFailed(
|
||||
"a plexer socket was created even though TAI_PLEXER_ACTIVE was already set under fish"
|
||||
)
|
||||
|
||||
|
||||
class FishTmuxActiveSkipTest(_ShellPlexerHookTestBase):
|
||||
name = "fish-config-hook-tmux-active-skip"
|
||||
shell = "fish"
|
||||
install_fn = staticmethod(tai.op_fish_plexer_hook)
|
||||
|
||||
async def run(self):
|
||||
proc = await self._spawn_shell(
|
||||
["-i"], env_overrides={"HOME": self.tmp_home, "TMUX": "dummy-session:0.0"}
|
||||
)
|
||||
await proc.send_line("echo tmux_skip_marker_ok")
|
||||
await proc.wait_for_pattern(re.compile(r"tmux_skip_marker_ok"), timeout=8.0)
|
||||
await asyncio.sleep(0.3)
|
||||
return {"sock_matches": self._sock_matches()}
|
||||
|
||||
async def verify(self, result):
|
||||
if result["sock_matches"]:
|
||||
raise TestAssertionFailed(
|
||||
"a plexer socket was created even though $TMUX was already set under fish"
|
||||
)
|
||||
|
||||
|
||||
class FishNonInteractiveSkipTest(_ShellPlexerHookTestBase):
|
||||
"""fish counterpart of NonInteractiveSkipTest. Unlike bash/zsh, fish
|
||||
sources config.fish on *every* startup, interactive or not (per fish's
|
||||
own docs) -- so this test does not need to explicitly source anything;
|
||||
a plain `fish -c` already exercises the hook automatically, which makes
|
||||
it the most direct possible test of the `status is-interactive` guard
|
||||
actually being the thing keeping a non-interactive fish safe."""
|
||||
|
||||
name = "fish-config-hook-non-interactive-skip"
|
||||
shell = "fish"
|
||||
install_fn = staticmethod(tai.op_fish_plexer_hook)
|
||||
mode = "pipe"
|
||||
|
||||
async def run(self):
|
||||
proc = await self._spawn_shell(
|
||||
["-c", "echo marker_non_interactive_ok"],
|
||||
mode="pipe",
|
||||
env_overrides={"HOME": self.tmp_home},
|
||||
)
|
||||
output = await proc.drain_until_eof(timeout=10.0)
|
||||
returncode = await proc.wait(timeout=5.0)
|
||||
return {"output": output, "returncode": returncode}
|
||||
|
||||
async def verify(self, result):
|
||||
if "marker_non_interactive_ok" not in result["output"]:
|
||||
raise TestAssertionFailed(
|
||||
"a non-interactive `fish -c` never reached past config.fish: %r" % result["output"][-500:]
|
||||
)
|
||||
if result["returncode"] != 0:
|
||||
raise TestAssertionFailed(
|
||||
"non-interactive fish exited %r after sourcing config.fish" % result["returncode"]
|
||||
)
|
||||
if self._sock_matches():
|
||||
raise TestAssertionFailed(
|
||||
"a plexer socket was created from a non-interactive fish -- the "
|
||||
"`status is-interactive` guard failed"
|
||||
)
|
||||
|
||||
|
||||
class BashrcInstallAndAutoWrapTest(_BashrcHookTestBase):
|
||||
@@ -376,6 +697,90 @@ class PlexerHookInstallUninstallRoundTripTest(TestCase):
|
||||
shutil.rmtree(self.tmp_dir, ignore_errors=True)
|
||||
|
||||
|
||||
class DualHookSameFileIsolationTest(TestCase):
|
||||
"""bash and zsh both put the command-not-found hook and the plexer hook
|
||||
into the *same* rc file (~/.bashrc / ~/.zshrc); fish puts both into the
|
||||
same config.fish too. Each pair uses a distinct marker (BASHRC_MARK_*
|
||||
vs PLEXER_BASHRC_MARK_*), but that's only a correctness claim until it's
|
||||
proven the two blocks never clobber or duplicate each other when they
|
||||
share one file -- which is exactly the failure mode the user asked never
|
||||
to risk. Deterministic, no subprocess, safe to run anywhere."""
|
||||
|
||||
name = "shell-hook-dual-block-same-file-isolation"
|
||||
timeout = 10.0
|
||||
|
||||
CASES = (
|
||||
("bash", tai.upsert_bashrc_block, tai.remove_bashrc_block, tai.hook_present,
|
||||
tai.upsert_plexer_bashrc_block, tai.remove_plexer_bashrc_block, tai.plexer_hook_present),
|
||||
("zsh", tai.upsert_zsh_hook_block, tai.remove_zsh_hook_block, tai.zsh_hook_present,
|
||||
tai.upsert_zsh_plexer_block, tai.remove_zsh_plexer_block, tai.zsh_plexer_hook_present),
|
||||
("fish", tai.upsert_fish_hook_block, tai.remove_fish_hook_block, tai.fish_hook_present,
|
||||
tai.upsert_fish_plexer_block, tai.remove_fish_plexer_block, tai.fish_plexer_hook_present),
|
||||
)
|
||||
|
||||
async def setup(self):
|
||||
self.tmp_dir = tempfile.mkdtemp(prefix="tai-test-dual-hook-")
|
||||
|
||||
async def run(self):
|
||||
outcomes = {}
|
||||
for shell, up_a, rm_a, present_a, up_b, rm_b, present_b in self.CASES:
|
||||
path = os.path.join(self.tmp_dir, shell + "rc")
|
||||
with open(path, "w", encoding="utf-8") as handle:
|
||||
handle.write("# pre-existing user content\nexport FOO=bar\n")
|
||||
|
||||
up_a(path)
|
||||
up_b(path)
|
||||
with open(path, encoding="utf-8") as handle:
|
||||
both_installed = handle.read()
|
||||
both_present_after_install = (present_a(path), present_b(path))
|
||||
|
||||
rm_a(path)
|
||||
with open(path, encoding="utf-8") as handle:
|
||||
after_remove_a = handle.read()
|
||||
presence_after_remove_a = (present_a(path), present_b(path))
|
||||
|
||||
up_a(path)
|
||||
with open(path, encoding="utf-8") as handle:
|
||||
after_reinstall_a = handle.read()
|
||||
|
||||
rm_a(path)
|
||||
rm_b(path)
|
||||
with open(path, encoding="utf-8") as handle:
|
||||
after_remove_both = handle.read()
|
||||
|
||||
outcomes[shell] = {
|
||||
"both_installed": both_installed,
|
||||
"both_present_after_install": both_present_after_install,
|
||||
"after_remove_a": after_remove_a,
|
||||
"presence_after_remove_a": presence_after_remove_a,
|
||||
"after_reinstall_a": after_reinstall_a,
|
||||
"after_remove_both": after_remove_both,
|
||||
}
|
||||
return outcomes
|
||||
|
||||
async def verify(self, result):
|
||||
for shell, data in result.items():
|
||||
if data["both_present_after_install"] != (True, True):
|
||||
raise TestAssertionFailed("%s: both hooks should be present after installing both: %r" % (shell, data["both_present_after_install"]))
|
||||
if data["presence_after_remove_a"] != (False, True):
|
||||
raise TestAssertionFailed(
|
||||
"%s: removing hook A should leave hook B alone, got present=(A, B)=%r" % (shell, data["presence_after_remove_a"])
|
||||
)
|
||||
if data["after_remove_a"].count("export FOO=bar") != 1:
|
||||
raise TestAssertionFailed("%s: removing hook A lost or duplicated pre-existing content" % shell)
|
||||
if "export FOO=bar" not in data["after_remove_both"]:
|
||||
raise TestAssertionFailed("%s: removing both hooks lost pre-existing content" % shell)
|
||||
if data["after_remove_both"].strip() != "# pre-existing user content\nexport FOO=bar".strip():
|
||||
raise TestAssertionFailed(
|
||||
"%s: file after removing both hooks has leftover marker text: %r" % (shell, data["after_remove_both"])
|
||||
)
|
||||
if data["after_reinstall_a"].count("export FOO=bar") != 1:
|
||||
raise TestAssertionFailed("%s: reinstalling hook A duplicated pre-existing content" % shell)
|
||||
|
||||
async def teardown(self):
|
||||
shutil.rmtree(self.tmp_dir, ignore_errors=True)
|
||||
|
||||
|
||||
ALL_CASES = [
|
||||
BashrcInstallAndAutoWrapTest(),
|
||||
AntiRecursionTest(),
|
||||
@@ -383,6 +788,15 @@ ALL_CASES = [
|
||||
NonInteractiveSkipTest(),
|
||||
TmuxControllerLiveTest(),
|
||||
PlexerHookInstallUninstallRoundTripTest(),
|
||||
ZshInstallAndAutoWrapTest(),
|
||||
ZshAntiRecursionTest(),
|
||||
ZshTmuxActiveSkipTest(),
|
||||
ZshNonInteractiveSkipTest(),
|
||||
FishInstallAndAutoWrapTest(),
|
||||
FishAntiRecursionTest(),
|
||||
FishTmuxActiveSkipTest(),
|
||||
FishNonInteractiveSkipTest(),
|
||||
DualHookSameFileIsolationTest(),
|
||||
]
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user