115 lines
4.5 KiB
Python
115 lines
4.5 KiB
Python
"""Exercise the installed dashboard PTY reattachment contract without model calls."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
|
|
|
|
ROOT = Path(os.environ.get("HERMES_SOURCE_ROOT", "/opt/hermes"))
|
|
|
|
|
|
def load_source(name: str):
|
|
"""Load one installed upstream module without starting the dashboard."""
|
|
spec = importlib.util.spec_from_file_location(name, ROOT / "hermes_cli" / f"{name}.py")
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
bridge_module = load_source("pty_bridge")
|
|
session_module = load_source("pty_session")
|
|
|
|
|
|
class DashboardRedrawTests(unittest.TestCase):
|
|
"""Only the opted-in, living dashboard child receives a redraw signal."""
|
|
|
|
def test_spawn_requires_both_dashboard_and_alternate_mode(self):
|
|
for env, expected in (
|
|
({}, False),
|
|
({"HERMES_TUI_DASHBOARD": "1"}, False),
|
|
({"HERMES_TUI_INLINE": "0"}, False),
|
|
({"HERMES_TUI_DASHBOARD": "1", "HERMES_TUI_INLINE": "1"}, False),
|
|
({"HERMES_TUI_DASHBOARD": "1", "HERMES_TUI_INLINE": "0"}, True),
|
|
):
|
|
with self.subTest(env=env):
|
|
proc = SimpleNamespace(fd=12, pid=321, isalive=lambda: True)
|
|
spawn = Mock(return_value=proc)
|
|
backend = SimpleNamespace(PtyProcess=SimpleNamespace(spawn=spawn))
|
|
with patch.object(bridge_module, "ptyprocess", backend), patch.object(
|
|
bridge_module, "_PTY_AVAILABLE", True
|
|
), patch.object(bridge_module.os, "kill") as kill, patch.object(
|
|
bridge_module.os, "write"
|
|
) as write:
|
|
bridge = bridge_module.PtyBridge.spawn(["node", "entry.js"], env=env)
|
|
bridge.redraw_after_attach()
|
|
self.assertEqual(kill.call_count, int(expected))
|
|
if expected:
|
|
kill.assert_called_once_with(321, signal.SIGCONT)
|
|
write.assert_not_called()
|
|
|
|
def test_dead_or_closed_children_are_not_signalled(self):
|
|
for alive, closed in ((False, False), (True, True)):
|
|
proc = SimpleNamespace(fd=12, pid=321, isalive=lambda: alive)
|
|
bridge = bridge_module.PtyBridge(proc)
|
|
bridge._dashboard_redraw_on_attach = True
|
|
bridge._closed = closed
|
|
with patch.object(bridge_module.os, "kill") as kill:
|
|
bridge.redraw_after_attach()
|
|
kill.assert_not_called()
|
|
|
|
def test_child_exit_race_does_not_break_attachment(self):
|
|
proc = SimpleNamespace(fd=12, pid=321, isalive=lambda: True)
|
|
bridge = bridge_module.PtyBridge(proc)
|
|
bridge._dashboard_redraw_on_attach = True
|
|
with patch.object(bridge_module.os, "kill", side_effect=ProcessLookupError):
|
|
bridge.redraw_after_attach()
|
|
|
|
|
|
class ReplayTests(unittest.IsolatedAsyncioTestCase):
|
|
"""Reattachment restores the terminal after bounded replay completes."""
|
|
|
|
async def test_truncated_replay_is_followed_by_redraw_without_input(self):
|
|
events = []
|
|
bridge = SimpleNamespace(redraw_after_attach=lambda: events.append("redraw"))
|
|
session = session_module.PtySession("test", bridge, buffer_cap=8, read_timeout=.1)
|
|
|
|
class Socket:
|
|
async def send_bytes(self, data):
|
|
events.append(data)
|
|
|
|
async def close(self, code):
|
|
events.append(code)
|
|
|
|
first = Socket()
|
|
await session.attach(first)
|
|
self.assertEqual(events, [])
|
|
session.detach(first)
|
|
session.buffer.append(b"\x1b[?1049h" + b"x" * 64)
|
|
self.assertTrue(session.buffer.truncated)
|
|
self.assertEqual(session.buffer.snapshot(), b"x" * 8)
|
|
second = Socket()
|
|
await session.attach(second)
|
|
self.assertEqual(events, [b"x" * 8, "redraw"])
|
|
self.assertTrue(session.attached)
|
|
# A superseded socket must not detach the active viewer.
|
|
session.detach(first)
|
|
self.assertTrue(session.attached)
|
|
|
|
async def test_legacy_bridges_still_reattach(self):
|
|
session = session_module.PtySession("test", object(), buffer_cap=8, read_timeout=.1)
|
|
first = SimpleNamespace()
|
|
await session.attach(first)
|
|
session.detach(first)
|
|
await session.attach(SimpleNamespace())
|
|
self.assertTrue(session.attached)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|