forked from cardosofelipe/fast-next-template
Felipe Cardoso
0a624a94af
- Introduced extensive test coverage for FastAPI endpoints, including health check, MCP tools, and JSON-RPC operations. - Added tests for Git operations MCP tools, including cloning, status, branching, committing, and provider detection. - Mocked dependencies and ensured reliable test isolation with unittest.mock and pytest fixtures. - Validated error handling, workspace management, tool execution, and type conversion functions.
1171 lines
38 KiB
Python
1171 lines
38 KiB
Python
"""
|
|
Comprehensive tests for server MCP tools.
|
|
|
|
Tests the actual tool execution with mocked dependencies.
|
|
"""
|
|
|
|
from datetime import UTC, datetime
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from exceptions import ErrorCode
|
|
from models import CreatePRResult, GetPRResult, ListPRsResult, MergePRResult
|
|
from server import (
|
|
_get_auth_token_for_url,
|
|
_get_provider_for_url,
|
|
_validate_branch,
|
|
_validate_id,
|
|
_validate_url,
|
|
checkout,
|
|
clone_repository,
|
|
commit,
|
|
create_branch,
|
|
create_pull_request,
|
|
diff,
|
|
get_pull_request,
|
|
get_workspace,
|
|
git_status,
|
|
list_branches,
|
|
list_pull_requests,
|
|
lock_workspace,
|
|
log,
|
|
merge_pull_request,
|
|
pull,
|
|
push,
|
|
unlock_workspace,
|
|
)
|
|
from workspace import WorkspaceInfo, WorkspaceState
|
|
|
|
|
|
def make_clone_result(workspace_path="/tmp/test", branch="main", commit_sha="abc123"):
|
|
"""Create a mock clone result."""
|
|
result = MagicMock()
|
|
result.workspace_path = workspace_path
|
|
result.branch = branch
|
|
result.commit_sha = commit_sha
|
|
return result
|
|
|
|
|
|
def make_status_result(
|
|
branch="main",
|
|
commit_sha="abc123",
|
|
is_clean=True,
|
|
staged=None,
|
|
unstaged=None,
|
|
untracked=None,
|
|
ahead=0,
|
|
behind=0,
|
|
):
|
|
"""Create a mock status result."""
|
|
result = MagicMock()
|
|
result.branch = branch
|
|
result.commit_sha = commit_sha
|
|
result.is_clean = is_clean
|
|
result.staged = staged or []
|
|
result.unstaged = unstaged or []
|
|
result.untracked = untracked or []
|
|
result.ahead = ahead
|
|
result.behind = behind
|
|
return result
|
|
|
|
|
|
def make_branch_result(branch="feature", commit_sha="abc123", is_current=True):
|
|
"""Create a mock branch result."""
|
|
result = MagicMock()
|
|
result.branch = branch
|
|
result.commit_sha = commit_sha
|
|
result.is_current = is_current
|
|
return result
|
|
|
|
|
|
def make_list_branches_result(current="main", local=None, remote=None):
|
|
"""Create a mock list branches result."""
|
|
result = MagicMock()
|
|
result.current_branch = current
|
|
result.local_branches = local or ["main"]
|
|
result.remote_branches = remote or []
|
|
return result
|
|
|
|
|
|
def make_checkout_result(ref="main", commit_sha="abc123"):
|
|
"""Create a mock checkout result."""
|
|
result = MagicMock()
|
|
result.ref = ref
|
|
result.commit_sha = commit_sha
|
|
return result
|
|
|
|
|
|
def make_commit_result(
|
|
commit_sha="abc123",
|
|
short_sha="abc123",
|
|
message="Test",
|
|
files_changed=1,
|
|
insertions=10,
|
|
deletions=5,
|
|
):
|
|
"""Create a mock commit result."""
|
|
result = MagicMock()
|
|
result.commit_sha = commit_sha
|
|
result.short_sha = short_sha
|
|
result.message = message
|
|
result.files_changed = files_changed
|
|
result.insertions = insertions
|
|
result.deletions = deletions
|
|
return result
|
|
|
|
|
|
def make_push_result(branch="main", remote="origin", commits_pushed=1):
|
|
"""Create a mock push result."""
|
|
result = MagicMock()
|
|
result.branch = branch
|
|
result.remote = remote
|
|
result.commits_pushed = commits_pushed
|
|
return result
|
|
|
|
|
|
def make_pull_result(
|
|
branch="main", commits_received=1, fast_forward=True, conflicts=None
|
|
):
|
|
"""Create a mock pull result."""
|
|
result = MagicMock()
|
|
result.branch = branch
|
|
result.commits_received = commits_received
|
|
result.fast_forward = fast_forward
|
|
result.conflicts = conflicts or []
|
|
return result
|
|
|
|
|
|
def make_diff_result(
|
|
base="HEAD~1", head="HEAD", files=None, additions=10, deletions=5, files_changed=2
|
|
):
|
|
"""Create a mock diff result."""
|
|
result = MagicMock()
|
|
result.base = base
|
|
result.head = head
|
|
result.files = files or []
|
|
result.total_additions = additions
|
|
result.total_deletions = deletions
|
|
result.files_changed = files_changed
|
|
return result
|
|
|
|
|
|
def make_log_result(commits=None, total=2):
|
|
"""Create a mock log result."""
|
|
result = MagicMock()
|
|
result.commits = commits or [{"sha": "abc123"}, {"sha": "def456"}]
|
|
result.total_commits = total
|
|
return result
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_workspace_info():
|
|
"""Create a mock workspace info."""
|
|
return WorkspaceInfo(
|
|
project_id="test-project",
|
|
path="/tmp/test-workspace",
|
|
repo_url="https://gitea.test.com/owner/repo.git",
|
|
current_branch="main",
|
|
state=WorkspaceState.READY,
|
|
last_accessed=datetime.now(UTC),
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_github_workspace_info():
|
|
"""Create a mock workspace info for GitHub."""
|
|
return WorkspaceInfo(
|
|
project_id="github-project",
|
|
path="/tmp/github-workspace",
|
|
repo_url="https://github.com/owner/repo.git",
|
|
current_branch="main",
|
|
state=WorkspaceState.READY,
|
|
last_accessed=datetime.now(UTC),
|
|
)
|
|
|
|
|
|
class TestProviderDetection:
|
|
"""Tests for provider URL detection."""
|
|
|
|
def test_get_provider_for_github_url(self, test_settings):
|
|
"""Test GitHub URL detection."""
|
|
with (
|
|
patch("server._settings", test_settings),
|
|
patch("server._github_provider", MagicMock(name="github")),
|
|
patch("server._gitea_provider", MagicMock(name="gitea")),
|
|
):
|
|
provider = _get_provider_for_url("https://github.com/owner/repo.git")
|
|
assert provider is not None
|
|
|
|
def test_get_provider_for_gitea_url(self, test_settings):
|
|
"""Test Gitea URL detection."""
|
|
with (
|
|
patch("server._settings", test_settings),
|
|
patch("server._github_provider", MagicMock(name="github")),
|
|
patch("server._gitea_provider", MagicMock(name="gitea")),
|
|
):
|
|
provider = _get_provider_for_url("https://gitea.test.com/owner/repo.git")
|
|
assert provider is not None
|
|
|
|
def test_get_provider_no_settings(self):
|
|
"""Test provider detection without settings."""
|
|
with patch("server._settings", None):
|
|
provider = _get_provider_for_url("https://github.com/owner/repo.git")
|
|
assert provider is None
|
|
|
|
def test_get_auth_token_for_github(self, test_settings):
|
|
"""Test GitHub token selection."""
|
|
with patch("server._settings", test_settings):
|
|
token = _get_auth_token_for_url("https://github.com/owner/repo.git")
|
|
assert token == test_settings.github_token
|
|
|
|
def test_get_auth_token_for_gitea(self, test_settings):
|
|
"""Test Gitea token selection."""
|
|
with patch("server._settings", test_settings):
|
|
token = _get_auth_token_for_url("https://gitea.test.com/owner/repo.git")
|
|
assert token == test_settings.gitea_token
|
|
|
|
def test_get_auth_token_no_settings(self):
|
|
"""Test token selection without settings."""
|
|
with patch("server._settings", None):
|
|
token = _get_auth_token_for_url("https://github.com/owner/repo.git")
|
|
assert token is None
|
|
|
|
|
|
class TestCloneRepository:
|
|
"""Tests for clone_repository tool."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clone_success(self, test_settings, mock_workspace_info):
|
|
"""Test successful clone."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.create_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
mock_manager.update_workspace_branch = AsyncMock()
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.clone = AsyncMock(
|
|
return_value=make_clone_result(
|
|
workspace_path=mock_workspace_info.path,
|
|
branch="main",
|
|
commit_sha="abc123",
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await clone_repository.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
repo_url="https://gitea.test.com/owner/repo.git",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["branch"] == "main"
|
|
assert result["commit_sha"] == "abc123"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clone_with_branch_and_depth(
|
|
self, test_settings, mock_workspace_info
|
|
):
|
|
"""Test clone with branch and depth options."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.create_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
mock_manager.update_workspace_branch = AsyncMock()
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.clone = AsyncMock(
|
|
return_value=make_clone_result(
|
|
workspace_path=mock_workspace_info.path,
|
|
branch="develop",
|
|
commit_sha="def456",
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await clone_repository.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
repo_url="https://gitea.test.com/owner/repo.git",
|
|
branch="develop",
|
|
depth=1,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["branch"] == "develop"
|
|
|
|
|
|
class TestGitStatus:
|
|
"""Tests for git_status tool."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_success(self, test_settings, mock_workspace_info):
|
|
"""Test successful status."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.status = AsyncMock(
|
|
return_value=make_status_result(
|
|
branch="main",
|
|
commit_sha="abc123",
|
|
is_clean=True,
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await git_status.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["is_clean"] is True
|
|
assert result["branch"] == "main"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_with_changes(self, test_settings, mock_workspace_info):
|
|
"""Test status with uncommitted changes."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.status = AsyncMock(
|
|
return_value=make_status_result(
|
|
branch="feature",
|
|
commit_sha="abc123",
|
|
is_clean=False,
|
|
staged=["file1.py"],
|
|
unstaged=["file2.py"],
|
|
untracked=["file3.py"],
|
|
ahead=2,
|
|
behind=1,
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await git_status.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
include_untracked=True,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["is_clean"] is False
|
|
assert len(result["staged"]) == 1
|
|
assert len(result["unstaged"]) == 1
|
|
assert len(result["untracked"]) == 1
|
|
|
|
|
|
class TestBranchOperations:
|
|
"""Tests for branch operation tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_branch_success(self, test_settings, mock_workspace_info):
|
|
"""Test successful branch creation."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
mock_manager.update_workspace_branch = AsyncMock()
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.create_branch = AsyncMock(
|
|
return_value=make_branch_result(
|
|
branch="feature-new",
|
|
commit_sha="abc123",
|
|
is_current=True,
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await create_branch.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
branch_name="feature-new",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["branch"] == "feature-new"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_branch_from_ref(self, test_settings, mock_workspace_info):
|
|
"""Test branch creation from specific ref."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
mock_manager.update_workspace_branch = AsyncMock()
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.create_branch = AsyncMock(
|
|
return_value=make_branch_result(
|
|
branch="hotfix",
|
|
commit_sha="def456",
|
|
is_current=False,
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await create_branch.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
branch_name="hotfix",
|
|
from_ref="v1.0.0",
|
|
checkout=False,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["is_current"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_branches_success(self, test_settings, mock_workspace_info):
|
|
"""Test listing branches."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.list_branches = AsyncMock(
|
|
return_value=make_list_branches_result(
|
|
current="main",
|
|
local=["main", "develop", "feature-1"],
|
|
remote=["origin/main", "origin/develop"],
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await list_branches.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
include_remote=True,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["current_branch"] == "main"
|
|
assert len(result["local_branches"]) == 3
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_checkout_success(self, test_settings, mock_workspace_info):
|
|
"""Test checkout."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
mock_manager.update_workspace_branch = AsyncMock()
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.checkout = AsyncMock(
|
|
return_value=make_checkout_result(
|
|
ref="develop",
|
|
commit_sha="abc123",
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await checkout.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
ref="develop",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["ref"] == "develop"
|
|
|
|
|
|
class TestCommitOperations:
|
|
"""Tests for commit operation tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_commit_success(self, test_settings, mock_workspace_info):
|
|
"""Test successful commit."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.commit = AsyncMock(
|
|
return_value=make_commit_result(
|
|
commit_sha="abc123def456",
|
|
short_sha="abc123d",
|
|
message="Test commit",
|
|
files_changed=2,
|
|
insertions=10,
|
|
deletions=5,
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await commit.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
message="Test commit",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["files_changed"] == 2
|
|
assert result["insertions"] == 10
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_commit_with_author(self, test_settings, mock_workspace_info):
|
|
"""Test commit with custom author."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.commit = AsyncMock(
|
|
return_value=make_commit_result(
|
|
commit_sha="abc123",
|
|
short_sha="abc123",
|
|
message="Custom author commit",
|
|
files_changed=1,
|
|
insertions=5,
|
|
deletions=0,
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await commit.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
message="Custom author commit",
|
|
author_name="Custom Author",
|
|
author_email="custom@test.com",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
|
|
|
|
class TestPushPullOperations:
|
|
"""Tests for push/pull operation tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_push_success(self, test_settings, mock_workspace_info):
|
|
"""Test successful push."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.push = AsyncMock(
|
|
return_value=make_push_result(
|
|
branch="main",
|
|
remote="origin",
|
|
commits_pushed=2,
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await push.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["commits_pushed"] == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pull_success(self, test_settings, mock_workspace_info):
|
|
"""Test successful pull."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.pull = AsyncMock(
|
|
return_value=make_pull_result(
|
|
branch="main",
|
|
commits_received=3,
|
|
fast_forward=True,
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await pull.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["commits_received"] == 3
|
|
assert result["fast_forward"] is True
|
|
|
|
|
|
class TestDiffLogOperations:
|
|
"""Tests for diff/log operation tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_diff_success(self, test_settings, mock_workspace_info):
|
|
"""Test successful diff."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.diff = AsyncMock(
|
|
return_value=make_diff_result(
|
|
base="HEAD~1",
|
|
head="HEAD",
|
|
additions=10,
|
|
deletions=5,
|
|
files_changed=2,
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await diff.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["total_additions"] == 10
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_log_success(self, test_settings, mock_workspace_info):
|
|
"""Test successful log."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.log = AsyncMock(
|
|
return_value=make_log_result(
|
|
commits=[
|
|
{"sha": "abc123", "message": "Commit 1"},
|
|
{"sha": "def456", "message": "Commit 2"},
|
|
],
|
|
total=2,
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await log.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
limit=10,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["total_commits"] == 2
|
|
|
|
|
|
class TestPROperations:
|
|
"""Tests for PR operation tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_pr_success(self, test_settings, mock_workspace_info):
|
|
"""Test successful PR creation."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_provider = AsyncMock()
|
|
mock_provider.parse_repo_url = MagicMock(return_value=("owner", "repo"))
|
|
mock_provider.create_pr = AsyncMock(
|
|
return_value=CreatePRResult(
|
|
success=True,
|
|
pr_number=42,
|
|
pr_url="https://gitea.test.com/owner/repo/pull/42",
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server._get_provider_for_url", return_value=mock_provider),
|
|
):
|
|
result = await create_pull_request.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
title="Test PR",
|
|
source_branch="feature",
|
|
body="Test body",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["pr_number"] == 42
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_pr_success(self, test_settings, mock_workspace_info):
|
|
"""Test getting PR."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_provider = AsyncMock()
|
|
mock_provider.parse_repo_url = MagicMock(return_value=("owner", "repo"))
|
|
mock_provider.get_pr = AsyncMock(
|
|
return_value=GetPRResult(
|
|
success=True,
|
|
pr={"number": 42, "title": "Test PR", "state": "open"},
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server._get_provider_for_url", return_value=mock_provider),
|
|
):
|
|
result = await get_pull_request.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
pr_number=42,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["pr"]["number"] == 42
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_prs_success(self, test_settings, mock_workspace_info):
|
|
"""Test listing PRs."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_provider = AsyncMock()
|
|
mock_provider.parse_repo_url = MagicMock(return_value=("owner", "repo"))
|
|
mock_provider.list_prs = AsyncMock(
|
|
return_value=ListPRsResult(
|
|
success=True,
|
|
pull_requests=[{"number": 1}, {"number": 2}],
|
|
total_count=2,
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
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=None,
|
|
author=None,
|
|
limit=20,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["total_count"] == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_prs_with_state(self, test_settings, mock_workspace_info):
|
|
"""Test listing PRs with state filter."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_provider = AsyncMock()
|
|
mock_provider.parse_repo_url = MagicMock(return_value=("owner", "repo"))
|
|
mock_provider.list_prs = AsyncMock(
|
|
return_value=ListPRsResult(
|
|
success=True,
|
|
pull_requests=[{"number": 1, "state": "open"}],
|
|
total_count=1,
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
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="open",
|
|
author=None,
|
|
limit=20,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_merge_pr_success(self, test_settings, mock_workspace_info):
|
|
"""Test merging PR."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_provider = AsyncMock()
|
|
mock_provider.parse_repo_url = MagicMock(return_value=("owner", "repo"))
|
|
mock_provider.merge_pr = AsyncMock(
|
|
return_value=MergePRResult(
|
|
success=True,
|
|
merge_commit_sha="abc123",
|
|
branch_deleted=True,
|
|
)
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
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="squash",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["merge_commit_sha"] == "abc123"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pr_no_provider(self, test_settings, mock_workspace_info):
|
|
"""Test PR operation without provider."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server._get_provider_for_url", return_value=None),
|
|
):
|
|
result = await create_pull_request.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
title="Test PR",
|
|
source_branch="feature",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert "provider" in result["error"].lower()
|
|
|
|
|
|
class TestWorkspaceOperations:
|
|
"""Tests for workspace operation tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_workspace_success(self, test_settings, mock_workspace_info):
|
|
"""Test getting workspace."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
):
|
|
result = await get_workspace.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert "workspace" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_lock_workspace_success(self, test_settings, mock_workspace_info):
|
|
"""Test locking workspace."""
|
|
mock_workspace_info.lock_holder = "agent-1"
|
|
mock_workspace_info.lock_expires = datetime.now(UTC)
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.lock_workspace = AsyncMock(return_value=True)
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
):
|
|
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_settings):
|
|
"""Test unlocking workspace."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.unlock_workspace = AsyncMock(return_value=True)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
):
|
|
result = await unlock_workspace.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is True
|
|
|
|
|
|
class TestErrorHandling:
|
|
"""Tests for error handling in tools."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_git_ops_error_handling(self, test_settings, mock_workspace_info):
|
|
"""Test GitOpsError is handled properly."""
|
|
from exceptions import CloneError
|
|
|
|
mock_manager = AsyncMock()
|
|
mock_manager.create_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
mock_git = MagicMock()
|
|
mock_git.clone = AsyncMock(
|
|
side_effect=CloneError("https://test.com/repo.git", "Clone failed")
|
|
)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
patch("server.GitWrapper", return_value=mock_git),
|
|
):
|
|
result = await clone_repository.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
repo_url="https://gitea.test.com/owner/repo.git",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert "Clone failed" in result["error"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unexpected_error_handling(self, test_settings, mock_workspace_info):
|
|
"""Test unexpected errors are handled."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(side_effect=RuntimeError("Unexpected"))
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
):
|
|
result = await git_status.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["code"] == ErrorCode.INTERNAL_ERROR.value
|
|
|
|
|
|
class TestValidationFunctions:
|
|
"""Tests for input validation functions."""
|
|
|
|
def test_validate_id_non_string(self):
|
|
"""Test validation when value is not a string."""
|
|
result = _validate_id(123, "project_id")
|
|
assert result == "project_id must be a string"
|
|
|
|
def test_validate_id_empty(self):
|
|
"""Test validation when value is empty."""
|
|
result = _validate_id("", "project_id")
|
|
assert result == "project_id is required"
|
|
|
|
def test_validate_id_invalid_chars(self):
|
|
"""Test validation when value has invalid characters."""
|
|
result = _validate_id("project@!#", "project_id")
|
|
assert "Invalid project_id" in result
|
|
|
|
def test_validate_id_valid(self):
|
|
"""Test validation with valid ID."""
|
|
result = _validate_id("valid-project_123", "project_id")
|
|
assert result is None
|
|
|
|
def test_validate_branch_non_string(self):
|
|
"""Test validation when branch is not a string."""
|
|
result = _validate_branch(123)
|
|
assert result == "Branch name must be a string"
|
|
|
|
def test_validate_branch_empty(self):
|
|
"""Test validation when branch is empty."""
|
|
result = _validate_branch("")
|
|
assert result == "Branch name is required"
|
|
|
|
def test_validate_branch_invalid(self):
|
|
"""Test validation with invalid branch name."""
|
|
result = _validate_branch("branch with spaces")
|
|
assert "Invalid branch name" in result
|
|
|
|
def test_validate_branch_valid(self):
|
|
"""Test validation with valid branch name."""
|
|
result = _validate_branch("feature/my-branch.v1")
|
|
assert result is None
|
|
|
|
def test_validate_url_non_string(self):
|
|
"""Test validation when URL is not a string."""
|
|
result = _validate_url(123)
|
|
assert result == "Repository URL must be a string"
|
|
|
|
def test_validate_url_empty(self):
|
|
"""Test validation when URL is empty."""
|
|
result = _validate_url("")
|
|
assert result == "Repository URL is required"
|
|
|
|
def test_validate_url_invalid(self):
|
|
"""Test validation with invalid URL."""
|
|
result = _validate_url("not-a-url")
|
|
assert "Invalid repository URL" in result
|
|
|
|
def test_validate_url_valid_https(self):
|
|
"""Test validation with valid HTTPS URL."""
|
|
result = _validate_url("https://github.com/owner/repo.git")
|
|
assert result is None
|
|
|
|
def test_validate_url_valid_ssh(self):
|
|
"""Test validation with valid SSH URL."""
|
|
result = _validate_url("git@github.com:owner/repo.git")
|
|
assert result is None
|
|
|
|
|
|
class TestProviderDetectionAdvanced:
|
|
"""Additional tests for provider detection."""
|
|
|
|
def test_get_provider_github_enterprise(self, test_settings):
|
|
"""Test provider detection for GitHub Enterprise URL."""
|
|
test_settings.github_api_url = "https://github.mycompany.com/api/v3"
|
|
|
|
with (
|
|
patch("server._settings", test_settings),
|
|
patch("server._github_provider", MagicMock()) as mock_github,
|
|
):
|
|
result = _get_provider_for_url(
|
|
"https://github.mycompany.com/owner/repo.git"
|
|
)
|
|
assert result == mock_github
|
|
|
|
def test_get_provider_default_to_gitea(self, test_settings):
|
|
"""Test default to Gitea for unknown URLs."""
|
|
test_settings.gitea_base_url = "https://gitea.example.com"
|
|
test_settings.github_api_url = None
|
|
|
|
with (
|
|
patch("server._settings", test_settings),
|
|
patch("server._gitea_provider", MagicMock()) as mock_gitea,
|
|
):
|
|
# A random URL that doesn't match github patterns
|
|
result = _get_provider_for_url("https://git.example.com/owner/repo.git")
|
|
assert result == mock_gitea
|
|
|
|
|
|
class TestCloneValidation:
|
|
"""Tests for clone tool validation."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clone_invalid_project_id(self, test_settings):
|
|
"""Test clone with invalid project_id."""
|
|
with patch("server._settings", test_settings):
|
|
result = await clone_repository.fn(
|
|
project_id="invalid!project",
|
|
agent_id="agent-1",
|
|
repo_url="https://github.com/owner/repo.git",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["code"] == ErrorCode.INVALID_REQUEST.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clone_invalid_agent_id(self, test_settings):
|
|
"""Test clone with invalid agent_id."""
|
|
with patch("server._settings", test_settings):
|
|
result = await clone_repository.fn(
|
|
project_id="valid-project",
|
|
agent_id="invalid!agent",
|
|
repo_url="https://github.com/owner/repo.git",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["code"] == ErrorCode.INVALID_REQUEST.value
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clone_invalid_url(self, test_settings):
|
|
"""Test clone with invalid URL."""
|
|
with patch("server._settings", test_settings):
|
|
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 result["code"] == ErrorCode.INVALID_REQUEST.value
|
|
|
|
|
|
class TestBranchValidation:
|
|
"""Tests for branch tool validation."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_branch_invalid_name(self, test_settings, mock_workspace_info):
|
|
"""Test create_branch with invalid branch name."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=mock_workspace_info)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
):
|
|
result = await create_branch.fn(
|
|
project_id="test-project",
|
|
agent_id="agent-1",
|
|
branch_name="invalid branch name",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["code"] == ErrorCode.INVALID_REQUEST.value
|
|
|
|
|
|
class TestStatusWithWorkspace:
|
|
"""Additional tests for status operations."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_workspace_not_found(self, test_settings):
|
|
"""Test status when workspace doesn't exist."""
|
|
mock_manager = AsyncMock()
|
|
mock_manager.get_workspace = AsyncMock(return_value=None)
|
|
|
|
with (
|
|
patch("server._workspace_manager", mock_manager),
|
|
patch("server._settings", test_settings),
|
|
):
|
|
result = await git_status.fn(
|
|
project_id="nonexistent-project",
|
|
agent_id="agent-1",
|
|
)
|
|
|
|
assert result["success"] is False
|
|
assert result["code"] == ErrorCode.WORKSPACE_NOT_FOUND.value
|