Files
Gitea-Tools/tests/test_reconciliation_mcp_tools.py
T
sysadmin 749e480baf feat: add reconciliation workflow MCP tools for already-landed PRs (Closes #301)
Expose read-only assess/scan tools and capability planning so already-landed
open PRs can be reconciled without review or merge. Register the
gitea-reconcile-landed-pr skill and workflow-source verification.
2026-07-07 09:25:55 -04:00

106 lines
3.5 KiB
Python

"""Tests for reconciliation MCP assessment tools (#301)."""
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import mcp_server
from mcp_server import (
gitea_assess_already_landed_reconciliation,
gitea_scan_already_landed_open_prs,
)
AUTHOR_PROFILE = {
"profile_name": "prgs-author",
"allowed_operations": [
"gitea.read",
"gitea.pr.comment",
"gitea.issue.comment",
"gitea.issue.close",
],
"forbidden_operations": ["gitea.pr.close"],
"audit_label": "prgs-author",
}
OPEN_PR = {
"number": 278,
"title": "Landed (Closes #263)",
"body": "",
"state": "open",
"head": {"ref": "feat/x", "sha": "a" * 40},
"base": {"ref": "master"},
}
class TestReconciliationMcpTools(unittest.TestCase):
def setUp(self):
self.mock_api = patch("mcp_server.api_request").start()
self.mock_auth = patch(
"mcp_server.get_auth_header", return_value="token test"
).start()
patch("mcp_server.get_profile", return_value=AUTHOR_PROFILE).start()
mcp_server._IDENTITY_CACHE.clear()
def tearDown(self):
patch.stopall()
mcp_server._IDENTITY_CACHE.clear()
def test_assess_returns_plan_without_mutations(self):
self.mock_api.return_value = OPEN_PR
with patch(
"mcp_server.reconciliation_workflow.assess_open_pr_reconciliation",
return_value={
"reconciliation_allowed": True,
"eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
"candidate_head_sha": OPEN_PR["head"]["sha"],
"target_branch_sha": "deadbeef",
"linked_issue": 263,
"review_merge_allowed": False,
},
):
res = gitea_assess_already_landed_reconciliation(
pr_number=278, remote="prgs"
)
self.assertTrue(res["success"])
self.assertFalse(res["performed"])
self.assertEqual(res["plan"]["outcome"], "PARTIAL_RECONCILE_COMMENT_THEN_STOP")
self.assertFalse(res["review_merge_allowed"])
patch_calls = [
c for c in self.mock_api.call_args_list if c.args[0] == "PATCH"
]
self.assertEqual(patch_calls, [])
def test_scan_returns_candidates_with_pagination(self):
with patch(
"mcp_server.api_fetch_page",
return_value=([OPEN_PR], {
"page": 1,
"per_page": 50,
"is_final_page": True,
"has_more": False,
"next_page": None,
}),
), patch(
"mcp_server.reconciliation_workflow.fetch_target_branch",
return_value={
"success": True,
"target_branch_sha": "deadbeef",
"git_fetch_command": "git fetch prgs master",
},
), patch(
"mcp_server.reconciliation_workflow.assess_open_pr_reconciliation",
return_value={
"pr_number": 278,
"eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
"reconciliation_allowed": True,
},
):
res = gitea_scan_already_landed_open_prs(remote="prgs", limit=50)
self.assertTrue(res["success"])
self.assertEqual(res["candidate_count"], 1)
self.assertTrue(res["pagination"]["inventory_complete"])
if __name__ == "__main__":
unittest.main()