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