246 lines
8.8 KiB
Python
246 lines
8.8 KiB
Python
"""Issue #525: reconciler supersession close path."""
|
|
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
import mcp_server # noqa: E402
|
|
import role_session_router # noqa: E402
|
|
import task_capability_map # noqa: E402
|
|
from review_proofs import assess_reconciler_supersession_close_gate # noqa: E402
|
|
|
|
|
|
HEAD = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
|
|
MERGE = "1111111111111111111111111111111111111111"
|
|
TARGET = "2222222222222222222222222222222222222222"
|
|
|
|
|
|
def _target_pr(mergeable=False):
|
|
return {
|
|
"number": 490,
|
|
"state": "open",
|
|
"mergeable": mergeable,
|
|
"head": {"sha": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
|
|
}
|
|
|
|
|
|
def _superseding_pr():
|
|
return {
|
|
"number": 502,
|
|
"state": "closed",
|
|
"merged": True,
|
|
"merge_commit_sha": MERGE,
|
|
"head": {"sha": HEAD},
|
|
}
|
|
|
|
|
|
class TestReconcilerSupersessionCapabilityMap(unittest.TestCase):
|
|
def test_supersession_tasks_route_to_reconciler(self):
|
|
self.assertEqual(
|
|
task_capability_map.required_permission(
|
|
"reconcile_close_superseded_pr"),
|
|
"gitea.pr.close",
|
|
)
|
|
self.assertEqual(
|
|
task_capability_map.required_role("reconcile_close_superseded_pr"),
|
|
"reconciler",
|
|
)
|
|
self.assertEqual(
|
|
task_capability_map.required_permission(
|
|
"reconcile_close_satisfied_issue"),
|
|
"gitea.issue.close",
|
|
)
|
|
self.assertEqual(
|
|
task_capability_map.required_role("reconcile_create_followup_issue"),
|
|
"reconciler",
|
|
)
|
|
|
|
def test_author_session_cannot_route_supersession_close(self):
|
|
result = role_session_router.route_task_session(
|
|
"reconcile_close_superseded_pr",
|
|
active_profile="prgs-author",
|
|
active_role_kind="author",
|
|
allowed_in_current_session=False,
|
|
)
|
|
self.assertEqual(result["route_result"], "wrong_role_stop")
|
|
self.assertFalse(result["downstream_allowed"])
|
|
|
|
|
|
class TestReconcilerSupersessionCloseGate(unittest.TestCase):
|
|
def _gate(self, **overrides):
|
|
kwargs = {
|
|
"target_pr_number": 490,
|
|
"target_pr_state": "open",
|
|
"target_pr_mergeable": False,
|
|
"target_independently_required": False,
|
|
"superseding_pr_number": 502,
|
|
"superseding_pr_state": "closed",
|
|
"superseding_pr_merged": True,
|
|
"superseding_merge_commit_sha": MERGE,
|
|
"superseding_head_sha": HEAD,
|
|
"target_branch": "master",
|
|
"target_branch_sha": TARGET,
|
|
"superseding_head_is_ancestor_of_target": True,
|
|
"canonical_comment_valid": True,
|
|
"canonical_comment_mentions_superseding_pr": True,
|
|
"canonical_comment_mentions_merge_commit": True,
|
|
"close_pr_capability": True,
|
|
"close_issue_capability": True,
|
|
"target_issue_state": "open",
|
|
"issue_satisfied_by_superseding": True,
|
|
}
|
|
kwargs.update(overrides)
|
|
return assess_reconciler_supersession_close_gate(**kwargs)
|
|
|
|
def test_supersession_close_allowed_with_full_proof(self):
|
|
result = self._gate()
|
|
self.assertEqual(result["outcome"], "SUPERSESSION_CLOSE_ALLOWED")
|
|
self.assertTrue(result["pr_close_allowed"])
|
|
self.assertTrue(result["issue_close_allowed"])
|
|
self.assertFalse(result["review_allowed"])
|
|
self.assertFalse(result["merge_allowed"])
|
|
|
|
def test_mergeable_target_still_blocks(self):
|
|
result = self._gate(target_pr_mergeable=True)
|
|
self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
|
|
self.assertFalse(result["pr_close_allowed"])
|
|
self.assertTrue(any("mergeable" in r for r in result["reasons"]))
|
|
|
|
def test_requires_canonical_comment_with_merge_commit(self):
|
|
result = self._gate(canonical_comment_mentions_merge_commit=False)
|
|
self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
|
|
self.assertFalse(result["pr_close_allowed"])
|
|
self.assertTrue(any("merge commit" in r for r in result["reasons"]))
|
|
|
|
def test_superseding_head_must_be_on_target_branch(self):
|
|
result = self._gate(superseding_head_is_ancestor_of_target=False)
|
|
self.assertEqual(result["outcome"], "SUPERSEDING_PR_NOT_ON_TARGET")
|
|
self.assertFalse(result["pr_close_allowed"])
|
|
|
|
def test_close_capability_required(self):
|
|
result = self._gate(close_pr_capability=False)
|
|
self.assertEqual(result["outcome"], "RECOVERY_HANDOFF_REQUIRED")
|
|
self.assertFalse(result["pr_close_allowed"])
|
|
|
|
|
|
class TestReconcilerSupersessionMcpTool(unittest.TestCase):
|
|
def _comment(self):
|
|
return f"""## Canonical PR State
|
|
STATE: superseded
|
|
NEXT_ACTION: Close superseded PR #490 and issue #470; PR #502 is canonical.
|
|
NEXT_ACTOR: prgs-reconciler
|
|
SUMMARY: PR #490 is superseded by merged PR #502 at merge commit {MERGE}.
|
|
CANONICAL_ITEM: PR #502
|
|
SUPERSEDED_ITEM: PR #490
|
|
CLOSE_OR_KEEP_OPEN: close superseded PR #490 and satisfied issue #470
|
|
"""
|
|
|
|
@patch("mcp_server.verify_preflight_purity")
|
|
@patch("mcp_server._canonical_comment_gate")
|
|
@patch("already_landed_reconcile.is_head_ancestor_of_ref", return_value=True)
|
|
@patch("already_landed_reconcile.fetch_target_branch")
|
|
@patch("mcp_server._profile_operation_gate", return_value=[])
|
|
@patch("mcp_server._auth", return_value="token test")
|
|
@patch("mcp_server.api_request")
|
|
def test_tool_posts_comment_and_closes_superseded_pr_issue(
|
|
self,
|
|
api,
|
|
_auth,
|
|
_profile_gate,
|
|
fetch_target,
|
|
_ancestor,
|
|
canonical_gate,
|
|
verify_preflight,
|
|
):
|
|
fetch_target.return_value = {
|
|
"success": True,
|
|
"target_branch": "master",
|
|
"target_ref": "prgs/master",
|
|
"target_branch_sha": TARGET,
|
|
}
|
|
canonical_gate.return_value = {
|
|
"blocked": False,
|
|
"canonical_comment_validation": {"valid": True},
|
|
}
|
|
api.side_effect = [
|
|
_target_pr(),
|
|
_superseding_pr(),
|
|
{"number": 470, "state": "open"},
|
|
{"id": 123},
|
|
{"state": "closed"},
|
|
{"state": "closed"},
|
|
]
|
|
|
|
result = mcp_server.gitea_reconcile_superseded_by_merged_pr(
|
|
target_pr_number=490,
|
|
superseding_pr_number=502,
|
|
target_issue_number=470,
|
|
target_independently_required=False,
|
|
issue_satisfied_by_superseding=True,
|
|
close_pr=True,
|
|
close_issue=True,
|
|
post_comment=True,
|
|
comment_body=self._comment(),
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
)
|
|
|
|
self.assertTrue(result["success"])
|
|
self.assertTrue(result["performed"])
|
|
self.assertTrue(result["pr_comment_posted"])
|
|
self.assertTrue(result["pr_closed"])
|
|
self.assertTrue(result["issue_closed"])
|
|
verify_preflight.assert_called_once_with(
|
|
"prgs", task="reconcile_close_superseded_pr")
|
|
|
|
@patch("mcp_server.verify_preflight_purity")
|
|
@patch("mcp_server._canonical_comment_gate")
|
|
@patch("already_landed_reconcile.is_head_ancestor_of_ref", return_value=True)
|
|
@patch("already_landed_reconcile.fetch_target_branch")
|
|
@patch("mcp_server._profile_operation_gate", return_value=[])
|
|
@patch("mcp_server._auth", return_value="token test")
|
|
@patch("mcp_server.api_request")
|
|
def test_tool_does_not_mutate_when_target_still_mergeable(
|
|
self,
|
|
api,
|
|
_auth,
|
|
_profile_gate,
|
|
fetch_target,
|
|
_ancestor,
|
|
canonical_gate,
|
|
verify_preflight,
|
|
):
|
|
fetch_target.return_value = {
|
|
"success": True,
|
|
"target_branch": "master",
|
|
"target_ref": "prgs/master",
|
|
"target_branch_sha": TARGET,
|
|
}
|
|
canonical_gate.return_value = {
|
|
"blocked": False,
|
|
"canonical_comment_validation": {"valid": True},
|
|
}
|
|
api.side_effect = [_target_pr(mergeable=True), _superseding_pr()]
|
|
|
|
result = mcp_server.gitea_reconcile_superseded_by_merged_pr(
|
|
target_pr_number=490,
|
|
superseding_pr_number=502,
|
|
target_independently_required=False,
|
|
close_pr=True,
|
|
post_comment=True,
|
|
comment_body=self._comment(),
|
|
remote="prgs",
|
|
org="Scaled-Tech-Consulting",
|
|
repo="Gitea-Tools",
|
|
)
|
|
|
|
self.assertFalse(result["success"])
|
|
self.assertFalse(result["performed"])
|
|
self.assertFalse(result["pr_closed"])
|
|
self.assertFalse(result["pr_comment_posted"])
|
|
verify_preflight.assert_not_called()
|
|
self.assertEqual(api.call_count, 2)
|