from __future__ import annotations


import sys as _sys
from pathlib import Path as _Path
_sys.path.insert(0, str(_Path(__file__).resolve().parent))
from mutation_profile_fixture import shared_mutation_env  # noqa: E402

import os
import unittest
from unittest.mock import patch

import mcp_server
import stale_review_decision_lock as srdl
from mcp_server import (
    gitea_cleanup_stale_review_decision_lock,
    gitea_mark_final_review_decision,
)

HEAD = "a" * 40
FAKE_AUTH = "Basic dGVzdDp0ZXN0"

REVIEWER_ENV = {
    "GITEA_PROFILE_NAME": "prgs-reviewer",
    "GITEA_ALLOWED_OPERATIONS": (
        "gitea.read,gitea.pr.review,gitea.pr.approve,"
        "gitea.pr.request_changes,gitea.pr.comment"
    ),
    "GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer",
}

REVIEWER_PROFILE = {
    "profile_name": "prgs-reviewer",
    "allowed_operations": [
        "gitea.read",
        "gitea.pr.review",
        "gitea.pr.approve",
        "gitea.pr.request_changes",
        "gitea.pr.comment",
    ],
    "forbidden_operations": [
        "gitea.pr.merge",
        "gitea.pr.create",
        "gitea.branch.push",
    ],
}


def _reviewer_profile_patch():
    return patch.object(mcp_server, "get_profile", return_value=REVIEWER_PROFILE)


def _lock(mutations=None, correction=False):
    return {
        "task": "review_pr",
        "remote": "prgs",
        "org": "Scaled-Tech-Consulting",
        "repo": "Gitea-Tools",
        "session_pid": os.getpid(),
        "session_profile": "prgs-reviewer",
        "session_profile_lock": "prgs-reviewer",
        "profile_identity": "prgs-reviewer",
        "final_review_decision_ready": False,
        "ready_pr_number": None,
        "ready_action": None,
        "ready_expected_head_sha": None,
        "ready_remote": None,
        "ready_org": None,
        "ready_repo": None,
        "live_mutations": list(mutations or []),
        "correction_authorized": correction,
        "correction_reason": None,
    }


APPROVED_586 = {
    "pr_number": 586,
    "action": "approve",
    "review_id": 395,
    "review_state": "approve",
}
APPROVED_OPEN = {
    "pr_number": 700,
    "action": "approve",
    "review_id": 1,
    "review_state": "approve",
}
RC_OPEN = {
    "pr_number": 701,
    "action": "request_changes",
    "review_id": 2,
    "review_state": "request_changes",
}


def _seed(mutations=None, correction=False):
    import review_workflow_load

    review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
    mcp_server._save_review_decision_lock(_lock(mutations, correction))
    mcp_server.gitea_load_review_workflow()


def _merged_pr(pr_number=586, sha="6913ac9" + "0" * 33):
    return {
        "number": pr_number,
        "state": "closed",
        "merged": True,
        "merged_at": "2026-07-09T18:20:03Z",
        "merge_commit_sha": sha,
    }


def _open_pr(pr_number=700):
    return {
        "number": pr_number,
        "state": "open",
        "merged": False,
        "merged_at": None,
        "merge_commit_sha": None,
    }


# --------------------------------------------------------------------------- #
# Pure assessment
# --------------------------------------------------------------------------- #
class TestAssessStaleLock(unittest.TestCase):
    def test_no_lock(self):
        a = srdl.assess_stale_review_decision_lock(None)
        self.assertFalse(a["cleanup_allowed"])
        self.assertFalse(a["is_moot"])

    def test_no_terminal_mutations(self):
        a = srdl.assess_stale_review_decision_lock(_lock([]))
        self.assertFalse(a["cleanup_allowed"])

    def test_open_pr_not_cleanable(self):
        a = srdl.assess_stale_review_decision_lock(
            _lock([APPROVED_OPEN]), pr_live=_open_pr(700)
        )
        self.assertFalse(a["is_moot"])
        self.assertFalse(a["cleanup_allowed"])
        self.assertTrue(any("open PR" in r for r in a["reasons"]))

    def test_merged_pr_is_moot_and_cleanable(self):
        a = srdl.assess_stale_review_decision_lock(
            _lock([APPROVED_586]), pr_live=_merged_pr(586)
        )
        self.assertTrue(a["is_moot"])
        self.assertTrue(a["cleanup_allowed"])
        self.assertEqual(a["last_terminal_pr"], 586)

    def test_closed_unmerged_is_moot(self):
        a = srdl.assess_stale_review_decision_lock(
            _lock([RC_OPEN]),
            pr_live={"number": 701, "state": "closed", "merged": False},
        )
        self.assertTrue(a["is_moot"])
        self.assertTrue(a["cleanup_allowed"])

    def test_profile_mismatch_blocks(self):
        a = srdl.assess_stale_review_decision_lock(
            _lock([APPROVED_586]),
            pr_live=_merged_pr(586),
            active_profile_identity="other-profile",
        )
        self.assertFalse(a["cleanup_allowed"])
        self.assertFalse(a["profile_match"])

    def test_lookup_error_fail_closed(self):
        a = srdl.assess_stale_review_decision_lock(
            _lock([APPROVED_586]),
            pr_lookup_error="timeout",
        )
        self.assertFalse(a["cleanup_allowed"])
        self.assertTrue(any("timeout" in r for r in a["reasons"]))


# --------------------------------------------------------------------------- #
# Hard-stop still blocks active locks
# --------------------------------------------------------------------------- #
class TestActiveHardStopPreserved(unittest.TestCase):
    def tearDown(self):
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()

    def test_open_approve_blocks_same_pr_not_other_pr_mark_ready(self):
        _seed([APPROVED_OPEN])  # approve on PR #700
        # Same PR still hard-stopped for mark_ready.
        reasons_same = mcp_server.terminal_review_hard_stop_reasons(
            700, "mark_ready"
        )
        self.assertTrue(reasons_same)
        self.assertIn("#332", reasons_same[0])
        # #693: other PR may mark_ready without cleanup.
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready"),
            [],
        )

    def test_request_changes_still_blocks_same_pr(self):
        _seed([RC_OPEN])  # request_changes on PR #701
        for op in ("merge", "mark_ready", "review"):
            self.assertTrue(
                mcp_server.terminal_review_hard_stop_reasons(701, op),
                msg=op,
            )
        # Foreign PR review path is open (#693).
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready"),
            [],
        )


# --------------------------------------------------------------------------- #
# MCP cleanup tool
# --------------------------------------------------------------------------- #
class TestCleanupTool(unittest.TestCase):
    def setUp(self):
        self._env = patch.dict(os.environ, REVIEWER_ENV, clear=False)
        self._env.start()

    def tearDown(self):
        self._env.stop()
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()

    def test_read_only_reports_moot_without_clearing(self):
        _seed([APPROVED_586])
        with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \
                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
                patch.object(
                    mcp_server, "_authenticated_username", return_value="sysadmin"
                ), \
                _reviewer_profile_patch(), \
                patch.object(mcp_server, "_profile_operation_gate", return_value=[]):
            r = gitea_cleanup_stale_review_decision_lock(
                apply=False, remote="prgs",
                org="Scaled-Tech-Consulting", repo="Gitea-Tools",
            )
        self.assertTrue(r["success"])
        self.assertTrue(r["is_moot"])
        self.assertTrue(r["cleanup_allowed"])
        self.assertFalse(r["cleanup_performed"])
        self.assertIsNotNone(mcp_server._load_review_decision_lock())

    def test_apply_clears_moot_lock(self):
        _seed([APPROVED_586])
        posts = []

        def _api(method, url, auth, payload=None):
            if method == "GET" and "/pulls/586" in url:
                return _merged_pr()
            if method == "POST" and "/comments" in url:
                posts.append(payload)
                return {"id": 9999}
            return {}

        with patch.object(mcp_server, "api_request", side_effect=_api), \
                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
                patch.object(
                    mcp_server, "_authenticated_username", return_value="sysadmin"
                ), \
                _reviewer_profile_patch(), \
                patch.object(mcp_server, "_profile_operation_gate", return_value=[]), \
                patch.object(
                    mcp_server, "_audited",
                    side_effect=lambda *a, **k: __import__(
                        "contextlib"
                    ).nullcontext(),
                ):
            r = gitea_cleanup_stale_review_decision_lock(
                apply=True,
                expected_terminal_pr=586,
                remote="prgs",
                org="Scaled-Tech-Consulting",
                repo="Gitea-Tools",
            )
        self.assertTrue(r["cleanup_performed"], r)
        self.assertTrue(r["audit"]["applied"])
        self.assertIsNone(mcp_server._load_review_decision_lock())
        self.assertEqual(r.get("audit_comment_id"), 9999)
        self.assertTrue(posts)

    def test_apply_refuses_open_pr(self):
        _seed([APPROVED_OPEN])
        with patch.object(mcp_server, "api_request", return_value=_open_pr(700)), \
                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
                patch.object(
                    mcp_server, "_authenticated_username", return_value="sysadmin"
                ), \
                _reviewer_profile_patch(), \
                patch.object(mcp_server, "_profile_operation_gate", return_value=[]):
            r = gitea_cleanup_stale_review_decision_lock(
                apply=True, remote="prgs",
                org="Scaled-Tech-Consulting", repo="Gitea-Tools",
            )
        self.assertFalse(r["cleanup_performed"])
        self.assertFalse(r["cleanup_allowed"])
        self.assertIsNotNone(mcp_server._load_review_decision_lock())

    def test_expected_terminal_pr_pin(self):
        _seed([APPROVED_586])
        with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \
                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
                patch.object(
                    mcp_server, "_authenticated_username", return_value="sysadmin"
                ), \
                _reviewer_profile_patch(), \
                patch.object(mcp_server, "_profile_operation_gate", return_value=[]):
            r = gitea_cleanup_stale_review_decision_lock(
                apply=True,
                expected_terminal_pr=999,
                remote="prgs",
                org="Scaled-Tech-Consulting",
                repo="Gitea-Tools",
            )
        self.assertFalse(r["cleanup_performed"])
        self.assertTrue(any("expected_terminal_pr" in x for x in r["reasons"]))

    def test_after_moot_cleanup_other_pr_mark_ready_unblocked(self):
        """PR #592-style: after clearing merged #586 lock, new mark_ready works."""
        _seed([APPROVED_586])
        with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \
                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
                patch.object(
                    mcp_server, "_authenticated_username", return_value="sysadmin"
                ), \
                _reviewer_profile_patch(), \
                patch.object(mcp_server, "_profile_operation_gate", return_value=[]), \
                patch.object(
                    mcp_server, "_audited",
                    side_effect=lambda *a, **k: __import__(
                        "contextlib"
                    ).nullcontext(),
                ):
            cleaned = gitea_cleanup_stale_review_decision_lock(
                apply=True,
                expected_terminal_pr=586,
                remote="prgs",
                org="Scaled-Tech-Consulting",
                repo="Gitea-Tools",
                post_audit_comment=False,
            )
        self.assertTrue(cleaned["cleanup_performed"], cleaned)

        # Hard-stop gone; re-init clean lock like a fresh review_pr resolve.
        mcp_server.init_review_decision_lock(remote="prgs", task="review_pr")
        mcp_server.gitea_load_review_workflow()
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready"),
            [],
        )

        no_lease = {"block": False, "reasons": [], "mutation_allowed": True}
        with patch.object(mcp_server, "_list_pr_lease_comments", return_value=[]), \
                patch.object(
                    mcp_server, "_pr_work_lease_reviewer_block", return_value=no_lease
                ), \
                patch.object(
                    mcp_server,
                    "gitea_check_pr_eligibility",
                    return_value={
                        "eligible": True,
                        "reasons": [],
                        "authenticated_user": "sysadmin",
                        "pr_author": "jcwalker3",
                        "self_author": False,
                    },
                ), \
                patch.object(
                    mcp_server, "_authenticated_username", return_value="sysadmin"
                ):
            marked = gitea_mark_final_review_decision(
                pr_number=592,
                action="approve",
                expected_head_sha=HEAD,
                remote="prgs",
                org="Scaled-Tech-Consulting",
                repo="Gitea-Tools",
            )
        self.assertTrue(marked.get("marked_ready"), marked)


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