forked from cardosofelipe/fast-next-template
Felipe Cardoso
9dfa76aa41
Implements the Git Operations MCP server (Issue #58) providing: Core features: - GitPython wrapper for local repository operations (clone, commit, push, pull, diff, log) - Branch management (create, delete, list, checkout) - Workspace isolation per project with file-based locking - Gitea provider for remote PR operations MCP Tools (17 registered): - clone_repository, git_status, create_branch, list_branches - checkout, commit, push, pull, diff, log - create_pull_request, get_pull_request, list_pull_requests - merge_pull_request, get_workspace, lock_workspace, unlock_workspace Technical details: - FastMCP + FastAPI with JSON-RPC 2.0 protocol - pydantic-settings for configuration (env prefix: GIT_OPS_) - Comprehensive error hierarchy with structured codes - 131 tests passing with 67% coverage - Async operations via ThreadPoolExecutor Closes: #105, #106, #107, #108, #109 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
515 lines
16 KiB
Python
515 lines
16 KiB
Python
"""
|
|
Tests for the MCP server and tools.
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from exceptions import ErrorCode
|
|
|
|
|
|
class TestInputValidation:
|
|
"""Tests for input validation functions."""
|
|
|
|
def test_validate_id_valid(self):
|
|
"""Test valid IDs pass validation."""
|
|
from server import _validate_id
|
|
|
|
assert _validate_id("test-123", "project_id") is None
|
|
assert _validate_id("my_project", "project_id") is None
|
|
assert _validate_id("Agent-001", "agent_id") is None
|
|
|
|
def test_validate_id_empty(self):
|
|
"""Test empty ID fails validation."""
|
|
from server import _validate_id
|
|
|
|
error = _validate_id("", "project_id")
|
|
assert error is not None
|
|
assert "required" in error.lower()
|
|
|
|
def test_validate_id_too_long(self):
|
|
"""Test too-long ID fails validation."""
|
|
from server import _validate_id
|
|
|
|
error = _validate_id("a" * 200, "project_id")
|
|
assert error is not None
|
|
assert "1-128" in error
|
|
|
|
def test_validate_id_invalid_chars(self):
|
|
"""Test invalid characters fail validation."""
|
|
from server import _validate_id
|
|
|
|
assert _validate_id("test@invalid", "project_id") is not None
|
|
assert _validate_id("test!project", "project_id") is not None
|
|
assert _validate_id("test project", "project_id") is not None
|
|
|
|
def test_validate_branch_valid(self):
|
|
"""Test valid branch names."""
|
|
from server import _validate_branch
|
|
|
|
assert _validate_branch("main") is None
|
|
assert _validate_branch("feature/new-thing") is None
|
|
assert _validate_branch("release-1.0.0") is None
|
|
assert _validate_branch("hotfix.urgent") is None
|
|
|
|
def test_validate_branch_invalid(self):
|
|
"""Test invalid branch names."""
|
|
from server import _validate_branch
|
|
|
|
assert _validate_branch("") is not None
|
|
assert _validate_branch("a" * 300) is not None
|
|
|
|
def test_validate_url_valid(self):
|
|
"""Test valid repository URLs."""
|
|
from server import _validate_url
|
|
|
|
assert _validate_url("https://github.com/owner/repo.git") is None
|
|
assert _validate_url("https://gitea.example.com/owner/repo") is None
|
|
assert _validate_url("git@github.com:owner/repo.git") is None
|
|
|
|
def test_validate_url_invalid(self):
|
|
"""Test invalid repository URLs."""
|
|
from server import _validate_url
|
|
|
|
assert _validate_url("") is not None
|
|
assert _validate_url("not-a-url") is not None
|
|
assert _validate_url("ftp://invalid.com/repo") is not None
|
|
|
|
|
|
class TestHealthCheck:
|
|
"""Tests for health check endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_check_structure(self):
|
|
"""Test health check returns proper structure."""
|
|
from server import health_check
|
|
|
|
with patch("server._gitea_provider", None), \
|
|
patch("server._workspace_manager", None):
|
|
result = await health_check()
|
|
|
|
assert "status" in result
|
|
assert "service" in result
|
|
assert "version" in result
|
|
assert "timestamp" in result
|
|
assert "dependencies" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_check_no_providers(self):
|
|
"""Test health check without providers configured."""
|
|
from server import health_check
|
|
|
|
with patch("server._gitea_provider", None), \
|
|
patch("server._workspace_manager", None):
|
|
result = await health_check()
|
|
|
|
assert result["dependencies"]["gitea"] == "not configured"
|
|
|
|
|
|
class TestToolRegistry:
|
|
"""Tests for tool registration."""
|
|
|
|
def test_tool_registry_populated(self):
|
|
"""Test that tools are registered."""
|
|
from server import _tool_registry
|
|
|
|
assert len(_tool_registry) > 0
|
|
assert "clone_repository" in _tool_registry
|
|
assert "git_status" in _tool_registry
|
|
assert "create_branch" in _tool_registry
|
|
assert "commit" in _tool_registry
|
|
|
|
def test_tool_schema_structure(self):
|
|
"""Test tool schemas have proper structure."""
|
|
from server import _tool_registry
|
|
|
|
for name, info in _tool_registry.items():
|
|
assert "func" in info
|
|
assert "description" in info
|
|
assert "schema" in info
|
|
assert info["schema"]["type"] == "object"
|
|
assert "properties" in info["schema"]
|
|
|
|
|
|
class TestCloneRepository:
|
|
"""Tests for clone_repository tool."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clone_invalid_project_id(self):
|
|
"""Test clone with invalid project ID."""
|
|
from server import clone_repository
|
|
|
|
# Access the underlying function via .fn
|
|
result = await clone_repository.fn(
|
|
project_id="invalid@id",
|
|
agent_id="agent-1",
|
|
repo_url="https://github.com/owner/repo.git",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert "project_id" in result["error"].lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clone_invalid_repo_url(self):
|
|
"""Test clone with invalid repo URL."""
|
|
from server import clone_repository
|
|
|
|
result = await clone_repository.fn(
|
|
project_id="valid-project",
|
|
agent_id="agent-1",
|
|
repo_url="not-a-valid-url",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert "url" in result["error"].lower()
|
|
|
|
|
|
class TestGitStatus:
|
|
"""Tests for git_status tool."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_workspace_not_found(self):
|
|
"""Test status when workspace doesn't exist."""
|
|
from server import git_status
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=None)
|
|
|
|
with patch("server._workspace_manager", mock_manager):
|
|
result = await git_status.fn(
|
|
project_id="nonexistent",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["code"] == ErrorCode.WORKSPACE_NOT_FOUND.value
|
|
|
|
|
|
class TestBranchOperations:
|
|
"""Tests for branch operation tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_branch_invalid_name(self):
|
|
"""Test creating branch with invalid name."""
|
|
from server import create_branch
|
|
|
|
result = await create_branch.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
branch_name="", # Invalid
|
|
)
|
|
|
|
assert result["success"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_branches_workspace_not_found(self):
|
|
"""Test listing branches when workspace doesn't exist."""
|
|
from server import list_branches
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=None)
|
|
|
|
with patch("server._workspace_manager", mock_manager):
|
|
result = await list_branches.fn(
|
|
project_id="nonexistent",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_checkout_invalid_project(self):
|
|
"""Test checkout with invalid project ID."""
|
|
from server import checkout
|
|
|
|
result = await checkout.fn(
|
|
project_id="inv@lid",
|
|
agent_id="agent-1",
|
|
ref="main",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
|
|
|
|
class TestCommitOperations:
|
|
"""Tests for commit operation tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_commit_invalid_project(self):
|
|
"""Test commit with invalid project ID."""
|
|
from server import commit
|
|
|
|
result = await commit.fn(
|
|
project_id="inv@lid",
|
|
agent_id="agent-1",
|
|
message="Test commit",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
|
|
|
|
class TestPushPullOperations:
|
|
"""Tests for push/pull operation tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_push_workspace_not_found(self):
|
|
"""Test push when workspace doesn't exist."""
|
|
from server import push
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=None)
|
|
|
|
with patch("server._workspace_manager", mock_manager):
|
|
result = await push.fn(
|
|
project_id="nonexistent",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pull_workspace_not_found(self):
|
|
"""Test pull when workspace doesn't exist."""
|
|
from server import pull
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=None)
|
|
|
|
with patch("server._workspace_manager", mock_manager):
|
|
result = await pull.fn(
|
|
project_id="nonexistent",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
|
|
|
|
class TestDiffLogOperations:
|
|
"""Tests for diff and log operation tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_diff_workspace_not_found(self):
|
|
"""Test diff when workspace doesn't exist."""
|
|
from server import diff
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=None)
|
|
|
|
with patch("server._workspace_manager", mock_manager):
|
|
result = await diff.fn(
|
|
project_id="nonexistent",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_workspace_not_found(self):
|
|
"""Test log when workspace doesn't exist."""
|
|
from server import log
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=None)
|
|
|
|
with patch("server._workspace_manager", mock_manager):
|
|
result = await log.fn(
|
|
project_id="nonexistent",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
|
|
|
|
class TestPROperations:
|
|
"""Tests for pull request operation tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_pr_no_repo_url(self):
|
|
"""Test create PR when workspace has no repo URL."""
|
|
from models import WorkspaceInfo, WorkspaceState
|
|
from server import create_pull_request
|
|
|
|
mock_workspace = WorkspaceInfo(
|
|
project_id="test-project",
|
|
path="/tmp/test",
|
|
state=WorkspaceState.READY,
|
|
repo_url=None, # No repo URL
|
|
)
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace)
|
|
|
|
with patch("server._workspace_manager", mock_manager):
|
|
result = await create_pull_request.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
title="Test PR",
|
|
source_branch="feature",
|
|
target_branch="main",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert "repository URL" in result["error"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_prs_invalid_state(self):
|
|
"""Test list PRs with invalid state filter."""
|
|
from models import WorkspaceInfo, WorkspaceState
|
|
from server import list_pull_requests
|
|
|
|
mock_workspace = WorkspaceInfo(
|
|
project_id="test-project",
|
|
path="/tmp/test",
|
|
state=WorkspaceState.READY,
|
|
repo_url="https://gitea.test.com/owner/repo.git",
|
|
)
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace)
|
|
|
|
mock_provider = AsyncMock()
|
|
mock_provider.parse_repo_url = MagicMock(return_value=("owner", "repo"))
|
|
|
|
with patch("server._workspace_manager", mock_manager), \
|
|
patch("server._get_provider_for_url", return_value=mock_provider):
|
|
result = await list_pull_requests.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
state="invalid-state",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert "Invalid state" in result["error"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_merge_pr_invalid_strategy(self):
|
|
"""Test merge PR with invalid strategy."""
|
|
from models import WorkspaceInfo, WorkspaceState
|
|
from server import merge_pull_request
|
|
|
|
mock_workspace = WorkspaceInfo(
|
|
project_id="test-project",
|
|
path="/tmp/test",
|
|
state=WorkspaceState.READY,
|
|
repo_url="https://gitea.test.com/owner/repo.git",
|
|
)
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace)
|
|
|
|
mock_provider = AsyncMock()
|
|
mock_provider.parse_repo_url = MagicMock(return_value=("owner", "repo"))
|
|
|
|
with patch("server._workspace_manager", mock_manager), \
|
|
patch("server._get_provider_for_url", return_value=mock_provider):
|
|
result = await merge_pull_request.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
pr_number=42,
|
|
merge_strategy="invalid-strategy",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert "Invalid strategy" in result["error"]
|
|
|
|
|
|
class TestWorkspaceOperations:
|
|
"""Tests for workspace operation tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_workspace_not_found(self):
|
|
"""Test get workspace when it doesn't exist."""
|
|
from server import get_workspace
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=None)
|
|
|
|
with patch("server._workspace_manager", mock_manager):
|
|
result = await get_workspace.fn(
|
|
project_id="nonexistent",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lock_workspace_success(self):
|
|
"""Test successful workspace locking."""
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from models import WorkspaceInfo, WorkspaceState
|
|
from server import lock_workspace
|
|
|
|
mock_workspace = WorkspaceInfo(
|
|
project_id="test-project",
|
|
path="/tmp/test",
|
|
state=WorkspaceState.LOCKED,
|
|
lock_holder="agent-1",
|
|
lock_expires=datetime.now(UTC) + timedelta(seconds=300),
|
|
)
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.lock_workspace = AsyncMock(return_value=True)
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace)
|
|
|
|
with patch("server._workspace_manager", mock_manager):
|
|
result = await lock_workspace.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
timeout=300,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["lock_holder"] == "agent-1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unlock_workspace_success(self):
|
|
"""Test successful workspace unlocking."""
|
|
from server import unlock_workspace
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.unlock_workspace = AsyncMock(return_value=True)
|
|
|
|
with patch("server._workspace_manager", mock_manager):
|
|
result = await unlock_workspace.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
|
|
|
|
class TestJSONRPCEndpoint:
|
|
"""Tests for the JSON-RPC endpoint."""
|
|
|
|
def test_python_type_to_json_schema_str(self):
|
|
"""Test string type conversion."""
|
|
from server import _python_type_to_json_schema
|
|
|
|
result = _python_type_to_json_schema(str)
|
|
assert result["type"] == "string"
|
|
|
|
def test_python_type_to_json_schema_int(self):
|
|
"""Test int type conversion."""
|
|
from server import _python_type_to_json_schema
|
|
|
|
result = _python_type_to_json_schema(int)
|
|
assert result["type"] == "integer"
|
|
|
|
def test_python_type_to_json_schema_bool(self):
|
|
"""Test bool type conversion."""
|
|
from server import _python_type_to_json_schema
|
|
|
|
result = _python_type_to_json_schema(bool)
|
|
assert result["type"] == "boolean"
|
|
|
|
def test_python_type_to_json_schema_list(self):
|
|
"""Test list type conversion."""
|
|
|
|
from server import _python_type_to_json_schema
|
|
|
|
result = _python_type_to_json_schema(list[str])
|
|
assert result["type"] == "array"
|
|
assert result["items"]["type"] == "string"
|