From 0a624a94af3abcd35156313fc04ca85bdb5334d9 Mon Sep 17 00:00:00 2001 From: Felipe Cardoso Date: Wed, 7 Jan 2026 09:17:32 +0100 Subject: [PATCH] **test(git-ops): add comprehensive tests for server and API tools** - 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. --- .../git-ops/tests/test_api_endpoints.py | 440 +++++++ .../git-ops/tests/test_server_tools.py | 1170 +++++++++++++++++ 2 files changed, 1610 insertions(+) create mode 100644 mcp-servers/git-ops/tests/test_api_endpoints.py create mode 100644 mcp-servers/git-ops/tests/test_server_tools.py diff --git a/mcp-servers/git-ops/tests/test_api_endpoints.py b/mcp-servers/git-ops/tests/test_api_endpoints.py new file mode 100644 index 0000000..b9a1f6f --- /dev/null +++ b/mcp-servers/git-ops/tests/test_api_endpoints.py @@ -0,0 +1,440 @@ +""" +Tests for FastAPI endpoints. + +Tests health check and MCP JSON-RPC endpoints. +""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + + +class TestHealthEndpoint: + """Tests for health check endpoint.""" + + @pytest.mark.asyncio + async def test_health_no_providers(self): + """Test health check when no providers configured.""" + from httpx import ASGITransport, AsyncClient + + from server import app + + with ( + patch("server._settings", MagicMock()), + patch("server._workspace_manager", None), + patch("server._gitea_provider", None), + patch("server._github_provider", None), + ): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.get("/health") + + assert response.status_code == 200 + data = response.json() + assert data["status"] in ["healthy", "degraded"] + assert data["service"] == "git-ops" + + @pytest.mark.asyncio + async def test_health_with_gitea_connected(self): + """Test health check with Gitea provider connected.""" + from httpx import ASGITransport, AsyncClient + + from server import app + + mock_gitea = AsyncMock() + mock_gitea.is_connected = AsyncMock(return_value=True) + mock_gitea.get_authenticated_user = AsyncMock(return_value="test-user") + + with ( + patch("server._settings", MagicMock()), + patch("server._workspace_manager", None), + patch("server._gitea_provider", mock_gitea), + patch("server._github_provider", None), + ): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.get("/health") + + assert response.status_code == 200 + data = response.json() + assert "gitea" in data["dependencies"] + + @pytest.mark.asyncio + async def test_health_with_gitea_not_connected(self): + """Test health check when Gitea is not connected.""" + from httpx import ASGITransport, AsyncClient + + from server import app + + mock_gitea = AsyncMock() + mock_gitea.is_connected = AsyncMock(return_value=False) + + with ( + patch("server._settings", MagicMock()), + patch("server._workspace_manager", None), + patch("server._gitea_provider", mock_gitea), + patch("server._github_provider", None), + ): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.get("/health") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "degraded" + + @pytest.mark.asyncio + async def test_health_with_gitea_error(self): + """Test health check when Gitea throws error.""" + from httpx import ASGITransport, AsyncClient + + from server import app + + mock_gitea = AsyncMock() + mock_gitea.is_connected = AsyncMock(side_effect=Exception("Connection failed")) + + with ( + patch("server._settings", MagicMock()), + patch("server._workspace_manager", None), + patch("server._gitea_provider", mock_gitea), + patch("server._github_provider", None), + ): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.get("/health") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "degraded" + + @pytest.mark.asyncio + async def test_health_with_github_connected(self): + """Test health check with GitHub provider connected.""" + from httpx import ASGITransport, AsyncClient + + from server import app + + mock_github = AsyncMock() + mock_github.is_connected = AsyncMock(return_value=True) + mock_github.get_authenticated_user = AsyncMock(return_value="github-user") + + with ( + patch("server._settings", MagicMock()), + patch("server._workspace_manager", None), + patch("server._gitea_provider", None), + patch("server._github_provider", mock_github), + ): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.get("/health") + + assert response.status_code == 200 + data = response.json() + assert "github" in data["dependencies"] + + @pytest.mark.asyncio + async def test_health_with_github_not_connected(self): + """Test health check when GitHub is not connected.""" + from httpx import ASGITransport, AsyncClient + + from server import app + + mock_github = AsyncMock() + mock_github.is_connected = AsyncMock(return_value=False) + + with ( + patch("server._settings", MagicMock()), + patch("server._workspace_manager", None), + patch("server._gitea_provider", None), + patch("server._github_provider", mock_github), + ): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.get("/health") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "degraded" + + @pytest.mark.asyncio + async def test_health_with_github_error(self): + """Test health check when GitHub throws error.""" + from httpx import ASGITransport, AsyncClient + + from server import app + + mock_github = AsyncMock() + mock_github.is_connected = AsyncMock(side_effect=Exception("Auth failed")) + + with ( + patch("server._settings", MagicMock()), + patch("server._workspace_manager", None), + patch("server._gitea_provider", None), + patch("server._github_provider", mock_github), + ): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.get("/health") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "degraded" + + @pytest.mark.asyncio + async def test_health_with_workspace_manager(self): + """Test health check with workspace manager.""" + from pathlib import Path + + from httpx import ASGITransport, AsyncClient + + from server import app + + mock_manager = AsyncMock() + mock_manager.base_path = Path("/tmp/workspaces") + mock_manager.list_workspaces = AsyncMock(return_value=[]) + + with ( + patch("server._settings", MagicMock()), + patch("server._workspace_manager", mock_manager), + patch("server._gitea_provider", None), + patch("server._github_provider", None), + ): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.get("/health") + + assert response.status_code == 200 + data = response.json() + assert "workspace" in data["dependencies"] + + @pytest.mark.asyncio + async def test_health_workspace_error(self): + """Test health check when workspace manager throws error.""" + from pathlib import Path + + from httpx import ASGITransport, AsyncClient + + from server import app + + mock_manager = AsyncMock() + mock_manager.base_path = Path("/tmp/workspaces") + mock_manager.list_workspaces = AsyncMock(side_effect=Exception("Disk full")) + + with ( + patch("server._settings", MagicMock()), + patch("server._workspace_manager", mock_manager), + patch("server._gitea_provider", None), + patch("server._github_provider", None), + ): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.get("/health") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "degraded" + + +class TestMCPToolsEndpoint: + """Tests for MCP tools list endpoint.""" + + @pytest.mark.asyncio + async def test_list_mcp_tools(self): + """Test listing MCP tools.""" + from httpx import ASGITransport, AsyncClient + + from server import app + + with patch("server._settings", MagicMock()): + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.get("/mcp/tools") + + assert response.status_code == 200 + data = response.json() + assert "tools" in data + + +class TestMCPRPCEndpoint: + """Tests for MCP JSON-RPC endpoint.""" + + @pytest.mark.asyncio + async def test_mcp_rpc_invalid_json(self): + """Test RPC with invalid JSON.""" + from httpx import ASGITransport, AsyncClient + + from server import app + + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.post( + "/mcp", + content="not valid json", + headers={"Content-Type": "application/json"}, + ) + + assert response.status_code == 400 + data = response.json() + assert data["error"]["code"] == -32700 + + @pytest.mark.asyncio + async def test_mcp_rpc_invalid_jsonrpc(self): + """Test RPC with invalid jsonrpc version.""" + from httpx import ASGITransport, AsyncClient + + from server import app + + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.post( + "/mcp", json={"jsonrpc": "1.0", "method": "test", "id": 1} + ) + + assert response.status_code == 400 + data = response.json() + assert data["error"]["code"] == -32600 + + @pytest.mark.asyncio + async def test_mcp_rpc_missing_method(self): + """Test RPC with missing method.""" + from httpx import ASGITransport, AsyncClient + + from server import app + + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.post("/mcp", json={"jsonrpc": "2.0", "id": 1}) + + assert response.status_code == 400 + data = response.json() + assert data["error"]["code"] == -32600 + + @pytest.mark.asyncio + async def test_mcp_rpc_method_not_found(self): + """Test RPC with unknown method.""" + from httpx import ASGITransport, AsyncClient + + from server import app + + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.post( + "/mcp", + json={ + "jsonrpc": "2.0", + "method": "unknown_method", + "params": {}, + "id": 1, + }, + ) + + assert response.status_code == 404 + data = response.json() + assert data["error"]["code"] == -32601 + + +class TestTypeSchemaConversion: + """Tests for type to JSON schema conversion.""" + + def test_python_type_to_json_schema_str(self): + """Test converting str type to JSON schema.""" + 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 converting int type to JSON schema.""" + 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_float(self): + """Test converting float type to JSON schema.""" + from server import _python_type_to_json_schema + + result = _python_type_to_json_schema(float) + assert result == {"type": "number"} + + def test_python_type_to_json_schema_bool(self): + """Test converting bool type to JSON schema.""" + 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_none(self): + """Test converting NoneType to JSON schema.""" + from server import _python_type_to_json_schema + + result = _python_type_to_json_schema(type(None)) + assert result == {"type": "null"} + + def test_python_type_to_json_schema_list(self): + """Test converting list type to JSON schema.""" + from server import _python_type_to_json_schema + + result = _python_type_to_json_schema(list[str]) + assert result["type"] == "array" + + def test_python_type_to_json_schema_dict(self): + """Test converting dict type to JSON schema.""" + from server import _python_type_to_json_schema + + result = _python_type_to_json_schema(dict[str, int]) + assert result == {"type": "object"} + + def test_python_type_to_json_schema_optional(self): + """Test converting Optional type to JSON schema.""" + from server import _python_type_to_json_schema + + result = _python_type_to_json_schema(str | None) + # The function returns object type for complex union types + assert "type" in result + + +class TestToolSchema: + """Tests for tool schema extraction.""" + + def test_get_tool_schema_simple(self): + """Test getting schema from simple function.""" + from server import _get_tool_schema + + def simple_func(name: str, count: int) -> str: + return f"{name}: {count}" + + result = _get_tool_schema(simple_func) + assert "properties" in result + assert "name" in result["properties"] + assert "count" in result["properties"] + + def test_register_and_get_tool(self): + """Test registering a tool.""" + from server import _register_tool, _tool_registry + + async def test_tool(x: str) -> str: + """A test tool.""" + return x + + _register_tool("test_tool", test_tool, "Test description") + + assert "test_tool" in _tool_registry + assert _tool_registry["test_tool"]["description"] == "Test description" + + # Clean up + del _tool_registry["test_tool"] diff --git a/mcp-servers/git-ops/tests/test_server_tools.py b/mcp-servers/git-ops/tests/test_server_tools.py new file mode 100644 index 0000000..7746e2c --- /dev/null +++ b/mcp-servers/git-ops/tests/test_server_tools.py @@ -0,0 +1,1170 @@ +""" +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