mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-06-09 08:21:50 +00:00
feat(windows): enable dashboard /chat tab via ConPTY (win_pty_bridge) + tests (#42251)
* feat(windows): enable dashboard chat tab via ConPTY (win_pty_bridge) Add hermes_cli/win_pty_bridge.py — a pywinpty-backed drop-in for PtyBridge with the same spawn/read/write/resize/close surface — and wire it into the web_server PTY import block so Windows picks it up instead of falling back to None. pywinpty is already a declared win32 dependency (pyproject.toml). The ConPTY read path runs inside run_in_executor so the event loop is never blocked. Spawn/read/write/terminate call shapes are taken directly from tools/process_registry.py which already exercises the same pywinpty version. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * docs: remove WSL2-only caveat for dashboard chat tab The chat pane now works on native Windows via the ConPTY bridge added in the previous commit. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> * test(windows): cover ConPTY bridge + web_server platform-branched import Companion to the bridge added in the previous commits. Verified live on native Windows 11 (pywinpty 2.0.15) against `hermes dashboard`'s `/api/pty` WebSocket: the spawned `hermes --tui` (node entry.js) renders through ConPTY, resize escapes reach `setwinsize`, and closing the WS reaps both the node child and the pywinpty agent with zero orphans. tests/hermes_cli/test_win_pty_bridge.py Mirrors the layout of the existing POSIX test_pty_bridge.py: spawn/io/resize/close/env coverage against cmd.exe and python -c, plus the cross-platform fallback surface (PtyUnavailableError, the off-Windows `spawn -> raises PtyUnavailableError` guard, and the load-bearing _clamp() helper that protects setwinsize from garbage winsize values out of xterm.js). tests/hermes_cli/test_web_server_pty_import.py Asserts that web_server.PtyBridge resolves to WinPtyBridge on win32 and to the POSIX PtyBridge on POSIX, that PtyUnavailableError is the matching class on each side (so isinstance checks in /api/pty's spawn fallback path work), and a source-text check that pins the platform-branched import shape so a future refactor can't quietly collapse it back to a POSIX-only import. scripts/release.py AUTHOR_MAP entries so CI release-note generation can resolve both authors' plain (non-noreply) emails to their GitHub logins. Co-Authored-By: JoelJJohnson <josephjohnson.joel@gmail.com> Co-Authored-By: Nea74 <andreas@schwarz-ketsch.de> --------- Co-authored-by: JoelJJohnson <josephjohnson.joel@gmail.com> Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com> Co-authored-by: Nea74 <andreas@schwarz-ketsch.de>
This commit is contained in:
parent
c6d27addf7
commit
abcf996b1f
6 changed files with 605 additions and 14 deletions
|
|
@ -53,7 +53,7 @@ If you already have Git installed, the installer detects it and uses that instea
|
|||
|
||||
> **Android / Termux:** The tested manual path is documented in the [Termux guide](https://hermes-agent.nousresearch.com/docs/getting-started/termux). On Termux, Hermes installs a curated `.[termux]` extra because the full `.[all]` extra currently pulls Android-incompatible voice dependencies.
|
||||
>
|
||||
> **Windows:** Native Windows is fully supported — the PowerShell one-liner above installs everything. If you'd rather use WSL2, the Linux command works there too. Native Windows install lives under `%LOCALAPPDATA%\hermes`; WSL2 installs under `~/.hermes` as on Linux. The only Hermes feature that currently needs WSL2 specifically is the browser-based dashboard chat pane (it uses a POSIX PTY — classic CLI and gateway both run natively).
|
||||
> **Windows:** Native Windows is fully supported — the PowerShell one-liner above installs everything. If you'd rather use WSL2, the Linux command works there too. Native Windows install lives under `%LOCALAPPDATA%\hermes`; WSL2 installs under `~/.hermes` as on Linux.
|
||||
|
||||
After installation:
|
||||
|
||||
|
|
|
|||
|
|
@ -8288,20 +8288,32 @@ async def get_models_analytics(days: int = 30):
|
|||
# though uvicorn binds to 127.0.0.1.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# PTY bridge is POSIX-only (depends on fcntl/termios/ptyprocess). On native
|
||||
# Windows the import raises; catch and leave PtyBridge=None so the rest of
|
||||
# the dashboard (sessions, jobs, metrics, config editor) still loads and the
|
||||
# /api/pty endpoint cleanly refuses with a WSL-suggested message.
|
||||
try:
|
||||
from hermes_cli.pty_bridge import PtyBridge, PtyUnavailableError
|
||||
_PTY_BRIDGE_AVAILABLE = True
|
||||
except ImportError as _pty_import_err: # pragma: no cover - Windows-only path
|
||||
PtyBridge = None # type: ignore[assignment]
|
||||
_PTY_BRIDGE_AVAILABLE = False
|
||||
# PTY bridge: POSIX uses pty_bridge (fcntl/termios/ptyprocess); native Windows
|
||||
# uses win_pty_bridge (pywinpty/ConPTY, already a declared dependency). Both
|
||||
# expose the same public surface — spawn/read/write/resize/close/is_available —
|
||||
# so the /api/pty WebSocket handler needs no platform guards.
|
||||
if sys.platform.startswith("win"):
|
||||
try:
|
||||
from hermes_cli.win_pty_bridge import WinPtyBridge as PtyBridge, PtyUnavailableError
|
||||
_PTY_BRIDGE_AVAILABLE = True
|
||||
except ImportError: # pragma: no cover - pywinpty missing
|
||||
PtyBridge = None # type: ignore[assignment]
|
||||
_PTY_BRIDGE_AVAILABLE = False
|
||||
|
||||
class PtyUnavailableError(RuntimeError): # type: ignore[no-redef]
|
||||
"""Stub on platforms where pty_bridge can't be imported."""
|
||||
pass
|
||||
class PtyUnavailableError(RuntimeError): # type: ignore[no-redef]
|
||||
"""Stub when win_pty_bridge cannot be imported."""
|
||||
pass
|
||||
else:
|
||||
try:
|
||||
from hermes_cli.pty_bridge import PtyBridge, PtyUnavailableError
|
||||
_PTY_BRIDGE_AVAILABLE = True
|
||||
except ImportError: # pragma: no cover - dev env without ptyprocess
|
||||
PtyBridge = None # type: ignore[assignment]
|
||||
_PTY_BRIDGE_AVAILABLE = False
|
||||
|
||||
class PtyUnavailableError(RuntimeError): # type: ignore[no-redef]
|
||||
"""Stub on platforms where pty_bridge can't be imported."""
|
||||
pass
|
||||
|
||||
_RESIZE_RE = re.compile(rb"\x1b\[RESIZE:(\d+);(\d+)\]")
|
||||
_PTY_READ_CHUNK_TIMEOUT = 0.2
|
||||
|
|
|
|||
179
hermes_cli/win_pty_bridge.py
Normal file
179
hermes_cli/win_pty_bridge.py
Normal file
|
|
@ -0,0 +1,179 @@
|
|||
"""Windows ConPTY bridge for the `hermes dashboard` chat tab.
|
||||
|
||||
Drop-in counterpart to ``hermes_cli.pty_bridge.PtyBridge`` for native
|
||||
Windows. Mirrors the exact public surface the ``/api/pty`` WebSocket
|
||||
handler in ``hermes_cli.web_server`` consumes: ``spawn``, ``read``,
|
||||
``write``, ``resize``, ``close``, ``is_available``, plus the
|
||||
``PtyUnavailableError`` type.
|
||||
|
||||
Backed by ``pywinpty`` (already a declared win32 dependency in
|
||||
pyproject.toml) instead of ``ptyprocess``/``fcntl``/``termios``, none of
|
||||
which exist on native Windows. The read/write/terminate calls here match
|
||||
the working winpty usage already shipping in ``tools/process_registry.py``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from typing import Optional, Sequence
|
||||
|
||||
try:
|
||||
from winpty import PtyProcess # type: ignore
|
||||
_PTY_AVAILABLE = sys.platform.startswith("win")
|
||||
except ImportError: # pragma: no cover - non-Windows or pywinpty missing
|
||||
PtyProcess = None # type: ignore
|
||||
_PTY_AVAILABLE = False
|
||||
|
||||
|
||||
__all__ = ["WinPtyBridge", "PtyUnavailableError"]
|
||||
|
||||
|
||||
# Same clamp ceiling as the POSIX bridge: a broken winsize probe must never
|
||||
# reach the resize call. ConPTY tolerates large values better than ioctl,
|
||||
# but we keep parity to avoid layout surprises.
|
||||
_MIN_DIMENSION = 1
|
||||
_MAX_COLS = 2000
|
||||
_MAX_ROWS = 1000
|
||||
|
||||
|
||||
def _clamp(value: int, maximum: int) -> int:
|
||||
try:
|
||||
n = int(value)
|
||||
except (TypeError, ValueError, OverflowError):
|
||||
return _MIN_DIMENSION
|
||||
if n < _MIN_DIMENSION:
|
||||
return _MIN_DIMENSION
|
||||
if n > maximum:
|
||||
return maximum
|
||||
return n
|
||||
|
||||
|
||||
class PtyUnavailableError(RuntimeError):
|
||||
"""Raised when a PTY cannot be created on this platform."""
|
||||
|
||||
|
||||
class WinPtyBridge:
|
||||
"""pywinpty-backed bridge with the same interface as ``PtyBridge``.
|
||||
|
||||
``web_server`` calls :meth:`read` inside ``run_in_executor``, so a
|
||||
blocking/polling read here never stalls the event loop. ConPTY exposes
|
||||
no selectable fd, so we poll with a short sleep instead of ``select``.
|
||||
"""
|
||||
|
||||
def __init__(self, proc: "PtyProcess") -> None: # type: ignore[name-defined]
|
||||
self._proc = proc
|
||||
self._closed = False
|
||||
|
||||
# -- lifecycle --------------------------------------------------------
|
||||
|
||||
@classmethod
|
||||
def is_available(cls) -> bool:
|
||||
return bool(_PTY_AVAILABLE)
|
||||
|
||||
@classmethod
|
||||
def spawn(
|
||||
cls,
|
||||
argv: Sequence[str],
|
||||
*,
|
||||
cwd: Optional[str] = None,
|
||||
env: Optional[dict] = None,
|
||||
cols: int = 80,
|
||||
rows: int = 24,
|
||||
) -> "WinPtyBridge":
|
||||
if not _PTY_AVAILABLE:
|
||||
if PtyProcess is None:
|
||||
raise PtyUnavailableError(
|
||||
"pywinpty is not installed. Install with: pip install pywinpty"
|
||||
)
|
||||
raise PtyUnavailableError("ConPTY is unavailable on this platform.")
|
||||
spawn_env = (os.environ.copy() if env is None else dict(env))
|
||||
if not spawn_env.get("TERM"):
|
||||
spawn_env["TERM"] = "xterm-256color"
|
||||
# pywinpty mirrors ptyprocess: dimensions=(rows, cols).
|
||||
# This call shape is the one already used in tools/process_registry.py.
|
||||
proc = PtyProcess.spawn( # type: ignore[union-attr]
|
||||
list(argv),
|
||||
cwd=cwd,
|
||||
env=spawn_env,
|
||||
dimensions=(rows, cols),
|
||||
)
|
||||
return cls(proc)
|
||||
|
||||
@property
|
||||
def pid(self) -> int:
|
||||
return int(self._proc.pid)
|
||||
|
||||
def is_alive(self) -> bool:
|
||||
if self._closed:
|
||||
return False
|
||||
try:
|
||||
return bool(self._proc.isalive())
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
# -- I/O --------------------------------------------------------------
|
||||
|
||||
def read(self, timeout: float = 0.2) -> Optional[bytes]:
|
||||
"""Up to 64 KiB of child output.
|
||||
|
||||
Returns bytes, ``b""`` when nothing is available this tick, or
|
||||
``None`` once the child has exited (EOF).
|
||||
"""
|
||||
if self._closed:
|
||||
return None
|
||||
try:
|
||||
data = self._proc.read(65536) # pywinpty returns str
|
||||
except EOFError:
|
||||
return None
|
||||
except Exception:
|
||||
return None
|
||||
if not data:
|
||||
# No fd to select on; poll politely so the executor thread
|
||||
# doesn't pin a core while the TUI is idle.
|
||||
time.sleep(min(timeout, 0.02))
|
||||
return b""
|
||||
if isinstance(data, bytes):
|
||||
return data
|
||||
# NOTE: pywinpty decodes internally, so a multibyte UTF-8 sequence
|
||||
# can in theory split across reads. xterm.js tolerates the rare
|
||||
# replacement char; this is the one fidelity tradeoff vs the POSIX
|
||||
# raw-fd path.
|
||||
return data.encode("utf-8", errors="replace")
|
||||
|
||||
def write(self, data: bytes) -> None:
|
||||
if self._closed or not data:
|
||||
return
|
||||
try:
|
||||
# The dashboard sends raw keystroke bytes; pywinpty.write wants text.
|
||||
self._proc.write(data.decode("utf-8", errors="replace"))
|
||||
except Exception:
|
||||
return
|
||||
|
||||
def resize(self, cols: int, rows: int) -> None:
|
||||
if self._closed:
|
||||
return
|
||||
cols = _clamp(cols, _MAX_COLS)
|
||||
rows = _clamp(rows, _MAX_ROWS)
|
||||
try:
|
||||
self._proc.setwinsize(rows, cols) # pywinpty: (rows, cols)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# -- teardown ---------------------------------------------------------
|
||||
|
||||
def close(self) -> None:
|
||||
if self._closed:
|
||||
return
|
||||
self._closed = True
|
||||
try:
|
||||
self._proc.terminate(force=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def __enter__(self) -> "WinPtyBridge":
|
||||
return self
|
||||
|
||||
def __exit__(self, *_exc) -> None:
|
||||
self.close()
|
||||
|
|
@ -1490,6 +1490,8 @@ AUTHOR_MAP = {
|
|||
"leonard@sellem.me": "leonardsellem", # PR #37405 (desktop WS origin guard on remote/Tailscale binds)
|
||||
"42903577+ohMyJason@users.noreply.github.com": "ohMyJason", # PR #29810 (discover_models in custom_providers section 4)
|
||||
"singhsanidhya741@gmail.com": "sanidhyasin", # PR #40403 salvage (model.default_headers for custom OpenAI-compatible providers, #40033)
|
||||
"josephjohnson.joel@gmail.com": "JoelJJohnson", # PR #39913 salvage (Windows ConPTY dashboard chat bridge)
|
||||
"andreas@schwarz-ketsch.de": "Nea74", # PR #40022 co-author credit (same Windows ConPTY bridge design)
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
83
tests/hermes_cli/test_web_server_pty_import.py
Normal file
83
tests/hermes_cli/test_web_server_pty_import.py
Normal file
|
|
@ -0,0 +1,83 @@
|
|||
"""Test the platform-branched PTY bridge import in hermes_cli.web_server.
|
||||
|
||||
The /api/pty WebSocket handler in web_server.py picks its bridge at import
|
||||
time via ``sys.platform.startswith("win")`` — Windows gets the ConPTY
|
||||
backend, POSIX gets the fcntl/termios one. Both branches must:
|
||||
|
||||
1. Expose ``PtyBridge`` as the bridge class (or None) and
|
||||
``PtyUnavailableError`` as an exception class.
|
||||
2. Set ``_PTY_BRIDGE_AVAILABLE`` correctly.
|
||||
3. Never raise at import time when the platform-native dependency is
|
||||
missing — the dashboard's non-chat tabs must keep loading.
|
||||
|
||||
This test asserts the live state on whichever platform CI runs on, plus a
|
||||
source-text check confirming the branch shape is preserved so a future
|
||||
refactor can't accidentally collapse it back to a POSIX-only import.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli import web_server
|
||||
|
||||
|
||||
def test_web_server_exposes_pty_bridge_symbols():
|
||||
"""The two symbols /api/pty consumes must always exist."""
|
||||
assert hasattr(web_server, "PtyBridge")
|
||||
assert hasattr(web_server, "PtyUnavailableError")
|
||||
assert hasattr(web_server, "_PTY_BRIDGE_AVAILABLE")
|
||||
# PtyUnavailableError is always an exception class — either the real
|
||||
# one from the platform bridge, or the local fallback class.
|
||||
assert isinstance(web_server.PtyUnavailableError, type)
|
||||
assert issubclass(web_server.PtyUnavailableError, BaseException)
|
||||
|
||||
|
||||
@pytest.mark.skipif(not sys.platform.startswith("win"), reason="Windows-only")
|
||||
def test_web_server_uses_win_pty_bridge_on_windows():
|
||||
"""On native Windows, web_server.PtyBridge must be the ConPTY backend."""
|
||||
from hermes_cli.win_pty_bridge import WinPtyBridge
|
||||
|
||||
assert web_server.PtyBridge is WinPtyBridge
|
||||
assert web_server._PTY_BRIDGE_AVAILABLE is True
|
||||
# And the error class must be the one from the same module so isinstance
|
||||
# checks in /api/pty's spawn fallback path actually work.
|
||||
from hermes_cli.win_pty_bridge import PtyUnavailableError as WinErr
|
||||
|
||||
assert web_server.PtyUnavailableError is WinErr
|
||||
|
||||
|
||||
@pytest.mark.skipif(sys.platform.startswith("win"), reason="POSIX-only")
|
||||
def test_web_server_uses_posix_pty_bridge_on_posix():
|
||||
"""On POSIX, the bridge must be the fcntl/termios PtyBridge."""
|
||||
from hermes_cli.pty_bridge import PtyBridge as PosixBridge
|
||||
from hermes_cli.pty_bridge import PtyUnavailableError as PosixErr
|
||||
|
||||
assert web_server.PtyBridge is PosixBridge
|
||||
assert web_server._PTY_BRIDGE_AVAILABLE is True
|
||||
assert web_server.PtyUnavailableError is PosixErr
|
||||
|
||||
|
||||
def test_pty_bridge_import_block_is_platform_branched():
|
||||
"""Source-level guard: a future refactor must not collapse the branch
|
||||
back to a single POSIX import. Reads web_server.py directly so this
|
||||
fails the same way on every OS — the runtime symbol checks above can
|
||||
pass even when the branch shape is wrong on the current platform."""
|
||||
src = pytest.importorskip("inspect").getsource(web_server)
|
||||
# The shape we expect (from PR #39913):
|
||||
#
|
||||
# if sys.platform.startswith("win"):
|
||||
# try:
|
||||
# from hermes_cli.win_pty_bridge import WinPtyBridge as PtyBridge, ...
|
||||
# except ImportError:
|
||||
# PtyBridge = None
|
||||
# ...
|
||||
# else:
|
||||
# try:
|
||||
# from hermes_cli.pty_bridge import PtyBridge, PtyUnavailableError
|
||||
# ...
|
||||
assert 'sys.platform.startswith("win")' in src or "sys.platform.startswith('win')" in src
|
||||
assert "from hermes_cli.win_pty_bridge import" in src
|
||||
assert "from hermes_cli.pty_bridge import" in src
|
||||
315
tests/hermes_cli/test_win_pty_bridge.py
Normal file
315
tests/hermes_cli/test_win_pty_bridge.py
Normal file
|
|
@ -0,0 +1,315 @@
|
|||
"""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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue