wehub-resource-sync
280 lines
10 KiB
Python
280 lines
10 KiB
Python
"""Unit tests for the omni hermes CLI-side helpers + harness wiring (no server)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import click
|
|
import pytest
|
|
|
|
from omnigent import hermes_native as hn
|
|
|
|
|
|
def test_resolve_hermes_executable_found() -> None:
|
|
resolved = hn.resolve_hermes_executable(
|
|
env={}, which=lambda cmd: f"/usr/local/bin/{cmd}" if cmd == "hermes" else None
|
|
)
|
|
assert resolved == "/usr/local/bin/hermes"
|
|
|
|
|
|
def test_resolve_hermes_executable_honors_path_override() -> None:
|
|
resolved = hn.resolve_hermes_executable(
|
|
env={"OMNIGENT_HERMES_PATH": "/opt/hermes"},
|
|
which=lambda cmd: cmd if cmd == "/opt/hermes" else None,
|
|
)
|
|
assert resolved == "/opt/hermes"
|
|
|
|
|
|
def test_resolve_hermes_executable_missing_raises_with_hint() -> None:
|
|
with pytest.raises(click.ClickException) as exc:
|
|
hn.resolve_hermes_executable(env={}, which=lambda _cmd: None)
|
|
assert "hermes-agent.nousresearch.com" in str(exc.value)
|
|
|
|
|
|
def test_build_hermes_launch_argv() -> None:
|
|
launch = hn.build_hermes_launch(
|
|
["--resume", "x"],
|
|
env={},
|
|
which=lambda cmd: f"/bin/{cmd}",
|
|
)
|
|
assert launch.executable == "/bin/hermes"
|
|
assert launch.argv == ["/bin/hermes", "--resume", "x"]
|
|
|
|
|
|
def test_terminal_resource_id_stable() -> None:
|
|
assert hn.hermes_terminal_resource_id() == hn.hermes_terminal_resource_id()
|
|
|
|
|
|
def test_harness_registry_has_hermes_native() -> None:
|
|
from omnigent.runtime.harnesses import _HARNESS_MODULES
|
|
|
|
assert _HARNESS_MODULES["hermes-native"] == "omnigent.inner.hermes_native_harness"
|
|
|
|
|
|
def test_alias_and_native_membership() -> None:
|
|
from omnigent.harness_aliases import (
|
|
NATIVE_HARNESSES,
|
|
canonicalize_harness,
|
|
is_native_harness,
|
|
)
|
|
|
|
assert canonicalize_harness("native-hermes") == "hermes-native"
|
|
assert "hermes-native" in NATIVE_HARNESSES
|
|
assert "native-hermes" in NATIVE_HARNESSES
|
|
assert is_native_harness("hermes-native") is True
|
|
assert is_native_harness("native-hermes") is True
|
|
# The headless ``hermes`` harness is NOT a native CLI harness.
|
|
assert is_native_harness("hermes") is False
|
|
|
|
|
|
def test_native_coding_agent_resolves() -> None:
|
|
from omnigent._wrapper_labels import (
|
|
HERMES_NATIVE_WRAPPER_VALUE,
|
|
UI_MODE_LABEL_KEY,
|
|
UI_MODE_TERMINAL_VALUE,
|
|
WRAPPER_LABEL_KEY,
|
|
)
|
|
from omnigent.harness_plugins import HERMES_NATIVE_CODING_AGENT
|
|
from omnigent.native_coding_agents import native_coding_agent_for_harness
|
|
|
|
agent = native_coding_agent_for_harness("native-hermes")
|
|
assert agent is HERMES_NATIVE_CODING_AGENT
|
|
assert agent is native_coding_agent_for_harness("hermes-native")
|
|
assert agent.agent_name == "hermes-native-ui"
|
|
assert agent.terminal_name == "hermes"
|
|
assert agent.presentation_labels == {
|
|
UI_MODE_LABEL_KEY: UI_MODE_TERMINAL_VALUE,
|
|
WRAPPER_LABEL_KEY: HERMES_NATIVE_WRAPPER_VALUE,
|
|
}
|
|
|
|
|
|
def test_create_app_builds() -> None:
|
|
from omnigent.inner.hermes_native_harness import create_app
|
|
|
|
assert create_app() is not None
|
|
|
|
|
|
# --- CLI orchestration helpers (no server/daemon needed) ----------------------
|
|
|
|
|
|
def test_materialize_agent_spec_is_terminal_first_hermes_native(tmp_path, monkeypatch) -> None:
|
|
import yaml
|
|
|
|
# Pin the host shells so the declared terminals are deterministic: zsh is
|
|
# $SHELL (default, first), bash/fish resolve on PATH.
|
|
monkeypatch.setattr("shutil.which", lambda name: f"/usr/bin/{name}")
|
|
monkeypatch.setenv("SHELL", "/bin/zsh")
|
|
spec_path = hn._materialize_hermes_agent_spec(tmp_path)
|
|
raw = yaml.safe_load(spec_path.read_text())
|
|
assert raw["name"] == "hermes-native-ui"
|
|
assert raw["executor"] == {"harness": "hermes-native"}
|
|
assert raw["spawn"] is True
|
|
# One terminal per installed shell, $SHELL first — a non-empty block also
|
|
# gates the MCP relay's sys_terminal_* advertisement.
|
|
assert list(raw["terminals"]) == ["zsh", "bash", "fish"]
|
|
assert raw["terminals"]["zsh"]["command"] == "zsh"
|
|
|
|
|
|
def test_configured_hermes_command_default_and_override() -> None:
|
|
assert hn._configured_hermes_command({}) == "hermes"
|
|
assert hn._configured_hermes_command({"OMNIGENT_HERMES_PATH": "/opt/hermes"}) == "/opt/hermes"
|
|
|
|
|
|
def test_launched_terminal_from_payload_decodes_tmux_metadata() -> None:
|
|
term = hn._launched_hermes_terminal_from_payload(
|
|
{"id": "terminal_hermes_main", "metadata": {"tmux_socket": "/s", "tmux_target": "t:0.0"}}
|
|
)
|
|
assert term.terminal_id == "terminal_hermes_main"
|
|
assert str(term.tmux_socket) == "/s"
|
|
assert term.tmux_target == "t:0.0"
|
|
# No metadata → id only, sockets None.
|
|
bare = hn._launched_hermes_terminal_from_payload({"id": "terminal_hermes_main"})
|
|
assert bare.tmux_socket is None and bare.tmux_target is None
|
|
|
|
|
|
def test_launched_terminal_from_payload_rejects_bad_shapes() -> None:
|
|
with pytest.raises(click.ClickException):
|
|
hn._launched_hermes_terminal_from_payload(["not", "a", "dict"])
|
|
with pytest.raises(click.ClickException):
|
|
hn._launched_hermes_terminal_from_payload({"metadata": {}}) # no id
|
|
|
|
|
|
def test_direct_tmux_unavailable_reason_branches(tmp_path, monkeypatch) -> None:
|
|
def _prep(socket, target):
|
|
return hn.PreparedHermesTerminal(
|
|
session_id="c",
|
|
terminal_id="t",
|
|
tmux_socket=socket,
|
|
tmux_target=target,
|
|
reattached=False,
|
|
)
|
|
|
|
assert "socket path" in (hn._direct_tmux_unavailable_reason(_prep(None, "t")) or "")
|
|
assert "tmux target" in (hn._direct_tmux_unavailable_reason(_prep(tmp_path, None)) or "")
|
|
missing = tmp_path / "nope.sock"
|
|
assert "not reachable" in (hn._direct_tmux_unavailable_reason(_prep(missing, "t")) or "")
|
|
# Socket exists + tmux present → no reason (attach is available).
|
|
sock = tmp_path / "live.sock"
|
|
sock.write_text("")
|
|
monkeypatch.setattr(hn.shutil, "which", lambda _c: "/usr/bin/tmux")
|
|
assert hn._direct_tmux_unavailable_reason(_prep(sock, "t")) is None
|
|
# tmux absent → reason.
|
|
monkeypatch.setattr(hn.shutil, "which", lambda _c: None)
|
|
assert "tmux is not available" in (hn._direct_tmux_unavailable_reason(_prep(sock, "t")) or "")
|
|
|
|
|
|
def test_preflight_requires_tmux(monkeypatch) -> None:
|
|
monkeypatch.setattr(hn.shutil, "which", lambda _c: None)
|
|
with pytest.raises(click.ClickException, match="tmux"):
|
|
hn._preflight_local_tools()
|
|
monkeypatch.setattr(hn.shutil, "which", lambda _c: "/usr/bin/tmux")
|
|
hn._preflight_local_tools() # no raise
|
|
|
|
|
|
def test_update_startup_progress_is_noop_without_renderer() -> None:
|
|
hn._update_startup_progress(None, "hi") # no error
|
|
|
|
class _Progress:
|
|
def __init__(self) -> None:
|
|
self.messages: list[str] = []
|
|
|
|
def update(self, msg: str) -> None:
|
|
self.messages.append(msg)
|
|
|
|
prog = _Progress()
|
|
hn._update_startup_progress(prog, "Starting…")
|
|
assert prog.messages == ["Starting…"]
|
|
|
|
|
|
# --- daemon-flow HTTP helpers (fake async client; no real server) -------------
|
|
|
|
|
|
class _FakeResp:
|
|
def __init__(self, status: int, payload=None, text: str = "") -> None:
|
|
self.status_code = status
|
|
self._payload = payload
|
|
self.text = text
|
|
|
|
def json(self):
|
|
if self._payload is None:
|
|
raise ValueError("no json")
|
|
return self._payload
|
|
|
|
|
|
class _FakeAsyncClient:
|
|
def __init__(self, resp: _FakeResp) -> None:
|
|
self._resp = resp
|
|
|
|
async def post(self, *_a, **_k):
|
|
return self._resp
|
|
|
|
async def get(self, *_a, **_k):
|
|
return self._resp
|
|
|
|
async def patch(self, *_a, **_k):
|
|
return self._resp
|
|
|
|
|
|
async def test_create_hermes_session_returns_id_or_raises() -> None:
|
|
ok = _FakeAsyncClient(_FakeResp(200, {"session_id": "conv_x"}))
|
|
assert await hn._create_hermes_session(ok, b"bundle") == "conv_x"
|
|
with pytest.raises(click.ClickException):
|
|
await hn._create_hermes_session(_FakeAsyncClient(_FakeResp(500, {})), b"bundle")
|
|
with pytest.raises(click.ClickException, match="session_id"):
|
|
await hn._create_hermes_session(_FakeAsyncClient(_FakeResp(200, {})), b"bundle")
|
|
|
|
|
|
async def test_fetch_hermes_session_handles_status() -> None:
|
|
payload = {"labels": {"omnigent.wrapper": "hermes-native-ui"}}
|
|
assert (
|
|
await hn._fetch_hermes_session(_FakeAsyncClient(_FakeResp(200, payload)), "c") == payload
|
|
)
|
|
with pytest.raises(click.ClickException, match="not found"):
|
|
await hn._fetch_hermes_session(_FakeAsyncClient(_FakeResp(404)), "c")
|
|
with pytest.raises(click.ClickException):
|
|
await hn._fetch_hermes_session(_FakeAsyncClient(_FakeResp(500, {})), "c")
|
|
|
|
|
|
async def test_ensure_terminal_on_runner_raises_on_error() -> None:
|
|
await hn._ensure_hermes_terminal_on_runner(_FakeAsyncClient(_FakeResp(200, {})), "c") # ok
|
|
with pytest.raises(click.ClickException):
|
|
await hn._ensure_hermes_terminal_on_runner(_FakeAsyncClient(_FakeResp(500, {})), "c")
|
|
|
|
|
|
async def test_find_running_terminal_states() -> None:
|
|
# 404 → not created yet.
|
|
assert await hn._find_running_hermes_terminal(_FakeAsyncClient(_FakeResp(404)), "c") is None
|
|
# running:false → treated as absent.
|
|
not_running = _FakeResp(200, {"id": "terminal_hermes_main", "metadata": {"running": False}})
|
|
assert await hn._find_running_hermes_terminal(_FakeAsyncClient(not_running), "c") is None
|
|
# 409 not-bound → None (runner not ready), not an error.
|
|
notbound = _FakeResp(409, {"error": {"message": "session not bound to a runner"}})
|
|
assert await hn._find_running_hermes_terminal(_FakeAsyncClient(notbound), "c") is None
|
|
# Live terminal with tmux metadata → decoded.
|
|
live = _FakeResp(
|
|
200,
|
|
{"id": "terminal_hermes_main", "metadata": {"tmux_socket": "/s", "tmux_target": "t:0.0"}},
|
|
)
|
|
term = await hn._find_running_hermes_terminal(_FakeAsyncClient(live), "c")
|
|
assert term is not None and term.tmux_target == "t:0.0"
|
|
|
|
|
|
async def test_wait_for_terminal_ready_found_and_timeout(monkeypatch) -> None:
|
|
live = hn.LaunchedHermesTerminal(terminal_id="t", tmux_socket=None, tmux_target=None)
|
|
|
|
async def _found(_client, _sid):
|
|
return live
|
|
|
|
monkeypatch.setattr(hn, "_find_running_hermes_terminal", _found)
|
|
out = await hn._wait_for_hermes_terminal_ready(
|
|
_FakeAsyncClient(_FakeResp(200, {})), "c", timeout_s=5
|
|
)
|
|
assert out is live
|
|
|
|
async def _never(_client, _sid):
|
|
return None
|
|
|
|
monkeypatch.setattr(hn, "_find_running_hermes_terminal", _never)
|
|
with pytest.raises(click.ClickException, match="did not create"):
|
|
await hn._wait_for_hermes_terminal_ready(
|
|
_FakeAsyncClient(_FakeResp(200, {})), "c", timeout_s=0.0
|
|
)
|