254 lines
9.0 KiB
Python
254 lines
9.0 KiB
Python
"""Tests for the hermes-tools-as-MCP server module surface.
|
|
|
|
We don't run a live MCP session in unit tests — that requires the codex
|
|
subprocess + client + an event loop. These tests pin the static
|
|
contract: the module imports, the EXPOSED_TOOLS list is sane, and the
|
|
build helper assembles a server when the SDK is present.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import inspect
|
|
from typing import get_args
|
|
|
|
from agent.transports.hermes_tools_mcp_server import (
|
|
_signature_from_schema,
|
|
)
|
|
|
|
|
|
class TestSignatureFromSchema:
|
|
"""Test the JSON Schema -> Python signature conversion."""
|
|
|
|
def test_simple_required_string_param(self):
|
|
"""A required string param becomes str with no default."""
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {"query": {"type": "string"}},
|
|
"required": ["query"],
|
|
}
|
|
sig, annots = _signature_from_schema(schema)
|
|
|
|
assert len(sig.parameters) == 1
|
|
param = sig.parameters["query"]
|
|
assert param.name == "query"
|
|
assert param.kind == inspect.Parameter.KEYWORD_ONLY
|
|
assert annots["query"] == str
|
|
assert param.default is inspect.Parameter.empty
|
|
|
|
def test_optional_integer_param(self):
|
|
"""An optional param gets Optional[type] with default=None."""
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {"limit": {"type": "integer"}},
|
|
}
|
|
sig, annots = _signature_from_schema(schema)
|
|
|
|
param = sig.parameters["limit"]
|
|
# Optional[type] is type | None in Python 3.10+
|
|
assert param.default is None
|
|
|
|
def test_multiple_params_mixed_required_optional(self):
|
|
"""Mixed required and optional params are handled correctly."""
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
"limit": {"type": "integer"},
|
|
"offset": {"type": "integer"},
|
|
},
|
|
"required": ["query"],
|
|
}
|
|
sig, annots = _signature_from_schema(schema)
|
|
|
|
assert len(sig.parameters) == 3
|
|
|
|
# query: required str
|
|
assert annots["query"] == str
|
|
assert sig.parameters["query"].default is inspect.Parameter.empty
|
|
|
|
# limit: optional int
|
|
assert sig.parameters["limit"].default is None
|
|
|
|
# offset: optional int
|
|
assert sig.parameters["offset"].default is None
|
|
|
|
def test_skip_private_params(self):
|
|
"""Params starting with '_' are excluded from the signature."""
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"query": {"type": "string"},
|
|
"_internal": {"type": "string"},
|
|
},
|
|
"required": ["query", "_internal"],
|
|
}
|
|
sig, annots = _signature_from_schema(schema)
|
|
|
|
assert "_internal" not in sig.parameters
|
|
assert "_internal" not in annots
|
|
assert "query" in sig.parameters
|
|
|
|
def test_all_json_types(self):
|
|
"""All JSON schema types map to correct Python types."""
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {
|
|
"s": {"type": "string"},
|
|
"i": {"type": "integer"},
|
|
"n": {"type": "number"},
|
|
"b": {"type": "boolean"},
|
|
"a": {"type": "array"},
|
|
"o": {"type": "object"},
|
|
},
|
|
"required": ["s", "i", "n", "b", "a", "o"],
|
|
}
|
|
sig, annots = _signature_from_schema(schema)
|
|
|
|
assert annots["s"] == str
|
|
assert annots["i"] == int
|
|
assert annots["n"] == float
|
|
assert annots["b"] == bool
|
|
assert annots["a"] == list
|
|
assert annots["o"] == dict
|
|
|
|
def test_empty_schema(self):
|
|
"""Empty schema returns empty signature."""
|
|
sig, annots = _signature_from_schema(None)
|
|
assert len(sig.parameters) == 0
|
|
assert len(annots) == 0
|
|
|
|
def test_return_annotation_is_str(self):
|
|
"""All generated signatures have str as return type."""
|
|
schema = {
|
|
"type": "object",
|
|
"properties": {"query": {"type": "string"}},
|
|
}
|
|
sig, annots = _signature_from_schema(schema)
|
|
assert sig.return_annotation == str
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class TestModuleSurface:
|
|
def test_module_imports_clean(self):
|
|
from agent.transports import hermes_tools_mcp_server as m
|
|
assert callable(m.main)
|
|
assert callable(m._build_server)
|
|
assert isinstance(m.EXPOSED_TOOLS, tuple)
|
|
assert len(m.EXPOSED_TOOLS) > 0
|
|
|
|
def test_exposed_tools_are_safe_subset(self):
|
|
"""We MUST NOT expose tools codex already has, because codex'
|
|
own builtins are better-integrated with its sandbox + approvals.
|
|
Specifically: no terminal/shell, no read_file/write_file, no
|
|
patch — those are codex's built-in tools."""
|
|
from agent.transports.hermes_tools_mcp_server import EXPOSED_TOOLS
|
|
forbidden = {
|
|
"terminal", "shell", "read_file", "write_file", "patch",
|
|
"search_files", "process",
|
|
}
|
|
leaked = forbidden & set(EXPOSED_TOOLS)
|
|
assert not leaked, (
|
|
f"these tools must NOT be exposed via the codex callback "
|
|
f"because codex has built-in equivalents: {leaked}"
|
|
)
|
|
|
|
def test_expected_hermes_specific_tools_listed(self):
|
|
"""The Hermes-specific tools should be present so users on the
|
|
codex runtime keep access to them."""
|
|
from agent.transports.hermes_tools_mcp_server import EXPOSED_TOOLS
|
|
for required in (
|
|
"web_search",
|
|
"web_extract",
|
|
"browser_navigate",
|
|
"vision_analyze",
|
|
"image_generate",
|
|
"skill_view",
|
|
):
|
|
assert required in EXPOSED_TOOLS, f"missing {required!r}"
|
|
|
|
def test_agent_loop_tools_not_exposed(self):
|
|
"""delegate_task / memory / session_search / todo require the
|
|
running AIAgent context to dispatch, so a stateless MCP callback
|
|
can't drive them. They must NOT be in EXPOSED_TOOLS."""
|
|
from agent.transports.hermes_tools_mcp_server import EXPOSED_TOOLS
|
|
for agent_loop_tool in ("delegate_task", "memory", "session_search", "todo"):
|
|
assert agent_loop_tool not in EXPOSED_TOOLS, (
|
|
f"{agent_loop_tool!r} requires the agent loop context "
|
|
"and can't be reached through a stateless MCP callback"
|
|
)
|
|
|
|
def test_kanban_worker_tools_exposed(self):
|
|
"""Kanban workers run as `hermes chat -q` subprocesses; if they
|
|
come up on the codex_app_server runtime, the worker can do the
|
|
actual work via codex's shell but needs the kanban tools through
|
|
the MCP callback to report back to the kernel. Without these
|
|
tools available, the worker would hang at completion time."""
|
|
from agent.transports.hermes_tools_mcp_server import EXPOSED_TOOLS
|
|
# Worker handoff tools — every dispatched worker uses at least
|
|
# one of {complete, block, comment} to close out its task.
|
|
for worker_tool in (
|
|
"kanban_complete",
|
|
"kanban_block",
|
|
"kanban_comment",
|
|
"kanban_heartbeat",
|
|
):
|
|
assert worker_tool in EXPOSED_TOOLS, (
|
|
f"{worker_tool!r} missing from codex callback — kanban "
|
|
"workers on codex_app_server runtime would hang"
|
|
)
|
|
|
|
def test_kanban_orchestrator_tools_exposed(self):
|
|
"""Orchestrator agents need to dispatch new tasks, query the
|
|
board, and unblock/link tasks. Exposed so an orchestrator on
|
|
codex_app_server can do its job."""
|
|
from agent.transports.hermes_tools_mcp_server import EXPOSED_TOOLS
|
|
for orch_tool in (
|
|
"kanban_create",
|
|
"kanban_show",
|
|
"kanban_list",
|
|
"kanban_unblock",
|
|
"kanban_link",
|
|
):
|
|
assert orch_tool in EXPOSED_TOOLS, (
|
|
f"{orch_tool!r} missing from codex callback"
|
|
)
|
|
|
|
|
|
class TestMain:
|
|
def test_main_returns_2_when_mcp_unavailable(self, monkeypatch):
|
|
"""When the mcp package isn't installed, main() should exit
|
|
cleanly with code 2 and an install hint, not crash."""
|
|
import agent.transports.hermes_tools_mcp_server as m
|
|
|
|
def boom_build(*a, **kw):
|
|
raise ImportError("mcp not installed")
|
|
|
|
monkeypatch.setattr(m, "_build_server", boom_build)
|
|
rc = m.main(["--verbose"])
|
|
assert rc == 2
|
|
|
|
def test_main_handles_keyboard_interrupt(self, monkeypatch):
|
|
import agent.transports.hermes_tools_mcp_server as m
|
|
|
|
class FakeServer:
|
|
def run(self):
|
|
raise KeyboardInterrupt()
|
|
|
|
monkeypatch.setattr(m, "_build_server", lambda: FakeServer())
|
|
rc = m.main([])
|
|
assert rc == 0
|
|
|
|
def test_main_returns_1_on_runtime_error(self, monkeypatch):
|
|
import agent.transports.hermes_tools_mcp_server as m
|
|
|
|
class CrashingServer:
|
|
def run(self):
|
|
raise RuntimeError("boom")
|
|
|
|
monkeypatch.setattr(m, "_build_server", lambda: CrashingServer())
|
|
rc = m.main([])
|
|
assert rc == 1
|