Files
Timmy-time-dashboard/tests/timmy/test_mcp_bridge.py
Claude (Opus 4.6) f2a277f7b5
Some checks failed
Tests / lint (push) Has been cancelled
Tests / test (push) Has been cancelled
[claude] Add vllm-mlx as high-performance local inference backend (#1069) (#1089)
Co-authored-by: Claude (Opus 4.6) <claude@hermes.local>
Co-committed-by: Claude (Opus 4.6) <claude@hermes.local>
2026-03-23 15:34:13 +00:00

608 lines
21 KiB
Python

"""Tests for the MCP bridge module (Qwen3 via Ollama)."""
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
from timmy.mcp_bridge import (
BridgeResult,
MCPBridge,
MCPToolDef,
_build_gitea_tools,
_build_shell_tool,
_mcp_schema_to_ollama_tool,
)
# ---------------------------------------------------------------------------
# _mcp_schema_to_ollama_tool
# ---------------------------------------------------------------------------
def test_schema_to_ollama_tool_basic():
"""Converts an MCPToolDef to Ollama tool format."""
tool = MCPToolDef(
name="test_tool",
description="A test tool",
parameters={
"type": "object",
"properties": {"arg1": {"type": "string"}},
"required": ["arg1"],
},
handler=AsyncMock(),
)
result = _mcp_schema_to_ollama_tool(tool)
assert result["type"] == "function"
assert result["function"]["name"] == "test_tool"
assert result["function"]["description"] == "A test tool"
assert result["function"]["parameters"]["type"] == "object"
assert "arg1" in result["function"]["parameters"]["properties"]
def test_schema_to_ollama_tool_wraps_bare_params():
"""Wraps bare parameter dicts in an object type."""
tool = MCPToolDef(
name="bare",
description="Bare params",
parameters={"x": {"type": "integer"}},
handler=AsyncMock(),
)
result = _mcp_schema_to_ollama_tool(tool)
params = result["function"]["parameters"]
assert params["type"] == "object"
assert "x" in params["properties"]
assert "x" in params["required"]
# ---------------------------------------------------------------------------
# _build_shell_tool
# ---------------------------------------------------------------------------
def test_build_shell_tool_returns_def():
"""Shell tool builder returns an MCPToolDef."""
tool = _build_shell_tool()
assert tool is not None
assert tool.name == "shell_exec"
assert "command" in tool.parameters["properties"]
def test_build_shell_tool_graceful_on_import_error():
"""Shell tool returns None when infrastructure is unavailable."""
with patch.dict("sys.modules", {"infrastructure.hands.shell": None}):
# Force re-import failure — but _build_shell_tool catches it
with patch(
"timmy.mcp_bridge._build_shell_tool",
wraps=_build_shell_tool,
):
# The real function should handle import errors
tool = _build_shell_tool()
# May return tool if import cache succeeds, or None if not
# Just verify it doesn't raise
assert tool is None or isinstance(tool, MCPToolDef)
# ---------------------------------------------------------------------------
# _build_gitea_tools
# ---------------------------------------------------------------------------
def test_gitea_tools_empty_when_disabled():
"""Gitea tools returns empty list when disabled."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.gitea_enabled = False
mock_settings.gitea_token = ""
result = _build_gitea_tools()
assert result == []
def test_gitea_tools_empty_when_no_token():
"""Gitea tools returns empty list when no token."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.gitea_enabled = True
mock_settings.gitea_token = ""
result = _build_gitea_tools()
assert result == []
def test_gitea_tools_returns_three_tools():
"""Gitea tools returns list_issues, create_issue, read_issue."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.gitea_enabled = True
mock_settings.gitea_token = "tok123"
mock_settings.gitea_url = "http://localhost:3000"
mock_settings.gitea_repo = "owner/repo"
result = _build_gitea_tools()
assert len(result) == 3
names = {t.name for t in result}
assert names == {"list_issues", "create_issue", "read_issue"}
# ---------------------------------------------------------------------------
# MCPBridge.__init__
# ---------------------------------------------------------------------------
def test_bridge_init_default():
"""MCPBridge initialises with default settings."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.ollama_model = "qwen3:14b"
mock_settings.normalized_ollama_url = "http://localhost:11434"
mock_settings.gitea_enabled = False
mock_settings.gitea_token = ""
bridge = MCPBridge(include_gitea=False, include_shell=False)
assert bridge.model == "qwen3:14b"
assert bridge.tool_names == []
def test_bridge_init_with_extra_tools():
"""MCPBridge accepts extra tool definitions."""
custom = MCPToolDef(
name="custom_tool",
description="Custom",
parameters={"type": "object", "properties": {}, "required": []},
handler=AsyncMock(),
)
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.ollama_model = "qwen3:14b"
mock_settings.normalized_ollama_url = "http://localhost:11434"
mock_settings.gitea_enabled = False
mock_settings.gitea_token = ""
bridge = MCPBridge(
include_gitea=False,
include_shell=False,
extra_tools=[custom],
)
assert "custom_tool" in bridge.tool_names
# ---------------------------------------------------------------------------
# MCPBridge.run — tool-call loop
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_bridge_run_simple_response():
"""Bridge returns model content when no tool calls are made."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.ollama_model = "qwen3:14b"
mock_settings.normalized_ollama_url = "http://localhost:11434"
mock_settings.ollama_num_ctx = 4096
mock_settings.mcp_bridge_timeout = 60
mock_settings.gitea_enabled = False
mock_settings.gitea_token = ""
bridge = MCPBridge(include_gitea=False, include_shell=False)
mock_resp = MagicMock()
mock_resp.json.return_value = {"message": {"role": "assistant", "content": "Hello!"}}
mock_resp.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.post = AsyncMock(return_value=mock_resp)
mock_client.aclose = AsyncMock()
bridge._client = mock_client
result = await bridge.run("Hi")
assert result.content == "Hello!"
assert result.rounds == 1
assert result.tool_calls_made == []
assert result.error == ""
@pytest.mark.asyncio
async def test_bridge_run_with_tool_call():
"""Bridge executes tool calls and returns final response."""
handler = AsyncMock(return_value="tool result data")
tool = MCPToolDef(
name="my_tool",
description="Test",
parameters={"type": "object", "properties": {}, "required": []},
handler=handler,
)
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.ollama_model = "qwen3:14b"
mock_settings.normalized_ollama_url = "http://localhost:11434"
mock_settings.ollama_num_ctx = 0
mock_settings.mcp_bridge_timeout = 60
mock_settings.gitea_enabled = False
mock_settings.gitea_token = ""
bridge = MCPBridge(
include_gitea=False,
include_shell=False,
extra_tools=[tool],
)
# Round 1: model requests tool call
tool_call_resp = MagicMock()
tool_call_resp.json.return_value = {
"message": {
"role": "assistant",
"content": "",
"tool_calls": [
{
"function": {
"name": "my_tool",
"arguments": {},
}
}
],
}
}
tool_call_resp.raise_for_status = MagicMock()
# Round 2: model returns final text
final_resp = MagicMock()
final_resp.json.return_value = {"message": {"role": "assistant", "content": "Done with tools!"}}
final_resp.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.post = AsyncMock(side_effect=[tool_call_resp, final_resp])
mock_client.aclose = AsyncMock()
bridge._client = mock_client
result = await bridge.run("Do something")
assert result.content == "Done with tools!"
assert result.rounds == 2
assert len(result.tool_calls_made) == 1
assert result.tool_calls_made[0]["tool"] == "my_tool"
handler.assert_awaited_once()
@pytest.mark.asyncio
async def test_bridge_run_unknown_tool():
"""Bridge handles calls to unknown tools gracefully."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.ollama_model = "qwen3:14b"
mock_settings.normalized_ollama_url = "http://localhost:11434"
mock_settings.ollama_num_ctx = 0
mock_settings.mcp_bridge_timeout = 60
mock_settings.gitea_enabled = False
mock_settings.gitea_token = ""
bridge = MCPBridge(include_gitea=False, include_shell=False)
# Model calls a tool that doesn't exist
tool_call_resp = MagicMock()
tool_call_resp.json.return_value = {
"message": {
"role": "assistant",
"content": "",
"tool_calls": [{"function": {"name": "nonexistent", "arguments": {}}}],
}
}
tool_call_resp.raise_for_status = MagicMock()
final_resp = MagicMock()
final_resp.json.return_value = {"message": {"role": "assistant", "content": "OK"}}
final_resp.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.post = AsyncMock(side_effect=[tool_call_resp, final_resp])
mock_client.aclose = AsyncMock()
bridge._client = mock_client
result = await bridge.run("test")
assert len(result.tool_calls_made) == 1
assert "unknown tool" in result.tool_calls_made[0]["result"]
@pytest.mark.asyncio
async def test_bridge_run_max_rounds():
"""Bridge stops after max_rounds and returns error."""
handler = AsyncMock(return_value="result")
tool = MCPToolDef(
name="loop_tool",
description="Loops forever",
parameters={"type": "object", "properties": {}, "required": []},
handler=handler,
)
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.ollama_model = "qwen3:14b"
mock_settings.normalized_ollama_url = "http://localhost:11434"
mock_settings.ollama_num_ctx = 0
mock_settings.mcp_bridge_timeout = 60
mock_settings.gitea_enabled = False
mock_settings.gitea_token = ""
bridge = MCPBridge(
include_gitea=False,
include_shell=False,
extra_tools=[tool],
max_rounds=2,
)
# Always return tool calls (never a final response)
tool_call_resp = MagicMock()
tool_call_resp.json.return_value = {
"message": {
"role": "assistant",
"content": "",
"tool_calls": [{"function": {"name": "loop_tool", "arguments": {}}}],
}
}
tool_call_resp.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.post = AsyncMock(return_value=tool_call_resp)
mock_client.aclose = AsyncMock()
bridge._client = mock_client
result = await bridge.run("loop")
assert "max tool-call rounds" in result.content
assert "Exceeded" in result.error
assert result.rounds == 2
@pytest.mark.asyncio
async def test_bridge_run_connection_error():
"""Bridge handles Ollama connection errors gracefully."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.ollama_model = "qwen3:14b"
mock_settings.normalized_ollama_url = "http://localhost:11434"
mock_settings.ollama_num_ctx = 0
mock_settings.mcp_bridge_timeout = 60
mock_settings.gitea_enabled = False
mock_settings.gitea_token = ""
bridge = MCPBridge(include_gitea=False, include_shell=False)
mock_client = AsyncMock()
mock_client.post = AsyncMock(side_effect=httpx.ConnectError("Connection refused"))
mock_client.aclose = AsyncMock()
bridge._client = mock_client
result = await bridge.run("test")
assert result.error
assert "connection" in result.error.lower()
assert result.content == ""
@pytest.mark.asyncio
async def test_bridge_run_http_error():
"""Bridge handles Ollama HTTP errors gracefully."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.ollama_model = "qwen3:14b"
mock_settings.normalized_ollama_url = "http://localhost:11434"
mock_settings.ollama_num_ctx = 0
mock_settings.mcp_bridge_timeout = 60
mock_settings.gitea_enabled = False
mock_settings.gitea_token = ""
bridge = MCPBridge(include_gitea=False, include_shell=False)
mock_response = MagicMock()
mock_response.status_code = 500
mock_client = AsyncMock()
mock_client.post = AsyncMock(
side_effect=httpx.HTTPStatusError(
"Server Error",
request=MagicMock(),
response=mock_response,
)
)
mock_client.aclose = AsyncMock()
bridge._client = mock_client
result = await bridge.run("test")
assert result.error
assert "500" in result.error
@pytest.mark.asyncio
async def test_bridge_run_without_context_manager():
"""Bridge returns error when used without async context manager."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.ollama_model = "qwen3:14b"
mock_settings.normalized_ollama_url = "http://localhost:11434"
mock_settings.gitea_enabled = False
mock_settings.gitea_token = ""
bridge = MCPBridge(include_gitea=False, include_shell=False)
result = await bridge.run("test")
assert result.error
assert "context manager" in result.error.lower()
# ---------------------------------------------------------------------------
# MCPBridge.status
# ---------------------------------------------------------------------------
def test_bridge_status():
"""Bridge status returns model and tool info."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.ollama_model = "qwen3:14b"
mock_settings.normalized_ollama_url = "http://localhost:11434"
mock_settings.gitea_enabled = False
mock_settings.gitea_token = ""
bridge = MCPBridge(include_gitea=False, include_shell=False)
status = bridge.status()
assert status["model"] == "qwen3:14b"
assert status["connected"] is False
assert isinstance(status["tools"], list)
# ---------------------------------------------------------------------------
# MCPBridge context manager
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_bridge_context_manager():
"""Bridge opens and closes httpx client via async context manager."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.ollama_model = "qwen3:14b"
mock_settings.normalized_ollama_url = "http://localhost:11434"
mock_settings.mcp_bridge_timeout = 60
mock_settings.gitea_enabled = False
mock_settings.gitea_token = ""
bridge = MCPBridge(include_gitea=False, include_shell=False)
assert bridge._client is None
async with bridge:
assert bridge._client is not None
assert bridge._client is None
# ---------------------------------------------------------------------------
# Gitea tool handlers (integration-style, mocked HTTP)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_gitea_list_issues_handler():
"""list_issues handler calls Gitea API and formats results."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.gitea_enabled = True
mock_settings.gitea_token = "tok123"
mock_settings.gitea_url = "http://localhost:3000"
mock_settings.gitea_repo = "owner/repo"
tools = _build_gitea_tools()
list_tool = next(t for t in tools if t.name == "list_issues")
mock_resp = MagicMock()
mock_resp.json.return_value = [
{"number": 1, "title": "Bug one", "labels": [{"name": "bug"}]},
{"number": 2, "title": "Feature two", "labels": []},
]
mock_resp.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.get = AsyncMock(return_value=mock_resp)
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
with patch("timmy.mcp_bridge.httpx.AsyncClient", return_value=mock_client):
result = await list_tool.handler(state="open", limit=10)
assert "#1: Bug one [bug]" in result
assert "#2: Feature two" in result
@pytest.mark.asyncio
async def test_gitea_create_issue_handler():
"""create_issue handler calls Gitea API and returns confirmation."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.gitea_enabled = True
mock_settings.gitea_token = "tok123"
mock_settings.gitea_url = "http://localhost:3000"
mock_settings.gitea_repo = "owner/repo"
tools = _build_gitea_tools()
create_tool = next(t for t in tools if t.name == "create_issue")
mock_resp = MagicMock()
mock_resp.json.return_value = {"number": 42, "title": "New bug"}
mock_resp.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.post = AsyncMock(return_value=mock_resp)
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
with patch("timmy.mcp_bridge.httpx.AsyncClient", return_value=mock_client):
result = await create_tool.handler(title="New bug", body="Description")
assert "#42" in result
assert "New bug" in result
@pytest.mark.asyncio
async def test_gitea_create_issue_requires_title():
"""create_issue handler returns error when title is missing."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.gitea_enabled = True
mock_settings.gitea_token = "tok123"
mock_settings.gitea_url = "http://localhost:3000"
mock_settings.gitea_repo = "owner/repo"
tools = _build_gitea_tools()
create_tool = next(t for t in tools if t.name == "create_issue")
result = await create_tool.handler()
assert "required" in result.lower()
@pytest.mark.asyncio
async def test_gitea_read_issue_handler():
"""read_issue handler calls Gitea API and formats result."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.gitea_enabled = True
mock_settings.gitea_token = "tok123"
mock_settings.gitea_url = "http://localhost:3000"
mock_settings.gitea_repo = "owner/repo"
tools = _build_gitea_tools()
read_tool = next(t for t in tools if t.name == "read_issue")
mock_resp = MagicMock()
mock_resp.json.return_value = {
"number": 5,
"title": "Test issue",
"state": "open",
"body": "Issue body text",
"labels": [{"name": "enhancement"}],
}
mock_resp.raise_for_status = MagicMock()
mock_client = AsyncMock()
mock_client.get = AsyncMock(return_value=mock_resp)
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
with patch("timmy.mcp_bridge.httpx.AsyncClient", return_value=mock_client):
result = await read_tool.handler(number=5)
assert "#5" in result
assert "Test issue" in result
assert "open" in result
assert "enhancement" in result
@pytest.mark.asyncio
async def test_gitea_read_issue_requires_number():
"""read_issue handler returns error when number is missing."""
with patch("timmy.mcp_bridge.settings") as mock_settings:
mock_settings.gitea_enabled = True
mock_settings.gitea_token = "tok123"
mock_settings.gitea_url = "http://localhost:3000"
mock_settings.gitea_repo = "owner/repo"
tools = _build_gitea_tools()
read_tool = next(t for t in tools if t.name == "read_issue")
result = await read_tool.handler()
assert "required" in result.lower()
# ---------------------------------------------------------------------------
# BridgeResult dataclass
# ---------------------------------------------------------------------------
def test_bridge_result_defaults():
"""BridgeResult has sensible defaults."""
r = BridgeResult(content="hello")
assert r.content == "hello"
assert r.tool_calls_made == []
assert r.rounds == 0
assert r.latency_ms == 0.0
assert r.model == ""
assert r.error == ""