Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 17 Pull Requests 1 Actions Packages Projects Releases Wiki Activity
Files
5aad2e62d99c7ef71777e4aa8c96b73444d55839
Gitea-Tools/tests/test_mcp_server.py
T
sysadmin 03e28c159e feat: add read-only gitea_whoami authenticated-user lookup (#11) 
Add a read-only MCP tool that calls Gitea's authenticated-user
endpoint (GET /api/v1/user) and returns safe identity metadata only:
username, display name, user id, email, server, and remote.

This lets future review/merge workflows prove which Gitea account the
MCP server is authenticated as, so self-review/self-merge can be
detected before acting — the blocker discovered during PR #8 dogfooding.

- Never returns the token, Authorization header, password, or secrets.
- Fails closed with a clear error if identity cannot be determined.
- No mutation; no profile switching; no review/approve/merge behavior.

Tests: identity mapping, secret-redaction, fail-closed, unknown-remote.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-01 12:42:37 -04:00

519 lines
20 KiB
Python
Raw Blame History

"""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,
gitea_whoami,
)
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)
# ---------------------------------------------------------------------------
# Whoami (authenticated-user identity lookup)
# ---------------------------------------------------------------------------
class TestWhoami(unittest.TestCase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_returns_safe_identity(self, _auth, mock_api):
mock_api.return_value = {
"id": 42,
"login": "reviewer-bot",
"full_name": "Reviewer Bot",
"email": "reviewer@example.com",
}
result = gitea_whoami(remote="prgs")
self.assertTrue(result["authenticated"])
self.assertEqual(result["username"], "reviewer-bot")
self.assertEqual(result["display_name"], "Reviewer Bot")
self.assertEqual(result["user_id"], 42)
self.assertEqual(result["server"], "https://gitea.prgs.cc")
self.assertEqual(result["remote"], "prgs")
# Read-only: GET against the authenticated-user endpoint.
call_args = mock_api.call_args
self.assertEqual(call_args[0][0], "GET")
self.assertTrue(call_args[0][1].endswith("/api/v1/user"))
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_never_exposes_secrets(self, _auth, mock_api):
mock_api.return_value = {"id": 1, "login": "someone"}
result = gitea_whoami(remote="prgs")
blob = repr(result).lower()
for secret in ("token", "authorization", "basic ", "password", FAKE_AUTH.lower()):
self.assertNotIn(secret, blob)
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_fails_closed_without_login(self, _auth, mock_api):
mock_api.return_value = {"id": 1} # no 'login'
with self.assertRaises(RuntimeError):
gitea_whoami(remote="prgs")
def test_rejects_unknown_remote(self):
with self.assertRaises(ValueError):
gitea_whoami(remote="nope")
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink