"""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 )