498d3d7061
Implements a lightweight MCP bridge that connects Qwen3 models in Ollama directly to Timmy's tool ecosystem (Gitea, shell) via Ollama's native /api/chat tool-calling API. This complements the existing Agno-based MCP integration with a simpler, standalone path for direct tool calling. - src/timmy/mcp_bridge.py: MCPBridge class with tool-call loop, Gitea tools (list/create/read issues), and shell execution via ShellHand - tests/timmy/test_mcp_bridge.py: 24 unit tests covering tool schema conversion, bridge lifecycle, tool-call loop, error handling - docs/mcp-setup.md: architecture, bridge option evaluation, setup guide - config.py: mcp_bridge_timeout setting (60s default) Fixes #1067 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
620 lines
21 KiB
Python
620 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 == ""
|