ryan
b1e210de5d
Allowlist homebox-mcp's general item editor so the agent can rename items and edit description, notes, quantity, purchase info, flags and custom fields. Only passed args change; the rest of the item is preserved. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
126 lines
4.2 KiB
Python
126 lines
4.2 KiB
Python
"""Tests for the admin-only Homebox MCP tool wrapping and agent routing."""
|
|
|
|
import os
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
from langchain_core.tools import StructuredTool
|
|
from quart import Quart
|
|
|
|
from utils.homebox_mcp import (
|
|
ADMIN_ONLY_MESSAGE,
|
|
HOMEBOX_TOOL_ALLOWLIST,
|
|
load_homebox_tools,
|
|
select_homebox_tools,
|
|
)
|
|
|
|
ADMIN = {"configurable": {"is_admin": True}}
|
|
NON_ADMIN = {"configurable": {"is_admin": False}}
|
|
|
|
|
|
def _mcp_tool(name: str, result=None, error: Exception | None = None):
|
|
"""Build a stub shaped like a langchain-mcp-adapters tool."""
|
|
calls = []
|
|
|
|
async def call(runtime=None, **arguments):
|
|
calls.append(arguments)
|
|
if error:
|
|
raise error
|
|
return (
|
|
result if result is not None else ([{"type": "text", "text": "ok"}], None)
|
|
)
|
|
|
|
tool = StructuredTool(
|
|
name=name,
|
|
description=f"{name} description",
|
|
args_schema={
|
|
"type": "object",
|
|
"properties": {"query": {"type": "string"}},
|
|
},
|
|
coroutine=call,
|
|
response_format="content_and_artifact",
|
|
)
|
|
return tool, calls
|
|
|
|
|
|
class TestSelectHomeboxTools:
|
|
def test_filters_to_allowlist_and_prefixes(self):
|
|
stubs = [_mcp_tool(n)[0] for n in ("search_items", "delete_item", "get_item")]
|
|
tools = select_homebox_tools(stubs)
|
|
assert [t.name for t in tools] == ["homebox_search_items", "homebox_get_item"]
|
|
|
|
def test_exposes_item_editor(self):
|
|
assert "set_item" in HOMEBOX_TOOL_ALLOWLIST
|
|
|
|
def test_never_exposes_deletes(self):
|
|
assert not any(name.startswith("delete_") for name in HOMEBOX_TOOL_ALLOWLIST)
|
|
|
|
def test_truncates_long_descriptions(self):
|
|
stub, _ = _mcp_tool("search_items")
|
|
stub.description = "word " * 200
|
|
[tool] = select_homebox_tools([stub])
|
|
assert len(tool.description) <= 300
|
|
|
|
|
|
class TestAdminGuard:
|
|
async def test_admin_call_delegates_and_flattens_content(self):
|
|
stub, calls = _mcp_tool(
|
|
"search_items",
|
|
result=(
|
|
[{"type": "text", "text": "Drill"}, {"type": "text", "text": "Garage"}],
|
|
None,
|
|
),
|
|
)
|
|
[tool] = select_homebox_tools([stub])
|
|
out = await tool.ainvoke({"query": "drill"}, config=ADMIN)
|
|
assert out == "Drill\nGarage"
|
|
assert calls == [{"query": "drill"}]
|
|
|
|
async def test_non_admin_is_refused_without_calling_server(self):
|
|
stub, calls = _mcp_tool("search_items")
|
|
[tool] = select_homebox_tools([stub])
|
|
assert (
|
|
await tool.ainvoke({"query": "drill"}, config=NON_ADMIN)
|
|
== ADMIN_ONLY_MESSAGE
|
|
)
|
|
assert await tool.ainvoke({"query": "drill"}) == ADMIN_ONLY_MESSAGE
|
|
assert calls == []
|
|
|
|
async def test_server_errors_are_returned_as_text(self):
|
|
stub, _ = _mcp_tool("search_items", error=RuntimeError("cannot reach Homebox"))
|
|
[tool] = select_homebox_tools([stub])
|
|
out = await tool.ainvoke({"query": "drill"}, config=ADMIN)
|
|
assert out == "Homebox error: cannot reach Homebox"
|
|
|
|
def test_config_not_exposed_in_schema(self):
|
|
[tool] = select_homebox_tools([_mcp_tool("search_items")[0]])
|
|
assert list(tool.tool_call_schema["properties"]) == ["query"]
|
|
|
|
|
|
class TestLoadHomeboxTools:
|
|
async def test_returns_empty_when_unconfigured(self):
|
|
with patch.dict(os.environ, {"HOMEBOX_URL": "", "HOMEBOX_TOKEN": ""}):
|
|
assert await load_homebox_tools() == []
|
|
|
|
async def test_returns_empty_when_server_fails(self):
|
|
env = {
|
|
"HOMEBOX_URL": "http://homebox",
|
|
"HOMEBOX_TOKEN": "hb_x",
|
|
"HOMEBOX_MCP_COMMAND": "/nonexistent/homebox-mcp",
|
|
}
|
|
with patch.dict(os.environ, env):
|
|
assert await load_homebox_tools() == []
|
|
|
|
|
|
class TestGetAgentFor:
|
|
async def test_routes_admin_to_admin_agent(self):
|
|
from blueprints.conversation.logic import get_agent_for
|
|
|
|
app = Quart(__name__)
|
|
app.main_agent, app.admin_agent = "main", "admin"
|
|
admin = SimpleNamespace(is_admin=lambda: True)
|
|
user = SimpleNamespace(is_admin=lambda: False)
|
|
async with app.app_context():
|
|
assert get_agent_for(admin) == "admin"
|
|
assert get_agent_for(user) == "main"
|