471 lines
18 KiB
Python
471 lines
18 KiB
Python
"""Tests for the MCP server tool functions.
|
|
|
|
Each tool is tested by calling the underlying function directly (not through
|
|
the MCP protocol) with mocked API responses.
|
|
"""
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
from mcp_server import ( # noqa: E402
|
|
gitea_create_issue,
|
|
gitea_create_pr,
|
|
gitea_close_issue,
|
|
gitea_list_issues,
|
|
gitea_view_issue,
|
|
gitea_mark_issue,
|
|
gitea_mirror_refs,
|
|
gitea_list_prs,
|
|
gitea_view_pr,
|
|
gitea_merge_pr,
|
|
gitea_review_pr,
|
|
gitea_delete_branch,
|
|
gitea_edit_pr,
|
|
gitea_get_file,
|
|
gitea_commit_files,
|
|
)
|
|
|
|
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Create Issue
|
|
# ---------------------------------------------------------------------------
|
|
class TestCreateIssue(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_creates_issue(self, _auth, mock_api):
|
|
mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"}
|
|
result = gitea_create_issue(title="Test issue", body="body text")
|
|
self.assertEqual(result["number"], 1)
|
|
self.assertIn("issues/1", result["url"])
|
|
mock_api.assert_called_once()
|
|
# Verify payload
|
|
call_args = mock_api.call_args
|
|
self.assertEqual(call_args[0][0], "POST")
|
|
self.assertEqual(call_args[0][3]["title"], "Test issue")
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_creates_on_prgs(self, _auth, mock_api):
|
|
mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"}
|
|
result = gitea_create_issue(title="Test", remote="prgs")
|
|
self.assertEqual(result["number"], 5)
|
|
url = mock_api.call_args[0][1]
|
|
self.assertIn("gitea.prgs.cc", url)
|
|
self.assertIn("Scaled-Tech-Consulting", url)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Create PR
|
|
# ---------------------------------------------------------------------------
|
|
class TestCreatePR(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_creates_pr(self, _auth, mock_api):
|
|
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
|
result = gitea_create_pr(title="feat: X", head="feat/x", base="main")
|
|
self.assertEqual(result["number"], 3)
|
|
payload = mock_api.call_args[0][3]
|
|
self.assertEqual(payload["head"], "feat/x")
|
|
self.assertEqual(payload["base"], "main")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Close Issue
|
|
# ---------------------------------------------------------------------------
|
|
class TestCloseIssue(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_closes_issue(self, _auth, mock_api):
|
|
mock_api.return_value = {"state": "closed"}
|
|
result = gitea_close_issue(issue_number=42)
|
|
self.assertTrue(result["success"])
|
|
self.assertIn("42", result["message"])
|
|
payload = mock_api.call_args[0][3]
|
|
self.assertEqual(payload["state"], "closed")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# List Issues
|
|
# ---------------------------------------------------------------------------
|
|
class TestListIssues(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_returns_formatted_list(self, _auth, mock_api):
|
|
mock_api.return_value = [
|
|
{
|
|
"number": 1, "title": "Bug", "state": "open",
|
|
"labels": [{"name": "bug"}],
|
|
"assignee": {"login": "alice"},
|
|
},
|
|
{
|
|
"number": 2, "title": "Feature", "state": "open",
|
|
"labels": [], "assignee": None,
|
|
},
|
|
]
|
|
result = gitea_list_issues()
|
|
self.assertEqual(len(result), 2)
|
|
self.assertEqual(result[0]["number"], 1)
|
|
self.assertEqual(result[0]["labels"], ["bug"])
|
|
self.assertEqual(result[0]["assignee"], "alice")
|
|
self.assertEqual(result[1]["assignee"], "")
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_passes_label_filter(self, _auth, mock_api):
|
|
mock_api.return_value = []
|
|
gitea_list_issues(label="important")
|
|
url = mock_api.call_args[0][1]
|
|
self.assertIn("labels=important", url)
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_passes_state_filter(self, _auth, mock_api):
|
|
mock_api.return_value = []
|
|
gitea_list_issues(state="closed")
|
|
url = mock_api.call_args[0][1]
|
|
self.assertIn("state=closed", url)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# View Issue
|
|
# ---------------------------------------------------------------------------
|
|
class TestViewIssue(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_returns_full_details(self, _auth, mock_api):
|
|
mock_api.return_value = {
|
|
"number": 7, "title": "MCP server", "body": "Build it",
|
|
"state": "open", "labels": [{"name": "important"}],
|
|
"assignee": {"login": "jason"},
|
|
"html_url": "https://gitea.prgs.cc/issues/7",
|
|
}
|
|
result = gitea_view_issue(issue_number=7, remote="prgs")
|
|
self.assertEqual(result["number"], 7)
|
|
self.assertEqual(result["body"], "Build it")
|
|
self.assertEqual(result["labels"], ["important"])
|
|
self.assertEqual(result["assignee"], "jason")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mark Issue
|
|
# ---------------------------------------------------------------------------
|
|
class TestMarkIssue(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_start_adds_label(self, _auth, mock_api):
|
|
# First call: get labels; second call: add label
|
|
mock_api.side_effect = [
|
|
[{"id": 10, "name": "status:in-progress"}],
|
|
[{"name": "status:in-progress"}],
|
|
]
|
|
result = gitea_mark_issue(issue_number=5, action="start")
|
|
self.assertTrue(result["success"])
|
|
self.assertIn("claimed", result["message"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_done_removes_label(self, _auth, mock_api):
|
|
mock_api.side_effect = [
|
|
[{"id": 10, "name": "status:in-progress"}],
|
|
None,
|
|
]
|
|
result = gitea_mark_issue(issue_number=5, action="done")
|
|
self.assertTrue(result["success"])
|
|
self.assertIn("released", result["message"])
|
|
|
|
def test_invalid_action_raises(self):
|
|
with self.assertRaises(ValueError):
|
|
gitea_mark_issue(issue_number=5, action="pause")
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_missing_label_raises(self, _auth, mock_api):
|
|
mock_api.return_value = [] # no labels exist
|
|
with self.assertRaises(RuntimeError):
|
|
gitea_mark_issue(issue_number=5, action="start")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Auth errors
|
|
# ---------------------------------------------------------------------------
|
|
class TestAuthErrors(unittest.TestCase):
|
|
|
|
@patch("mcp_server.get_auth_header", return_value=None)
|
|
def test_no_credentials_raises(self, _auth):
|
|
with self.assertRaises(RuntimeError):
|
|
gitea_create_issue(title="test")
|
|
|
|
def test_unknown_remote_raises(self):
|
|
with self.assertRaises(ValueError):
|
|
gitea_create_issue(title="test", remote="nonexistent")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mirror Refs
|
|
# ---------------------------------------------------------------------------
|
|
class TestMirrorRefs(unittest.TestCase):
|
|
|
|
@patch("mcp_server.subprocess.run")
|
|
def test_dry_run_default(self, mock_run):
|
|
mock_run.return_value = MagicMock(
|
|
stdout="DRY RUN\n", stderr="", returncode=0
|
|
)
|
|
result = gitea_mirror_refs()
|
|
self.assertEqual(result["return_code"], 0)
|
|
# Should NOT have --apply
|
|
args = mock_run.call_args[0][0]
|
|
self.assertNotIn("--apply", args)
|
|
self.assertNotIn("--force", args)
|
|
|
|
@patch("mcp_server.subprocess.run")
|
|
def test_apply_flag(self, mock_run):
|
|
mock_run.return_value = MagicMock(
|
|
stdout="done\n", stderr="", returncode=0
|
|
)
|
|
result = gitea_mirror_refs(apply=True)
|
|
args = mock_run.call_args[0][0]
|
|
self.assertIn("--apply", args)
|
|
|
|
@patch("mcp_server.subprocess.run")
|
|
def test_force_flag(self, mock_run):
|
|
mock_run.return_value = MagicMock(
|
|
stdout="", stderr="", returncode=0
|
|
)
|
|
gitea_mirror_refs(apply=True, force=True)
|
|
args = mock_run.call_args[0][0]
|
|
self.assertIn("--apply", args)
|
|
self.assertIn("--force", args)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# List PRs
|
|
# ---------------------------------------------------------------------------
|
|
class TestListPRs(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_list_prs(self, _auth, mock_api):
|
|
mock_api.return_value = [
|
|
{
|
|
"number": 1, "title": "PR 1", "state": "open",
|
|
"head": {"ref": "branch1"}, "base": {"ref": "main"},
|
|
"html_url": "http://url1", "mergeable": True
|
|
}
|
|
]
|
|
result = gitea_list_prs()
|
|
self.assertEqual(len(result), 1)
|
|
self.assertEqual(result[0]["number"], 1)
|
|
self.assertEqual(result[0]["head"], "branch1")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# View PR
|
|
# ---------------------------------------------------------------------------
|
|
class TestViewPR(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_view_pr(self, _auth, mock_api):
|
|
mock_api.return_value = {
|
|
"number": 1, "title": "PR 1", "state": "open",
|
|
"head": {"ref": "branch1"}, "base": {"ref": "main"},
|
|
"html_url": "http://url1", "mergeable": True, "body": "description",
|
|
"user": {"login": "user1"}
|
|
}
|
|
result = gitea_view_pr(pr_number=1)
|
|
self.assertEqual(result["number"], 1)
|
|
self.assertEqual(result["body"], "description")
|
|
self.assertEqual(result["user"], "user1")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Merge PR
|
|
# ---------------------------------------------------------------------------
|
|
class TestMergePR(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_merge_pr(self, _auth, mock_api):
|
|
mock_api.return_value = {}
|
|
result = gitea_merge_pr(pr_number=1, do="squash", title="T", message="M", force=True)
|
|
self.assertTrue(result["success"])
|
|
self.assertIn("merged", result["message"])
|
|
# Check payload
|
|
payload = mock_api.call_args[0][3]
|
|
self.assertEqual(payload["Do"], "squash")
|
|
self.assertEqual(payload["MergeTitleField"], "T")
|
|
self.assertEqual(payload["MergeMessageField"], "M")
|
|
self.assertEqual(payload["force_merge"], True)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Review PR
|
|
# ---------------------------------------------------------------------------
|
|
class TestReviewPR(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_review_pr_and_merge(self, _auth, mock_api):
|
|
# GET PR response (fetch head SHA)
|
|
mock_api.side_effect = [
|
|
{"head": {"sha": "sha-val-123"}}, # GET PR pulls/1
|
|
{}, # POST review
|
|
{}, # POST merge
|
|
]
|
|
result = gitea_review_pr(
|
|
pr_number=1,
|
|
event="APPROVE",
|
|
body="Looks good",
|
|
merge=True,
|
|
merge_method="squash"
|
|
)
|
|
self.assertTrue(result["success"])
|
|
self.assertIn("Successfully submitted review", result["message"])
|
|
self.assertIn("Successfully merged", result["message"])
|
|
|
|
# Check call counts and arguments
|
|
self.assertEqual(mock_api.call_count, 3)
|
|
|
|
# Verify GET PR
|
|
self.assertEqual(mock_api.call_args_list[0][0][0], "GET")
|
|
|
|
# Verify POST review
|
|
self.assertEqual(mock_api.call_args_list[1][0][0], "POST")
|
|
self.assertEqual(mock_api.call_args_list[1][0][3]["event"], "APPROVE")
|
|
self.assertEqual(mock_api.call_args_list[1][0][3]["commit_id"], "sha-val-123")
|
|
|
|
# Verify POST merge
|
|
self.assertEqual(mock_api.call_args_list[2][0][0], "POST")
|
|
self.assertEqual(mock_api.call_args_list[2][0][3]["Do"], "squash")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Delete Branch
|
|
# ---------------------------------------------------------------------------
|
|
class TestDeleteBranch(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_delete_branch(self, _auth, mock_api):
|
|
mock_api.return_value = {}
|
|
result = gitea_delete_branch(branch="feat/branch")
|
|
self.assertTrue(result["success"])
|
|
self.assertIn("deleted", result["message"])
|
|
# Check url encoding of branch name
|
|
url = mock_api.call_args[0][1]
|
|
self.assertIn("feat%2Fbranch", url)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Edit PR
|
|
# ---------------------------------------------------------------------------
|
|
class TestEditPR(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_edit_pr_success(self, _auth, mock_api):
|
|
mock_api.return_value = {
|
|
"number": 1,
|
|
"title": "New Title",
|
|
"body": "New description",
|
|
"state": "open",
|
|
"html_url": "https://gitea.example.com/pulls/1"
|
|
}
|
|
result = gitea_edit_pr(
|
|
pr_number=1,
|
|
title="New Title",
|
|
body="New description",
|
|
state="open"
|
|
)
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(result["title"], "New Title")
|
|
self.assertEqual(result["body"], "New description")
|
|
|
|
# Verify PATCH and payload
|
|
call_args = mock_api.call_args
|
|
self.assertEqual(call_args[0][0], "PATCH")
|
|
self.assertEqual(call_args[0][3]["title"], "New Title")
|
|
self.assertEqual(call_args[0][3]["body"], "New description")
|
|
self.assertEqual(call_args[0][3]["state"], "open")
|
|
|
|
def test_edit_pr_no_fields_raises(self):
|
|
with self.assertRaises(ValueError):
|
|
gitea_edit_pr(pr_number=1)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Get File
|
|
# ---------------------------------------------------------------------------
|
|
class TestGetFile(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_get_file_success(self, _auth, mock_api):
|
|
mock_api.return_value = {
|
|
"name": "README.md",
|
|
"path": "README.md",
|
|
"sha": "3a0b123",
|
|
"size": 100,
|
|
"encoding": "base64",
|
|
"content": "SGVsbG8gV29ybGQ="
|
|
}
|
|
result = gitea_get_file(filepath="README.md", ref="main")
|
|
self.assertEqual(result["name"], "README.md")
|
|
self.assertEqual(result["sha"], "3a0b123")
|
|
self.assertEqual(result["content"], "SGVsbG8gV29ybGQ=")
|
|
|
|
# Verify endpoint and GET method
|
|
call_args = mock_api.call_args
|
|
self.assertEqual(call_args[0][0], "GET")
|
|
self.assertIn("contents/README.md", call_args[0][1])
|
|
self.assertIn("ref=main", call_args[0][1])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Commit Files
|
|
# ---------------------------------------------------------------------------
|
|
class TestCommitFiles(unittest.TestCase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_commit_files_success(self, _auth, mock_api):
|
|
mock_api.return_value = {
|
|
"commit": {"sha": "commit-sha-123"},
|
|
"branch": {"name": "test-branch"}
|
|
}
|
|
files = [
|
|
{"operation": "create", "path": "test.txt", "content": "SGVsbG8="}
|
|
]
|
|
result = gitea_commit_files(
|
|
files=files,
|
|
message="Initial commit",
|
|
new_branch="test-branch"
|
|
)
|
|
self.assertTrue(result["success"])
|
|
self.assertEqual(result["commit"], "commit-sha-123")
|
|
self.assertEqual(result["branch"], "test-branch")
|
|
|
|
# Verify POST method and payload
|
|
call_args = mock_api.call_args
|
|
self.assertEqual(call_args[0][0], "POST")
|
|
self.assertIn("/contents", call_args[0][1])
|
|
payload = call_args[0][3]
|
|
self.assertEqual(payload["message"], "Initial commit")
|
|
self.assertEqual(payload["new_branch"], "test-branch")
|
|
self.assertEqual(payload["files"], files)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|