forked from Rockachopa/Timmy-time-dashboard
78167675f2
Replace the bespoke GiteaHand httpx client and tools_gitea.py wrappers with official MCP tool servers (gitea-mcp + filesystem MCP), wired into Agno via MCPTools. Switch all session functions to async (arun/acontinue_run) so MCP tools auto-connect. Delete ~1070 lines of custom Gitea code. - Create src/timmy/mcp_tools.py with MCP factories + standalone issue bridge - Wire MCPTools into agent.py tool list (Gitea + filesystem) - Switch session.py chat/chat_with_tools/continue_chat to async - Update all callers (dashboard routes, Discord vendor, CLI, thinking engine) - Add gitea_token fallback from ~/.config/gitea/token - Add MCP session cleanup to app shutdown hook - Update tool_safety.py for MCP tool names - 11 new tests, all 1417 passing, coverage 74.2% Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
257 lines
8.8 KiB
Python
257 lines
8.8 KiB
Python
"""Tests for the MCP tools module (factories + issue bridge)."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from timmy.mcp_tools import (
|
|
_bridge_to_work_order,
|
|
close_mcp_sessions,
|
|
create_filesystem_mcp_tools,
|
|
create_gitea_issue_via_mcp,
|
|
create_gitea_mcp_tools,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create_gitea_mcp_tools
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_gitea_mcp_returns_none_when_disabled():
|
|
"""Gitea MCP factory returns None when gitea_enabled=False."""
|
|
with patch("timmy.mcp_tools.settings") as mock_settings:
|
|
mock_settings.gitea_enabled = False
|
|
mock_settings.gitea_token = "some-token"
|
|
result = create_gitea_mcp_tools()
|
|
assert result is None
|
|
|
|
|
|
def test_gitea_mcp_returns_none_when_no_token():
|
|
"""Gitea MCP factory returns None when gitea_token is empty."""
|
|
with patch("timmy.mcp_tools.settings") as mock_settings:
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_token = ""
|
|
result = create_gitea_mcp_tools()
|
|
assert result is None
|
|
|
|
|
|
def test_gitea_mcp_returns_tools_when_configured():
|
|
"""Gitea MCP factory returns an MCPTools instance when properly configured."""
|
|
mock_mcp = MagicMock()
|
|
with (
|
|
patch("timmy.mcp_tools.settings") as mock_settings,
|
|
patch("agno.tools.mcp.MCPTools", return_value=mock_mcp) as mock_cls,
|
|
):
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_token = "tok123"
|
|
mock_settings.mcp_gitea_command = "gitea-mcp -t stdio"
|
|
mock_settings.gitea_url = "http://localhost:3000"
|
|
mock_settings.mcp_timeout = 15
|
|
result = create_gitea_mcp_tools()
|
|
|
|
assert result is mock_mcp
|
|
mock_cls.assert_called_once()
|
|
call_kwargs = mock_cls.call_args[1]
|
|
assert call_kwargs["command"] == "gitea-mcp -t stdio"
|
|
assert call_kwargs["env"]["GITEA_ACCESS_TOKEN"] == "tok123"
|
|
assert "create_issue" in call_kwargs["include_tools"]
|
|
|
|
|
|
def test_gitea_mcp_graceful_on_import_error():
|
|
"""Gitea MCP factory returns None if agno.tools.mcp isn't available."""
|
|
with (
|
|
patch("timmy.mcp_tools.settings") as mock_settings,
|
|
patch.dict("sys.modules", {"agno.tools.mcp": None}),
|
|
):
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_token = "tok123"
|
|
# This should gracefully return None (import will fail)
|
|
result = create_gitea_mcp_tools()
|
|
assert result is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create_filesystem_mcp_tools
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_filesystem_mcp_returns_tools():
|
|
"""Filesystem MCP factory returns an MCPTools instance."""
|
|
mock_mcp = MagicMock()
|
|
with (
|
|
patch("timmy.mcp_tools.settings") as mock_settings,
|
|
patch("agno.tools.mcp.MCPTools", return_value=mock_mcp) as mock_cls,
|
|
):
|
|
mock_settings.mcp_filesystem_command = "npx -y @modelcontextprotocol/server-filesystem"
|
|
mock_settings.repo_root = "/home/user/project"
|
|
mock_settings.mcp_timeout = 15
|
|
result = create_filesystem_mcp_tools()
|
|
|
|
assert result is mock_mcp
|
|
call_kwargs = mock_cls.call_args[1]
|
|
assert "/home/user/project" in call_kwargs["command"]
|
|
assert "read_file" in call_kwargs["include_tools"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# create_gitea_issue_via_mcp
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_issue_via_mcp_returns_not_configured():
|
|
"""Issue creation returns message when Gitea is not configured."""
|
|
with patch("timmy.mcp_tools.settings") as mock_settings:
|
|
mock_settings.gitea_enabled = False
|
|
mock_settings.gitea_token = ""
|
|
result = await create_gitea_issue_via_mcp("Test", "Body")
|
|
assert "not configured" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_issue_via_mcp_calls_tool():
|
|
"""Issue creation calls the MCP tool with correct arguments."""
|
|
import timmy.mcp_tools as mcp_mod
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.connect = AsyncMock()
|
|
mock_session.call_tool = AsyncMock(return_value="Issue #42 created")
|
|
mock_session._connected = False
|
|
|
|
with (
|
|
patch("timmy.mcp_tools.settings") as mock_settings,
|
|
patch("agno.tools.mcp.MCPTools", return_value=mock_session),
|
|
):
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_token = "tok123"
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
mock_settings.gitea_url = "http://localhost:3000"
|
|
mock_settings.mcp_gitea_command = "gitea-mcp -t stdio"
|
|
mock_settings.mcp_timeout = 15
|
|
mock_settings.repo_root = "/tmp/test"
|
|
|
|
# Reset module-level cache
|
|
mcp_mod._issue_session = None
|
|
|
|
result = await create_gitea_issue_via_mcp("Bug title", "Bug body", "bug")
|
|
|
|
assert "Bug title" in result
|
|
mock_session.connect.assert_awaited_once()
|
|
mock_session.call_tool.assert_awaited_once()
|
|
call_args = mock_session.call_tool.call_args
|
|
assert call_args[0][0] == "create_issue"
|
|
assert call_args[1]["arguments"]["owner"] == "owner"
|
|
assert call_args[1]["arguments"]["repo"] == "repo"
|
|
|
|
# Clean up
|
|
mcp_mod._issue_session = None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_issue_via_mcp_graceful_failure():
|
|
"""Issue creation returns error string on MCP failure."""
|
|
import timmy.mcp_tools as mcp_mod
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.connect = AsyncMock(side_effect=ConnectionError("no process"))
|
|
mock_session._connected = False
|
|
|
|
with (
|
|
patch("timmy.mcp_tools.settings") as mock_settings,
|
|
patch("agno.tools.mcp.MCPTools", return_value=mock_session),
|
|
):
|
|
mock_settings.gitea_enabled = True
|
|
mock_settings.gitea_token = "tok123"
|
|
mock_settings.gitea_repo = "owner/repo"
|
|
mock_settings.gitea_url = "http://localhost:3000"
|
|
mock_settings.mcp_gitea_command = "gitea-mcp -t stdio"
|
|
mock_settings.mcp_timeout = 15
|
|
mock_settings.repo_root = "/tmp/test"
|
|
|
|
mcp_mod._issue_session = None
|
|
|
|
result = await create_gitea_issue_via_mcp("Test", "Body")
|
|
|
|
assert "Failed" in result
|
|
mcp_mod._issue_session = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _bridge_to_work_order
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_bridge_to_work_order(tmp_path):
|
|
"""Work order bridge creates a record in SQLite."""
|
|
import sqlite3
|
|
|
|
with patch("timmy.mcp_tools.settings") as mock_settings:
|
|
mock_settings.repo_root = str(tmp_path)
|
|
_bridge_to_work_order("Test title", "Test body", "bug")
|
|
|
|
db_path = tmp_path / "data" / "work_orders.db"
|
|
assert db_path.exists()
|
|
|
|
conn = sqlite3.connect(str(db_path))
|
|
rows = conn.execute("SELECT title, category FROM work_orders").fetchall()
|
|
conn.close()
|
|
assert len(rows) == 1
|
|
assert rows[0][0] == "Test title"
|
|
assert rows[0][1] == "bug"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# close_mcp_sessions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_mcp_sessions():
|
|
"""close_mcp_sessions disconnects the cached session."""
|
|
import timmy.mcp_tools as mcp_mod
|
|
|
|
mock_session = MagicMock()
|
|
mock_session.disconnect = AsyncMock()
|
|
mcp_mod._issue_session = mock_session
|
|
|
|
await close_mcp_sessions()
|
|
|
|
mock_session.disconnect.assert_awaited_once()
|
|
assert mcp_mod._issue_session is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_close_mcp_sessions_noop_when_none():
|
|
"""close_mcp_sessions is a no-op when no session exists."""
|
|
import timmy.mcp_tools as mcp_mod
|
|
|
|
mcp_mod._issue_session = None
|
|
await close_mcp_sessions() # Should not raise
|
|
assert mcp_mod._issue_session is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tool safety integration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_mcp_tools_classified_in_safety():
|
|
"""MCP tool names are correctly classified in tool_safety."""
|
|
from timmy.tool_safety import DANGEROUS_TOOLS, SAFE_TOOLS, requires_confirmation
|
|
|
|
# Gitea MCP tools should be safe
|
|
assert "create_issue" in SAFE_TOOLS
|
|
assert "list_repo_issues" in SAFE_TOOLS
|
|
|
|
# Filesystem read-only MCP tools should be safe
|
|
assert "list_directory" in SAFE_TOOLS
|
|
assert "search_files" in SAFE_TOOLS
|
|
|
|
# write_file is dangerous (filesystem MCP)
|
|
assert "write_file" in DANGEROUS_TOOLS
|
|
|
|
# Verify requires_confirmation logic
|
|
assert not requires_confirmation("create_issue")
|
|
assert not requires_confirmation("list_directory")
|
|
assert requires_confirmation("write_file")
|