forked from Rockachopa/Timmy-time-dashboard
Co-authored-by: Claude (Opus 4.6) <claude@hermes.local> Co-committed-by: Claude (Opus 4.6) <claude@hermes.local>
This commit is contained in:
rockachopa
619
tests/timmy/test_mcp_bridge.py
Normal file
619
tests/timmy/test_mcp_bridge.py
Normal file
@@ -0,0 +1,619 @@
|
||||
"""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 == ""
|
||||
Reference in New Issue
Block a user