"""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()