mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-10 13:31:38 +00:00
e1067dbbe5
Main's rewritten test_tui_npm_install.py tests call _make_tui_argv expecting the Ink/npm flow unconditionally; with the dual-engine dispatch merged in, _resolve_tui_engine() auto-selects opentui whenever ui-opentui/dist is built in the repo, routing the call away from the path under test (first subprocess became 'node --version' instead of 'npm run build'). Pin the engine to ink via an autouse fixture, mirroring the existing pinning precedent in test_tui_resume_flow.py.
315 lines
12 KiB
Python
315 lines
12 KiB
Python
"""Unit tests for hermes_cli.win_pty_bridge — ConPTY spawning + byte forwarding.
|
|
|
|
Windows-only counterpart to tests/hermes_cli/test_pty_bridge.py. Drives
|
|
``WinPtyBridge`` with minimal Windows processes (``cmd.exe``, ``python -c …``)
|
|
to verify it behaves like a PTY you can read/write/resize/close, then a small
|
|
set of platform-fallback assertions (``is_available``, ``PtyUnavailableError``)
|
|
that run on every OS so the import surface stays exercised in CI.
|
|
|
|
The bridge is the ConPTY backend behind the dashboard ``/chat`` tab — see
|
|
``hermes_cli/web_server.py`` ``/api/pty`` handler — so these tests are the
|
|
unit-level half of the integration check that the dashboard chat pane is
|
|
actually live on native Windows.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
|
|
import pytest
|
|
|
|
# WinPtyBridge can be imported on every platform — ``is_available`` just
|
|
# returns False when pywinpty isn't usable. Importing the module itself
|
|
# must never raise, otherwise the web_server import branch becomes a trap.
|
|
from hermes_cli.win_pty_bridge import PtyUnavailableError, WinPtyBridge
|
|
|
|
windows_only = pytest.mark.skipif(
|
|
not sys.platform.startswith("win"),
|
|
reason="ConPTY bridge is Windows-only",
|
|
)
|
|
|
|
|
|
def _read_until(bridge: WinPtyBridge, needle: bytes, timeout: float = 10.0) -> bytes:
|
|
"""Accumulate PTY output until we see ``needle`` or time out.
|
|
|
|
Mirrors the helper in test_pty_bridge.py so failures look familiar.
|
|
"""
|
|
deadline = time.monotonic() + timeout
|
|
buf = bytearray()
|
|
while time.monotonic() < deadline:
|
|
chunk = bridge.read(timeout=0.2)
|
|
if chunk is None:
|
|
break
|
|
buf.extend(chunk)
|
|
if needle in buf:
|
|
return bytes(buf)
|
|
return bytes(buf)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cross-platform fallback semantics
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestWinPtyBridgeUnavailable:
|
|
"""Module-level surface that must stay importable on every OS so the
|
|
web_server platform branch doesn't blow up at import time when pywinpty
|
|
is missing or the host isn't Windows."""
|
|
|
|
def test_error_is_importable_and_carries_message(self):
|
|
err = PtyUnavailableError("conpty missing")
|
|
assert "conpty" in str(err)
|
|
|
|
def test_bridge_class_is_importable(self):
|
|
# The platform-branched import in web_server.py relies on this:
|
|
# from hermes_cli.win_pty_bridge import WinPtyBridge, PtyUnavailableError
|
|
# Both symbols must always exist; ``is_available()`` is the gate.
|
|
assert WinPtyBridge is not None
|
|
assert callable(WinPtyBridge.is_available)
|
|
|
|
@pytest.mark.skipif(sys.platform.startswith("win"), reason="non-Windows only")
|
|
def test_spawn_raises_unavailable_off_windows(self):
|
|
with pytest.raises(PtyUnavailableError):
|
|
WinPtyBridge.spawn(["true"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Windows-only end-to-end behaviour
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@windows_only
|
|
class TestWinPtyBridgeSpawn:
|
|
def test_is_available_on_windows(self):
|
|
assert WinPtyBridge.is_available() is True
|
|
|
|
def test_spawn_returns_bridge_with_pid(self):
|
|
bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "exit 0"])
|
|
try:
|
|
assert bridge.pid > 0
|
|
finally:
|
|
bridge.close()
|
|
|
|
def test_spawn_raises_on_missing_argv0(self, tmp_path):
|
|
# pywinpty wraps CreateProcessW failures; surface as OSError / RuntimeError.
|
|
bogus = str(tmp_path / "definitely-not-a-real-binary.exe")
|
|
with pytest.raises((FileNotFoundError, OSError, RuntimeError, PtyUnavailableError)):
|
|
WinPtyBridge.spawn([bogus])
|
|
|
|
|
|
@windows_only
|
|
class TestWinPtyBridgeIO:
|
|
def test_reads_child_stdout(self):
|
|
bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "echo hermes-ok"])
|
|
try:
|
|
output = _read_until(bridge, b"hermes-ok")
|
|
assert b"hermes-ok" in output
|
|
finally:
|
|
bridge.close()
|
|
|
|
def test_write_sends_to_child_stdin(self):
|
|
# python -c reads stdin, echoes a marker, exits. More reliable than
|
|
# ``cat`` (not on Windows) and doesn't depend on a particular shell.
|
|
script = (
|
|
"import sys; "
|
|
"line = sys.stdin.readline().strip(); "
|
|
"sys.stdout.write('GOT:' + line + '\\n'); "
|
|
"sys.stdout.flush()"
|
|
)
|
|
bridge = WinPtyBridge.spawn([sys.executable, "-c", script])
|
|
try:
|
|
bridge.write(b"hello-pty\r\n")
|
|
output = _read_until(bridge, b"GOT:hello-pty")
|
|
assert b"GOT:hello-pty" in output
|
|
finally:
|
|
bridge.close()
|
|
|
|
def test_write_after_close_is_silent(self):
|
|
bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "exit 0"])
|
|
bridge.close()
|
|
# Must not raise — the dashboard WebSocket reader sometimes writes
|
|
# a final keystroke after the user has already closed the tab.
|
|
bridge.write(b"ignored")
|
|
|
|
def test_read_returns_none_after_child_exits(self):
|
|
bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "echo done"])
|
|
try:
|
|
_read_until(bridge, b"done")
|
|
# Give the child a beat to exit, then drain until EOF.
|
|
deadline = time.monotonic() + 5.0
|
|
while bridge.is_alive() and time.monotonic() < deadline:
|
|
bridge.read(timeout=0.1)
|
|
got_none = False
|
|
for _ in range(20):
|
|
if bridge.read(timeout=0.1) is None:
|
|
got_none = True
|
|
break
|
|
assert got_none, "WinPtyBridge.read did not return None after child EOF"
|
|
finally:
|
|
bridge.close()
|
|
|
|
|
|
@windows_only
|
|
class TestWinPtyBridgeResize:
|
|
def test_resize_does_not_raise_on_live_child(self):
|
|
# ConPTY exposes no ioctl-equivalent for reading the child's current
|
|
# winsize from Python land, so we can't verify the new dimensions
|
|
# the way the POSIX test does (which reads TIOCGWINSZ). What we
|
|
# CAN guarantee is what the dashboard depends on: ``resize`` never
|
|
# raises, the bridge stays alive, and subsequent I/O still works.
|
|
bridge = WinPtyBridge.spawn(
|
|
[sys.executable, "-c", "import time; time.sleep(1.0)"],
|
|
cols=80,
|
|
rows=24,
|
|
)
|
|
try:
|
|
bridge.resize(cols=123, rows=45)
|
|
assert bridge.is_alive()
|
|
finally:
|
|
bridge.close()
|
|
|
|
def test_resize_clamps_garbage_dimensions(self):
|
|
# Mirror the POSIX clamp test: a broken winsize probe must never
|
|
# propagate to the ConPTY API. 131072 > unsigned short max — the
|
|
# bridge has to coerce it down without raising.
|
|
bridge = WinPtyBridge.spawn(
|
|
[sys.executable, "-c", "import time; time.sleep(1.0)"],
|
|
cols=80,
|
|
rows=24,
|
|
)
|
|
try:
|
|
bridge.resize(cols=131072, rows=1) # must not raise
|
|
bridge.resize(cols=0, rows=-5) # nor this
|
|
assert bridge.is_alive()
|
|
finally:
|
|
bridge.close()
|
|
|
|
def test_resize_after_close_is_silent(self):
|
|
bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "exit 0"])
|
|
bridge.close()
|
|
# Must not raise — closed bridges still receive late resize escapes
|
|
# from xterm.js when the browser tab is closed mid-stream.
|
|
bridge.resize(cols=100, rows=40)
|
|
|
|
|
|
@windows_only
|
|
class TestClampDimension:
|
|
"""The clamp helper is the load-bearing piece — the dashboard sends
|
|
untrusted winsize values straight from xterm.js, and pywinpty's
|
|
setwinsize will happily raise on out-of-range u16 values."""
|
|
|
|
def test_clamps_above_max(self):
|
|
from hermes_cli.win_pty_bridge import _MAX_COLS, _MAX_ROWS, _clamp
|
|
|
|
assert _clamp(131072, _MAX_COLS) == _MAX_COLS
|
|
assert _clamp(131072, _MAX_ROWS) == _MAX_ROWS
|
|
|
|
def test_floors_at_one(self):
|
|
from hermes_cli.win_pty_bridge import _MAX_COLS, _clamp
|
|
|
|
assert _clamp(0, _MAX_COLS) == 1
|
|
assert _clamp(-5, _MAX_COLS) == 1
|
|
|
|
def test_passes_through_sane_values(self):
|
|
from hermes_cli.win_pty_bridge import _MAX_COLS, _clamp
|
|
|
|
assert _clamp(80, _MAX_COLS) == 80
|
|
assert _clamp(2000, _MAX_COLS) == 2000
|
|
|
|
def test_non_numeric_falls_back_to_min(self):
|
|
from hermes_cli.win_pty_bridge import _MAX_COLS, _clamp
|
|
|
|
assert _clamp(None, _MAX_COLS) == 1 # type: ignore[arg-type]
|
|
assert _clamp("not-a-number", _MAX_COLS) == 1 # type: ignore[arg-type]
|
|
assert _clamp(float("nan"), _MAX_COLS) == 1 # type: ignore[arg-type]
|
|
assert _clamp(float("inf"), _MAX_COLS) == 1 # type: ignore[arg-type]
|
|
|
|
|
|
@windows_only
|
|
class TestWinPtyBridgeClose:
|
|
def test_close_is_idempotent(self):
|
|
bridge = WinPtyBridge.spawn(
|
|
[sys.executable, "-c", "import time; time.sleep(30)"]
|
|
)
|
|
bridge.close()
|
|
bridge.close() # must not raise
|
|
assert not bridge.is_alive()
|
|
|
|
def test_close_terminates_long_running_child(self):
|
|
bridge = WinPtyBridge.spawn(
|
|
[sys.executable, "-c", "import time; time.sleep(30)"]
|
|
)
|
|
pid = bridge.pid
|
|
assert bridge.is_alive(), f"child pid {pid} not alive before close"
|
|
bridge.close()
|
|
# The bridge itself reports liveness via pywinpty.isalive(), which is
|
|
# the same probe the dashboard PTY reader uses to decide when to stop
|
|
# forwarding bytes — verifying that flips to False is the contract
|
|
# that matters for /api/pty.
|
|
deadline = time.monotonic() + 5.0
|
|
while bridge.is_alive() and time.monotonic() < deadline:
|
|
time.sleep(0.1)
|
|
assert not bridge.is_alive(), (
|
|
f"WinPtyBridge.is_alive() still True after close(); pid {pid}"
|
|
)
|
|
|
|
|
|
@windows_only
|
|
class TestWinPtyBridgeEnv:
|
|
def test_cwd_is_respected(self, tmp_path):
|
|
bridge = WinPtyBridge.spawn(
|
|
[sys.executable, "-c", "import os; print(os.getcwd())"],
|
|
cwd=str(tmp_path),
|
|
)
|
|
try:
|
|
# Path is case-insensitive on Windows; compare lowercased.
|
|
needle_resolved = str(tmp_path.resolve()).lower().encode()
|
|
deadline = time.monotonic() + 5.0
|
|
buf = bytearray()
|
|
while time.monotonic() < deadline:
|
|
chunk = bridge.read(timeout=0.2)
|
|
if chunk is None:
|
|
break
|
|
buf.extend(chunk)
|
|
if needle_resolved in bytes(buf).lower():
|
|
break
|
|
assert needle_resolved in bytes(buf).lower(), (
|
|
f"cwd {tmp_path!s} not echoed by child; got {bytes(buf)!r}"
|
|
)
|
|
finally:
|
|
bridge.close()
|
|
|
|
def test_env_is_forwarded(self):
|
|
bridge = WinPtyBridge.spawn(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
"import os; print('HERMES_PTY_TEST=' + os.environ.get('HERMES_PTY_TEST',''))",
|
|
],
|
|
env={**os.environ, "HERMES_PTY_TEST": "pty-env-works"},
|
|
)
|
|
try:
|
|
output = _read_until(bridge, b"pty-env-works")
|
|
assert b"pty-env-works" in output
|
|
finally:
|
|
bridge.close()
|
|
|
|
def test_spawn_defaults_term_when_not_set(self):
|
|
# The bridge should set TERM=xterm-256color when the caller's env
|
|
# doesn't already carry one — xterm.js expects ANSI/SGR sequences.
|
|
env = {k: v for k, v in os.environ.items() if k.upper() != "TERM"}
|
|
bridge = WinPtyBridge.spawn(
|
|
[
|
|
sys.executable,
|
|
"-c",
|
|
"import os; print('TERM=' + os.environ.get('TERM',''))",
|
|
],
|
|
env=env,
|
|
)
|
|
try:
|
|
output = _read_until(bridge, b"TERM=")
|
|
assert b"TERM=xterm-256color" in output
|
|
finally:
|
|
bridge.close()
|