Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

Merge pull request 'feat: add prgs-reconciler profile and gated already-landed PR close (Closes #310)' (#381) from feat/issue-310-prgs-reconciler-profile into master

Browse Source
This commit was merged in pull request #381.
This commit is contained in:
sysadmin
2026-07-07 09:22:55 -05:00
parent 7db8298cdd 167257b8a3
commit 4243b60ca3
12 changed files with 720 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_already_landed_reconcile.py
+131
View File
@@ -0,0 +1,131 @@
"""Tests for already-landed PR reconciliation gates (#310)."""
import os
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import already_landed_reconcile
class TestAssessAlreadyLandedReconciliation(unittest.TestCase):
def test_open_pr_already_landed_allows_close(self):
with patch(
"already_landed_reconcile.is_head_ancestor_of_ref",
return_value=True,
):
assessment = already_landed_reconcile.assess_already_landed_reconciliation(
pr={
"number": 278,
"state": "open",
"title": "Fix thing (Closes #263)",
"body": "",
"head": {"ref": "feat/x", "sha": "abc123"},
"base": {"ref": "master"},
},
project_root="/tmp/repo",
remote="prgs",
target_branch="master",
target_fetch={
"success": True,
"target_branch": "master",
"target_ref": "prgs/master",
"target_branch_sha": "deadbeef",
"git_fetch_command": "git fetch prgs master",
},
)
self.assertTrue(assessment["close_allowed"])
self.assertEqual(
assessment["eligibility_class"],
already_landed_reconcile.ELIGIBILITY_ALREADY_LANDED,
)
self.assertEqual(assessment["linked_issue"], 263)
self.assertTrue(assessment["ancestor_proof"])
def test_not_landed_pr_denies_close(self):
with patch(
"already_landed_reconcile.is_head_ancestor_of_ref",
return_value=False,
):
assessment = already_landed_reconcile.assess_already_landed_reconciliation(
pr={
"number": 99,
"state": "open",
"title": "WIP",
"body": "",
"head": {"ref": "feat/y", "sha": "fff111"},
"base": {"ref": "master"},
},
project_root="/tmp/repo",
remote="prgs",
target_branch="master",
target_fetch={
"success": True,
"target_branch": "master",
"target_ref": "prgs/master",
"target_branch_sha": "deadbeef",
},
)
self.assertFalse(assessment["close_allowed"])
self.assertEqual(
assessment["eligibility_class"],
already_landed_reconcile.ELIGIBILITY_NOT_LANDED,
)
def test_stale_target_branch_denies_close(self):
assessment = already_landed_reconcile.assess_already_landed_reconciliation(
pr={
"number": 99,
"state": "open",
"title": "WIP",
"body": "",
"head": {"ref": "feat/y", "sha": "fff111"},
"base": {"ref": "master"},
},
project_root="/tmp/repo",
remote="prgs",
target_branch="master",
target_fetch={
"success": False,
"target_branch": "master",
"target_ref": "prgs/master",
"target_branch_sha": None,
"reasons": ["git fetch failed"],
},
)
self.assertFalse(assessment["close_allowed"])
self.assertEqual(
assessment["eligibility_class"],
already_landed_reconcile.ELIGIBILITY_STALE_TARGET,
)
def test_closed_pr_denies_close(self):
assessment = already_landed_reconcile.assess_already_landed_reconciliation(
pr={
"number": 50,
"state": "closed",
"title": "Done",
"body": "",
"head": {"ref": "feat/z", "sha": "aaa"},
"base": {"ref": "master"},
},
project_root="/tmp/repo",
remote="prgs",
target_branch="master",
target_fetch={
"success": True,
"target_branch": "master",
"target_ref": "prgs/master",
"target_branch_sha": "deadbeef",
},
)
self.assertFalse(assessment["close_allowed"])
self.assertEqual(
assessment["eligibility_class"],
already_landed_reconcile.ELIGIBILITY_PR_NOT_OPEN,
)
if __name__ == "__main__":
unittest.main()
tests/test_reconcile_already_landed_pr_tool.py
+149
View File
@@ -0,0 +1,149 @@
"""Tests for gitea_reconcile_already_landed_pr MCP tool (#310)."""
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_reconcile_already_landed_pr
RECONCILER_PROFILE = {
"profile_name": "prgs-reconciler",
"allowed_operations": [
"gitea.read",
"gitea.pr.comment",
"gitea.issue.comment",
"gitea.issue.close",
"gitea.pr.close",
],
"forbidden_operations": [
"gitea.pr.approve",
"gitea.pr.merge",
"gitea.pr.create",
"gitea.branch.push",
],
"audit_label": "prgs-reconciler",
}
AUTHOR_PROFILE = {
"profile_name": "prgs-author",
"allowed_operations": [
"gitea.read",
"gitea.pr.create",
"gitea.issue.comment",
],
"forbidden_operations": ["gitea.pr.close", "gitea.pr.approve", "gitea.pr.merge"],
"audit_label": "prgs-author",
}
OPEN_LANDED_PR = {
"number": 278,
"title": "Already landed (Closes #263)",
"body": "",
"state": "open",
"head": {"ref": "feat/issue-263", "sha": "2a544b78d1371925761d16f1c451bb3b2984470e"},
"base": {"ref": "master"},
}
class TestReconcileAlreadyLandedPrTool(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.verify_preflight_purity").start()
patch("gitea_audit.audit_enabled", return_value=False).start()
mcp_server._IDENTITY_CACHE.clear()
def tearDown(self):
patch.stopall()
mcp_server._IDENTITY_CACHE.clear()
def _set_profile(self, profile):
patch("mcp_server.get_profile", return_value=profile).start()
def test_close_blocked_without_pr_close_capability(self):
self._set_profile(AUTHOR_PROFILE)
res = gitea_reconcile_already_landed_pr(
pr_number=278,
close_pr=True,
remote="prgs",
)
self.assertFalse(res["success"])
self.assertFalse(res["performed"])
self.assertEqual(res["required_permission"], "gitea.pr.close")
self.mock_api.assert_not_called()
def test_not_landed_pr_close_denied_after_live_fetch(self):
self._set_profile(RECONCILER_PROFILE)
self.mock_api.return_value = OPEN_LANDED_PR
with patch(
"mcp_server.already_landed_reconcile.assess_already_landed_reconciliation",
return_value={
"pr_state": "open",
"candidate_head_sha": OPEN_LANDED_PR["head"]["sha"],
"target_branch": "master",
"target_branch_sha": "deadbeef",
"ancestor_proof": False,
"eligibility_class": "NOT_ALREADY_LANDED",
"linked_issue": 263,
"close_allowed": False,
"git_ref_mutations": ["git fetch prgs master"],
"reasons": ["not ancestor"],
},
):
res = gitea_reconcile_already_landed_pr(
pr_number=278,
close_pr=True,
remote="prgs",
)
self.assertFalse(res["success"])
self.assertFalse(res.get("pr_closed"))
patch_calls = [
c for c in self.mock_api.call_args_list if c.args[0] == "PATCH"
]
self.assertEqual(patch_calls, [])
def test_already_landed_close_allowed_with_capability(self):
self._set_profile(RECONCILER_PROFILE)
def api_side_effect(method, url, auth, payload=None):
if method == "GET" and "/pulls/278" in url:
return dict(OPEN_LANDED_PR)
if method == "PATCH" and "/pulls/278" in url:
closed = dict(OPEN_LANDED_PR)
closed["state"] = "closed"
return closed
return {}
self.mock_api.side_effect = api_side_effect
with patch(
"mcp_server.already_landed_reconcile.assess_already_landed_reconciliation",
return_value={
"pr_state": "open",
"candidate_head_sha": OPEN_LANDED_PR["head"]["sha"],
"target_branch": "master",
"target_branch_sha": "e017d80256217f17b0f421cd051cfa5cbcf5ae0f",
"ancestor_proof": True,
"eligibility_class": "ALREADY_LANDED_RECONCILE_REQUIRED",
"linked_issue": 263,
"close_allowed": True,
"git_ref_mutations": ["git fetch prgs master"],
"reasons": [],
},
):
res = gitea_reconcile_already_landed_pr(
pr_number=278,
close_pr=True,
remote="prgs",
)
self.assertTrue(res["success"])
self.assertTrue(res["performed"])
self.assertTrue(res["pr_closed"])
self.assertEqual(res["eligibility_class"], "ALREADY_LANDED_RECONCILE_REQUIRED")
if __name__ == "__main__":
unittest.main()
tests/test_resolve_task_capability.py
+12
View File
@@ -275,6 +275,18 @@ class TestResolveTaskCapability(unittest.TestCase):
self.assertEqual(res["required_role_kind"], "author")
self.assertIn("author", res["exact_safe_next_action"].lower())
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolve_reconcile_already_landed_pr_requires_close(self, _auth, _api):
with patch.dict(os.environ, self._env("author-profile")):
res = mcp_server.gitea_resolve_task_capability(
task="reconcile_already_landed_pr", remote="prgs"
)
self.assertEqual(res["requested_task"], "reconcile_already_landed_pr")
self.assertEqual(res["required_operation_permission"], "gitea.pr.close")
self.assertEqual(res["required_role_kind"], "reconciler")
self.assertFalse(res["allowed_in_current_session"])
@patch("mcp_server.api_request", return_value={"login": "author-user"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_resolve_pr_queue_cleanup_author_profile_blocked(self, _auth, _api):
tests/test_role_session_router.py
+35
View File
@@ -52,6 +52,22 @@ CONFIG = {
],
"execution_profile": "prgs-reviewer",
},
"prgs-reconciler": {
"enabled": True,
"context": "ctx",
"role": "reconciler",
"username": "sysadmin",
"auth": {"type": "env", "name": "GITEA_TOKEN_RECONCILER"},
"allowed_operations": [
"gitea.read", "gitea.pr.comment", "gitea.issue.comment",
"gitea.issue.close", "gitea.pr.close",
],
"forbidden_operations": [
"gitea.pr.approve", "gitea.pr.merge", "gitea.pr.create",
"gitea.branch.push",
],
"execution_profile": "prgs-reconciler",
},
},
"rules": {"allow_runtime_switching": False},
}
@@ -88,6 +104,7 @@ class TestRoleSessionRouter(unittest.TestCase):
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
"GITEA_TOKEN_RECONCILER": "reconciler-pass",
}
def test_reviewer_task_under_author_profile_wrong_role_stop(self):
@@ -197,6 +214,24 @@ class TestRoleSessionRouter(unittest.TestCase):
)
self.assertFalse(route["downstream_allowed"])
@patch("mcp_server.api_request", return_value={"login": "sysadmin"})
@patch("mcp_server.get_auth_header", return_value="token reconciler-pass")
def test_reconciler_task_under_reconciler_profile_allowed(self, _auth, _api):
with patch.dict(os.environ, self._env("prgs-reconciler")):
route = mcp_server.gitea_route_task_session(
task_type="reconcile_already_landed_pr", remote="prgs"
)
self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
self.assertTrue(route["downstream_allowed"])
def test_reconciler_task_under_author_profile_wrong_role_stop(self):
with patch.dict(os.environ, self._env("prgs-author")):
route = mcp_server.gitea_route_task_session(
task_type="reconcile_already_landed_pr", remote="prgs"
)
self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
self.assertFalse(route["downstream_allowed"])
def test_activate_profile_blocked_in_static_mode(self):
with patch.dict(os.environ, self._env("prgs-author")):
result = mcp_server.gitea_activate_profile(