"""Tests for per-PR reviewer leases (#407)."""

import sys
import unittest
from datetime import datetime, timedelta, 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",
    minutes_ago: int = 0,
    candidate_head: str = "a" * 40,
) -> dict:
    now = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago)
    body = leases.format_lease_body(
        repo="Scaled-Tech-Consulting/Gitea-Tools",
        pr_number=pr_number,
        issue_number=295,
        reviewer_identity="rev1",
        profile="prgs-reviewer",
        session_id=session_id,
        worktree="branches/review-pr382",
        phase=phase,
        candidate_head=candidate_head,
        target_branch="master",
        target_branch_sha="b" * 40,
        last_activity=now,
    )
    return {"id": 1, "body": body, "user": {"login": "rev1"}}


class TestReviewerLeaseAcquire(unittest.TestCase):
    def setUp(self):
        leases.clear_session_lease()

    def test_two_reviewers_cannot_lease_same_pr(self):
        comments = [_lease_comment(382, "session-a")]
        result = leases.assess_acquire_lease(
            comments,
            pr_number=382,
            reviewer_identity="rev2",
            profile="prgs-reviewer",
            session_id="session-b",
            repo="Scaled-Tech-Consulting/Gitea-Tools",
            issue_number=295,
            worktree="branches/review-pr382-b",
            candidate_head="c" * 40,
            target_branch="master",
            target_branch_sha="d" * 40,
        )
        self.assertFalse(result["acquire_allowed"])
        self.assertTrue(any("already has active" in r for r in result["reasons"]))

    def test_two_reviewers_can_lease_different_prs(self):
        comments = [_lease_comment(382, "session-a")]
        result = leases.assess_acquire_lease(
            comments,
            pr_number=383,
            reviewer_identity="rev2",
            profile="prgs-reviewer",
            session_id="session-b",
            repo="Scaled-Tech-Consulting/Gitea-Tools",
            issue_number=296,
            worktree="branches/review-pr383",
            candidate_head="c" * 40,
            target_branch="master",
            target_branch_sha="d" * 40,
        )
        self.assertTrue(result["acquire_allowed"])
        self.assertIsNotNone(result["lease_body"])


class TestReviewerLeaseFreshness(unittest.TestCase):
    def test_stale_warning_at_half_the_sliding_window(self):
        # #747 warns at half the 10-minute window, while the owner can still
        # heartbeat and keep the lease.
        lease = leases.parse_lease_comment(
            _lease_comment(382, "session-a", minutes_ago=6)["body"]
        )
        self.assertEqual(
            leases.classify_lease_freshness(lease),
            "stale_warning",
        )

    def test_expired_once_the_sliding_window_lapses(self):
        # Pre-#747 a 65-minute-idle lease was "reclaimable" and had to wait out
        # a second timer. It is now simply expired and immediately takeable.
        lease = leases.parse_lease_comment(
            _lease_comment(382, "session-a", minutes_ago=65)["body"]
        )
        self.assertEqual(
            leases.classify_lease_freshness(lease),
            "expired",
        )


class TestReviewerLeaseNewestWins(unittest.TestCase):
    """#577: terminal release must supersede older claimed markers."""

    def test_released_after_claimed_has_no_active_lease(self):
        claim = _lease_comment(516, "session-a", phase="claimed")
        claim["id"] = 1
        release = _lease_comment(516, "session-a", phase="released")
        release["id"] = 2
        active = leases.find_active_reviewer_lease(
            [claim, release], pr_number=516
        )
        self.assertIsNone(active)

    def test_new_claim_after_release_is_active(self):
        claim = _lease_comment(516, "session-a", phase="claimed")
        claim["id"] = 1
        release = _lease_comment(516, "session-a", phase="released")
        release["id"] = 2
        reclaim = _lease_comment(516, "session-b", phase="claimed")
        reclaim["id"] = 3
        active = leases.find_active_reviewer_lease(
            [claim, release, reclaim], pr_number=516
        )
        self.assertIsNotNone(active)
        self.assertEqual(active.get("session_id"), "session-b")
        self.assertEqual(active.get("phase"), "claimed")
        self.assertEqual(active.get("comment_id"), 3)

    def test_done_after_claimed_has_no_active_lease(self):
        claim = _lease_comment(516, "session-a", phase="claimed")
        claim["id"] = 10
        done = _lease_comment(516, "session-a", phase="done")
        done["id"] = 11
        active = leases.find_active_reviewer_lease(
            [claim, done], pr_number=516
        )
        self.assertIsNone(active)


class TestReviewerLeaseMutationGate(unittest.TestCase):
    def setUp(self):
        leases.clear_session_lease()

    def test_reviewer_without_lease_cannot_mutate(self):
        head = "f" * 40
        comments = [_lease_comment(382, "other-session", candidate_head=head)]
        result = leases.assess_mutation_lease_gate(
            pr_number=382,
            comments=comments,
            reviewer_identity="rev1",
            session_id="my-session",
            mutation="approve",
            live_head_sha=head,
            pinned_head_sha=head,
        )
        self.assertTrue(result["block"])

    def test_owned_lease_allows_mutation(self):
        head = "f" * 40
        comments = [_lease_comment(382, "my-session", candidate_head=head)]
        leases.record_session_lease({
            "pr_number": 382,
            "session_id": "my-session",
            "candidate_head": head,
            "target_branch": "master",
            "comment_id": 101,
        }, lease_provenance=mla.build_lease_provenance(
            source=mla.SOURCE_ACQUIRE,
            comment_id=101,
        ))
        result = leases.assess_mutation_lease_gate(
            pr_number=382,
            comments=comments,
            reviewer_identity="rev1",
            session_id="my-session",
            mutation="approve",
            live_head_sha=head,
            pinned_head_sha=head,
        )
        self.assertFalse(result["block"])

    def test_head_change_invalidates_lease(self):
        reviewed = "f" * 40
        live = "e" * 40
        comments = [_lease_comment(382, "my-session", candidate_head=reviewed)]
        leases.record_session_lease({
            "pr_number": 382,
            "session_id": "my-session",
            "candidate_head": reviewed,
            "comment_id": 102,
        }, lease_provenance=mla.build_lease_provenance(
            source=mla.SOURCE_ACQUIRE,
            comment_id=102,
        ))
        result = leases.assess_mutation_lease_gate(
            pr_number=382,
            comments=comments,
            reviewer_identity="rev1",
            session_id="my-session",
            mutation="merge",
            live_head_sha=live,
            pinned_head_sha=reviewed,
        )
        self.assertTrue(result["block"])
        self.assertTrue(any("head" in r.lower() for r in result["reasons"]))


class TestReviewerLeaseMcpGate(unittest.TestCase):
    def setUp(self):
        leases.clear_session_lease()
        patch("mcp_server.verify_preflight_purity").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", "review_pr")
        mcp_server.record_preflight_check("whoami")
        mcp_server.record_preflight_check("capability", "reviewer")

    def tearDown(self):
        patch.stopall()
        leases.clear_session_lease()

    def test_reviewer_pr_lease_gate_helper_blocks_without_session(self):
        import mcp_server
        head = "a" * 40
        with patch("mcp_server._fetch_pr_comments", return_value=[]):
            reasons = mcp_server._reviewer_pr_lease_gate(
                pr_number=382,
                remote="prgs",
                host=None,
                org=None,
                repo=None,
                mutation="approve",
                live_head_sha=head,
                pinned_head_sha=head,
            )
        self.assertTrue(any("lease" in r.lower() for r in reasons))



class TestReviewerLeaseHandoffDiagnose(unittest.TestCase):
    """#599: canonical next-action diagnosis for open-PR lease handoff."""

    def setUp(self):
        leases.clear_session_lease()

    def test_no_lease_next_action_acquire(self):
        result = leases.diagnose_reviewer_pr_lease_handoff(
            [],
            pr_number=592,
            current_session_id="me-1",
            current_reviewer_identity="sysadmin",
            proposed_worktree="branches/review-pr-592",
        )
        self.assertEqual(result["classification"], "no_lease")
        self.assertEqual(result["next_action"], leases.NEXT_ACTION_ACQUIRE)
        self.assertIsNone(result["active_lease"])

    def test_foreign_active_wait_pr592_style(self):
        # Instructed lease gone; newer foreign claim is active.
        old = _lease_comment(592, "23139-da018dab43ba", phase="released")
        old["id"] = 8640
        active = _lease_comment(592, "51515-2bfb70f0685f", phase="claimed")
        active["id"] = 8647
        result = leases.diagnose_reviewer_pr_lease_handoff(
            [old, active],
            pr_number=592,
            current_session_id="fresh-session",
            current_reviewer_identity="sysadmin",
            proposed_worktree="branches/review-pr-592-issue-590",
            instructed_session_id="23139-da018dab43ba",
            instructed_comment_id=8640,
        )
        self.assertEqual(
            result["classification"],
            "instructed_lease_missing_with_replacement",
        )
        self.assertEqual(result["next_action"], leases.NEXT_ACTION_WAIT)
        self.assertTrue(result["instructed_lease_missing_with_replacement"])
        self.assertEqual(result["active_lease"]["session_id"], "51515-2bfb70f0685f")
        self.assertEqual(result["active_lease"]["comment_id"], 8647)
        self.assertFalse(result["mutation_allowed"])

    def test_foreign_expired_release_expired(self):
        # Pre-#747 this classified as "foreign_reclaimable" after the 60-minute
        # activity band. Under the sliding TTL the lease is simply expired, and
        # the sanctioned next action is unchanged.
        reclaim = _lease_comment(
            592, "foreign-old", phase="claimed", minutes_ago=65
        )
        reclaim["id"] = 99
        result = leases.diagnose_reviewer_pr_lease_handoff(
            [reclaim],
            pr_number=592,
            current_session_id="me-1",
            current_reviewer_identity="sysadmin",
            proposed_worktree="branches/review-pr-592",
        )
        self.assertEqual(result["classification"], "foreign_expired")
        self.assertEqual(
            result["next_action"], leases.NEXT_ACTION_RELEASE_EXPIRED_LEASE
        )

    def test_own_active_resume(self):
        head = "a" * 40
        claim = _lease_comment(592, "me-1", phase="claimed", candidate_head=head)
        claim["id"] = 10
        leases.record_session_lease({
            "pr_number": 592,
            "session_id": "me-1",
            "candidate_head": head,
            "comment_id": 10,
        }, lease_provenance=mla.build_lease_provenance(
            source=mla.SOURCE_ACQUIRE,
            comment_id=10,
        ))
        result = leases.diagnose_reviewer_pr_lease_handoff(
            [claim],
            pr_number=592,
            current_session_id="me-1",
            current_reviewer_identity="rev1",
            proposed_worktree="branches/review-pr382",
        )
        self.assertEqual(result["classification"], "own_active")
        self.assertEqual(
            result["next_action"], leases.NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
        )
        self.assertTrue(result["mutation_allowed"])

    def test_worktree_binding_mismatch(self):
        claim = _lease_comment(592, "me-1", phase="claimed")
        claim["id"] = 11
        # _lease_comment uses branches/review-pr382
        result = leases.diagnose_reviewer_pr_lease_handoff(
            [claim],
            pr_number=592,
            current_session_id="me-1",
            current_reviewer_identity="rev1",
            proposed_worktree="branches/review-pr-592-issue-590",
            env_bound_worktree="branches/review-pr-538",
        )
        self.assertEqual(result["classification"], "worktree_binding_mismatch")
        self.assertEqual(
            result["next_action"], leases.NEXT_ACTION_REPAIR_WORKTREE_BINDING
        )
        self.assertFalse(result["worktree_binding"]["match"])
        self.assertFalse(result["mutation_allowed"])



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