Gitea-Tools/tests/test_merger_lease_adoption.py

"""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 TestDescribeSessionLeaseProof(unittest.TestCase):
    def setUp(self):
        leases.clear_session_lease()

    def test_empty_session_is_none_kind(self):
        proof = mla.describe_session_lease_proof(None)
        self.assertIsNone(proof["lease_proof_source"])
        self.assertEqual(proof["lease_proof_kind"], "none")
        self.assertFalse(proof["lease_proof_sanctioned"])

    def test_manual_seed_without_provenance(self):
        leases.record_session_lease({
            "pr_number": 536,
            "session_id": "manual",
            "comment_id": 1,
        })
        proof = mla.describe_session_lease_proof(leases.get_session_lease())
        self.assertIsNone(proof["lease_proof_source"])
        self.assertEqual(proof["lease_proof_kind"], "unsanctioned_manual_seed")
        self.assertFalse(proof["lease_proof_sanctioned"])

    def test_sanctioned_adopt_fields(self):
        leases.record_session_lease(
            {
                "pr_number": 536,
                "session_id": "merger-sess",
                "comment_id": 700,
            },
            lease_provenance=mla.build_lease_provenance(
                source=mla.SOURCE_ADOPT,
                comment_id=700,
                adopted_from_session_id="reviewer-sess",
                adoption_reason=mla.DEFAULT_ADOPTION_REASON,
            ),
        )
        proof = mla.describe_session_lease_proof(leases.get_session_lease())
        self.assertEqual(proof["lease_proof_source"], mla.SOURCE_ADOPT)
        self.assertEqual(proof["lease_proof_kind"], "sanctioned_adoption")
        self.assertTrue(proof["lease_proof_sanctioned"])
        self.assertEqual(
            proof["lease_recovery_reason"], mla.DEFAULT_ADOPTION_REASON
        )
        self.assertEqual(proof["lease_proof_comment_id"], 700)
        self.assertEqual(proof["lease_adopted_from_session_id"], "reviewer-sess")

    def test_sanctioned_acquire_fields(self):
        leases.record_session_lease(
            {
                "pr_number": 536,
                "session_id": "rev",
                "comment_id": 50,
            },
            lease_provenance=mla.build_lease_provenance(
                source=mla.SOURCE_ACQUIRE,
                comment_id=50,
            ),
        )
        proof = mla.describe_session_lease_proof(leases.get_session_lease())
        self.assertEqual(proof["lease_proof_source"], mla.SOURCE_ACQUIRE)
        self.assertEqual(proof["lease_proof_kind"], "sanctioned_acquire")
        self.assertTrue(proof["lease_proof_sanctioned"])


class TestManualLeaseProofHandoffAudit(unittest.TestCase):
    def test_clean_merge_report_passes(self):
        text = (
            "Merged via gitea_merge_pr after gitea_adopt_merger_pr_lease. "
            "lease_proof_source: gitea_adopt_merger_pr_lease"
        )
        result = mla.assess_manual_lease_proof_handoff(text)
        self.assertTrue(result["proven"])
        self.assertFalse(result["block"])

    def test_manual_seeded_claim_without_mcp_evidence_blocks(self):
        text = "Merged using manually seeded lease proof from prior session."
        result = mla.assess_manual_lease_proof_handoff(text)
        self.assertFalse(result["proven"])
        self.assertTrue(result["block"])
        self.assertTrue(any("#535" in r for r in result["reasons"]))

    def test_historical_manual_mention_allowed_with_adoption_evidence(self):
        text = (
            "Historical incident used manually seeded lease; this merge used "
            "gitea_adopt_merger_pr_lease with adoption_comment_id: 8288."
        )
        result = mla.assess_manual_lease_proof_handoff(text)
        self.assertTrue(result["proven"])
        self.assertFalse(result["block"])


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 only the role workspace verifier (which internally does the single
        # verify_preflight_purity); the direct duplicate call was removed to prevent
        # capability state clearing. See fix for #548.
        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"])

    def test_adopt_preserves_preflight_capability_state_no_duplicate_clear(self):
        """Prove fix for #548: adopt calls _verify (single purity) and does not
        trigger duplicate verify that would clear capability state, avoiding
        need for temp local patches or raw fallbacks.
        """
        from unittest.mock import patch, MagicMock
        import mcp_server as mcp_server_mod

        # fresh preflight state (as controller/reconciler would have from resolve)
        mcp_server_mod.record_preflight_check("whoami")
        mcp_server_mod.record_preflight_check(
            "capability", "merger", resolved_task="adopt_merger_pr_lease"
        )
        initial_cap_called = mcp_server_mod._preflight_capability_called

        verify_patch = patch("mcp_server.verify_preflight_purity", wraps=mcp_server_mod.verify_preflight_purity)
        def _role_side(*a, **k):
            # simulate the real _verify which calls verify once with path
            mcp_server_mod.verify_preflight_purity(k.get("remote") or a[0] if a else None, worktree_path=k.get("worktree") or "branches/work", task=k.get("task"))
            return "branches/work"
        role_patch = patch("mcp_server._verify_role_mutation_workspace", side_effect=_role_side)
        with verify_patch as vmock, role_patch, \
             patch("mcp_server.get_auth_header", return_value="token test"), \
             patch("mcp_server._authenticated_username", return_value="sysadmin"):
            # minimal mocks to let adopt reach end without full api
            with patch("mcp_server.get_profile") as gp:
                gp.return_value = {
                    "profile_name": "prgs-merger",
                    "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"],
                    "forbidden_operations": [],
                }
                with patch("mcp_server.api_request") as api:
                    def _side(url, *a, **k):
                        if "/pulls/" in str(url) and "/comments" not in str(url):
                            return {"state": "open", "head": {"sha": "deadbeef"*5}, "merged": False, "number": 999}
                        if "/comments" in str(url):
                            return []  # list of comments for fetch
                        return {"id": 42}
                    api.side_effect = _side
                    with patch("mcp_server.gitea_get_pr_review_feedback", return_value={"approval_at_current_head": True}):
                        with patch("merger_lease_adoption.assess_adopt_merger_lease", return_value={
                            "adopt_allowed": True,
                            "active_lease": {"session_id": "rev", "profile": "prgs-reviewer", "reviewer_identity": "sysadmin", "issue_number": 548, "target_branch": "master"},
                            "adoption_body": "<!-- mcp-review-lease-adoption:v1 --> adopted",
                        }):
                            res = mcp_server_mod.gitea_adopt_merger_pr_lease(
                                pr_number=999,
                                worktree="branches/merge-test",
                                expected_head_sha="deadbeef"*5,
                                issue_number=548,
                                remote="prgs",
                            )
        self.assertTrue(res.get("success"))
        # verify_preflight_purity should have been invoked exactly once (via the _verify_role path)
        # not twice (old dup would have cleared state mid-way, requiring temp patches)
        self.assertEqual(vmock.call_count, 1)
        # capability state management: the single verify consumes (clears) as designed;
        # prior state was valid, no unexpected clear between "checks" or hidden reset
        # (the removed dup line was the source of the #548 clearing bug)
        self.assertTrue(initial_cap_called)
        # final may be set by other records in flow, but key is single invocation and success without bypasses

if __name__ == "__main__":
    unittest.main()