"""Merger lease adoption / recovery (#536).""" import os import sys import unittest from datetime import datetime, timezone from unittest.mock import patch sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) import merger_lease_adoption as mla import reviewer_pr_lease as leases def _lease_comment( pr_number: int, session_id: str, *, phase: str = "claimed", profile: str = "prgs-reviewer", identity: str = "sysadmin", candidate_head: str = "a" * 40, comment_id: int = 100, ) -> dict: body = leases.format_lease_body( repo="Scaled-Tech-Consulting/Gitea-Tools", pr_number=pr_number, issue_number=536, reviewer_identity=identity, profile=profile, session_id=session_id, worktree="branches/review-pr536", phase=phase, candidate_head=candidate_head, target_branch="master", target_branch_sha="b" * 40, last_activity=datetime.now(timezone.utc), ) return {"id": comment_id, "body": body, "user": {"login": identity}} HEAD = "f" * 40 class TestSanctionedProvenance(unittest.TestCase): def setUp(self): leases.clear_session_lease() def test_manual_session_seed_rejected_by_mutation_gate(self): comments = [_lease_comment(536, "reviewer-session", candidate_head=HEAD)] leases.record_session_lease({ "pr_number": 536, "session_id": "merger-session", "candidate_head": HEAD, "comment_id": 999, }) result = leases.assess_mutation_lease_gate( pr_number=536, comments=comments, reviewer_identity="sysadmin", session_id="merger-session", mutation="merge", live_head_sha=HEAD, pinned_head_sha=HEAD, ) self.assertTrue(result["block"]) self.assertTrue( any("sanctioned provenance" in r for r in result["reasons"]) ) def test_sanctioned_acquire_provenance_allows_mutation(self): comments = [_lease_comment(536, "my-session", candidate_head=HEAD)] leases.record_session_lease( { "pr_number": 536, "session_id": "my-session", "candidate_head": HEAD, "comment_id": 101, }, lease_provenance=mla.build_lease_provenance( source=mla.SOURCE_ACQUIRE, comment_id=101, ), ) result = leases.assess_mutation_lease_gate( pr_number=536, comments=comments, reviewer_identity="sysadmin", session_id="my-session", mutation="merge", live_head_sha=HEAD, pinned_head_sha=HEAD, ) self.assertFalse(result["block"]) class TestAdoptAssessment(unittest.TestCase): def test_adopt_allowed_for_merger_with_reviewer_lease_and_approval(self): comments = [_lease_comment(536, "reviewer-sess", candidate_head=HEAD)] result = mla.assess_adopt_merger_lease( comments, pr_number=536, adopter_identity="sysadmin", adopter_profile="prgs-merger", adopter_session_id="merger-sess", repo="Scaled-Tech-Consulting/Gitea-Tools", issue_number=536, worktree="branches/merge-pr536", expected_head_sha=HEAD, live_head_sha=HEAD, approval_at_head=True, pr_open=True, ) self.assertTrue(result["adopt_allowed"]) self.assertIn(mla.ADOPTION_MARKER, result["adoption_body"]) self.assertIn("adopted_from_session_id: reviewer-sess", result["adoption_body"]) def test_adopt_blocked_without_approval(self): comments = [_lease_comment(536, "reviewer-sess", candidate_head=HEAD)] result = mla.assess_adopt_merger_lease( comments, pr_number=536, adopter_identity="sysadmin", adopter_profile="prgs-merger", adopter_session_id="merger-sess", repo="Scaled-Tech-Consulting/Gitea-Tools", issue_number=536, worktree="branches/merge-pr536", expected_head_sha=HEAD, live_head_sha=HEAD, approval_at_head=False, pr_open=True, ) self.assertFalse(result["adopt_allowed"]) self.assertTrue(any("APPROVED" in r for r in result["reasons"])) def test_adopt_blocked_for_non_merger_profile(self): comments = [_lease_comment(536, "reviewer-sess", candidate_head=HEAD)] result = mla.assess_adopt_merger_lease( comments, pr_number=536, adopter_identity="sysadmin", adopter_profile="prgs-reviewer", adopter_session_id="merger-sess", repo="Scaled-Tech-Consulting/Gitea-Tools", issue_number=536, worktree="branches/review-pr536", expected_head_sha=HEAD, live_head_sha=HEAD, approval_at_head=True, pr_open=True, ) self.assertFalse(result["adopt_allowed"]) self.assertTrue(any("merger profile" in r for r in result["reasons"])) def test_adopt_blocked_when_pr_not_open(self): comments = [_lease_comment(536, "reviewer-sess", candidate_head=HEAD)] result = mla.assess_adopt_merger_lease( comments, pr_number=536, adopter_identity="sysadmin", adopter_profile="prgs-merger", adopter_session_id="merger-sess", repo="Scaled-Tech-Consulting/Gitea-Tools", issue_number=536, worktree="branches/merge-pr536", expected_head_sha=HEAD, live_head_sha=HEAD, approval_at_head=True, pr_open=False, ) self.assertFalse(result["adopt_allowed"]) class TestMergerHandoffMutationGate(unittest.TestCase): def setUp(self): leases.clear_session_lease() def test_cross_session_merge_blocked_without_adoption_comment(self): reviewer_comment = _lease_comment(536, "reviewer-sess", candidate_head=HEAD) provenance = mla.build_lease_provenance( source=mla.SOURCE_ADOPT, comment_id=7001, adopted_from_session_id="reviewer-sess", ) leases.record_session_lease( { "pr_number": 536, "session_id": "merger-sess", "candidate_head": HEAD, "comment_id": 7001, "phase": "adopted", }, lease_provenance=provenance, ) result = leases.assess_mutation_lease_gate( pr_number=536, comments=[reviewer_comment], reviewer_identity="sysadmin", session_id="merger-sess", mutation="merge", live_head_sha=HEAD, pinned_head_sha=HEAD, ) self.assertTrue(result["block"]) self.assertTrue(any("owned by session_id=reviewer-sess" in r for r in result["reasons"])) def test_cross_session_merge_allowed_after_adoption_comment(self): reviewer_comment = _lease_comment(536, "reviewer-sess", candidate_head=HEAD) adoption_body = mla.format_adoption_body( repo="Scaled-Tech-Consulting/Gitea-Tools", pr_number=536, issue_number=536, adopter_identity="sysadmin", adopter_profile="prgs-merger", adopter_session_id="merger-sess", worktree="branches/merge-pr536", candidate_head=HEAD, target_branch="master", target_branch_sha="b" * 40, adopted_from_session_id="reviewer-sess", adopted_from_profile="prgs-reviewer", adopted_from_reviewer_identity="sysadmin", adopted_from_comment_id=100, ) adoption_comment = { "id": 7001, "body": adoption_body, "user": {"login": "sysadmin"}, } provenance = mla.build_lease_provenance( source=mla.SOURCE_ADOPT, comment_id=7001, adopted_from_session_id="reviewer-sess", ) leases.record_session_lease( { "pr_number": 536, "session_id": "merger-sess", "candidate_head": HEAD, "comment_id": 7001, "phase": "adopted", }, lease_provenance=provenance, ) result = leases.assess_mutation_lease_gate( pr_number=536, comments=[reviewer_comment, adoption_comment], reviewer_identity="sysadmin", session_id="merger-sess", mutation="merge", live_head_sha=HEAD, pinned_head_sha=HEAD, ) self.assertFalse(result["block"]) class TestAdoptTool(unittest.TestCase): def setUp(self): leases.clear_session_lease() patch("mcp_server.verify_preflight_purity", return_value=None).start() patch("mcp_server._verify_role_mutation_workspace", return_value=None).start() patch("gitea_audit.audit_enabled", return_value=False).start() mcp_server = __import__("mcp_server") mcp_server._IDENTITY_CACHE.clear() mcp_server.init_review_decision_lock("prgs", "adopt_merger_pr_lease") mcp_server.record_preflight_check("whoami") mcp_server.record_preflight_check( "capability", "reviewer", resolved_task="adopt_merger_pr_lease" ) def tearDown(self): patch.stopall() leases.clear_session_lease() @patch("mcp_server._authenticated_username", return_value="sysadmin") @patch("mcp_server.get_auth_header", return_value="Basic dGVzdDp0ZXN0") @patch("mcp_server.get_profile") @patch("mcp_server.api_request") def test_adopt_tool_posts_proof_and_records_sanctioned_session( self, mock_api, mock_profile, _mock_auth, _mock_user ): mock_profile.return_value = { "profile_name": "prgs-merger", "allowed_operations": [ "gitea.read", "gitea.pr.comment", "gitea.pr.merge", ], "forbidden_operations": [], } reviewer_comment = _lease_comment(536, "reviewer-sess", candidate_head=HEAD) posted_bodies: list[str] = [] def _side(method, url, auth=None, payload=None, *a, **k): m = (method or "").upper() if m == "POST": if payload and payload.get("body"): posted_bodies.append(payload["body"]) return {"id": 7001} if "/pulls/" in url and "/files" not in url: return { "state": "open", "number": 536, "head": {"sha": HEAD}, "merged": False, } if "/reviews" in url: return [] if "/comments" in url: return [reviewer_comment] return {} mock_api.side_effect = _side from mcp_server import gitea_adopt_merger_pr_lease with patch.object( __import__("mcp_server"), "gitea_get_pr_review_feedback", return_value={ "approval_at_current_head": True, "latest_approved_head_sha": HEAD, }, ): with patch.dict( os.environ, { "GITEA_PROFILE_NAME": "prgs-merger", "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment,gitea.pr.merge", }, clear=True, ): result = gitea_adopt_merger_pr_lease( pr_number=536, worktree="branches/merge-pr536", expected_head_sha=HEAD, issue_number=536, remote="prgs", ) self.assertTrue(result["success"]) self.assertTrue(result["adopted"]) self.assertEqual(result["adopted_from_session_id"], "reviewer-sess") self.assertEqual(result["adoption_comment_id"], 7001) self.assertEqual(len(posted_bodies), 1) self.assertIn(mla.ADOPTION_MARKER, posted_bodies[0]) session = leases.get_session_lease() self.assertTrue(mla.is_sanctioned_session_lease(session)) adoption_comment = { "id": 7001, "body": posted_bodies[0], "user": {"login": "sysadmin"}, } gate = leases.assess_mutation_lease_gate( pr_number=536, comments=[reviewer_comment, adoption_comment], reviewer_identity="sysadmin", session_id=result["session_id"], mutation="merge", live_head_sha=HEAD, pinned_head_sha=HEAD, ) self.assertFalse(gate["block"]) if __name__ == "__main__": unittest.main()