"""Tests for the session hard-stop after a terminal review mutation (#332/#693).

Extends the single-terminal-decision machinery (#211): after a live
REQUEST_CHANGES on a PR head the same-head path must stop; after an APPROVE
only the merge sequence for that same PR may continue; #693 allows a
*different* PR to be mark_ready/review in the same durable lock (cross-PR
isolation); duplicate REQUEST_CHANGES at an unchanged head is suppressed.
"""
import os
import unittest
from unittest.mock import patch

import mcp_server

HEAD_SHA = "a" * 40


def _lock(mutations=None, correction=False, correction_pr=None):
    mutations = list(mutations or [])
    if correction and correction_pr is None and mutations:
        correction_pr = mutations[-1].get("pr_number")
    return {
        "task": "review_pr",
        "remote": "prgs",
        "session_pid": os.getpid(),
        "session_profile": "prgs-reviewer",
        "session_profile_lock": "",
        "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": mutations,
        "correction_authorized": correction,
        "correction_reason": "test" if correction else None,
        "correction_pr_number": correction_pr if correction else None,
        "correction_head_sha": None,
    }


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



APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1,
              "review_state": "approve"}
RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2,
        "review_state": "request_changes"}


class TestTerminalHardStopReasons(unittest.TestCase):
    def tearDown(self):
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()

    def test_no_lock_no_reasons(self):
        mcp_server._save_review_decision_lock(None)
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])

    def test_no_terminal_mutation_no_reasons(self):
        _seed()
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])

    def test_request_changes_blocks_same_pr_allows_other_pr_review(self):
        _seed([RC_A])
        # Same PR: still hard-stopped for mark_ready/review/merge.
        for op in ("merge", "mark_ready", "review"):
            reasons = mcp_server.terminal_review_hard_stop_reasons(5, op)
            self.assertTrue(reasons, f"{op}/5")
            self.assertIn("request_changes on PR #5", reasons[0])
            self.assertIn("#332", reasons[0])
        # #693: different PR may mark_ready / review (not merge via hard-stop alone).
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready"), [])
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(6, "review"), [])
        # Merge of an unapproved other PR remains blocked by hard-stop path
        # (last terminal is not approve of PR 6).
        self.assertTrue(
            mcp_server.terminal_review_hard_stop_reasons(6, "merge"))

    def test_approve_allows_same_pr_merge_and_other_pr_review(self):
        _seed([APPROVED_A])
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
        self.assertTrue(
            mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
        # #693 cross-PR isolation: other PR formal review path is open.
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready"), [])
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(6, "review"), [])

    def test_correction_reopens_review_path_only(self):
        _seed([RC_A], correction=True)
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(5, "mark_ready"), [])
        self.assertEqual(
            mcp_server.terminal_review_hard_stop_reasons(5, "review"), [])
        # scoped correction never opens a cross-PR merge
        self.assertTrue(
            mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
        # unscoped/cross-PR correction must not lift PR 6 via correction alone
        # (PR 6 is allowed by isolation, not by correction scope)


class TestMergeHardStopWiring(unittest.TestCase):
    def tearDown(self):
        import review_workflow_load
        review_workflow_load.clear_review_workflow_load()
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()

    def test_merge_blocked_after_request_changes(self):
        _seed([RC_A])
        result = mcp_server.gitea_merge_pr(
            pr_number=6, confirmation="MERGE PR 6", remote="prgs")
        self.assertFalse(result["performed"])
        self.assertTrue(
            any("#332" in r for r in result["reasons"]), result["reasons"])

    def test_merge_of_other_pr_blocked_after_approve(self):
        _seed([APPROVED_A])
        result = mcp_server.gitea_merge_pr(
            pr_number=6, confirmation="MERGE PR 6", remote="prgs")
        self.assertFalse(result["performed"])
        self.assertTrue(
            any("#332" in r for r in result["reasons"]), result["reasons"])


class TestMarkFinalHardStopWiring(unittest.TestCase):
    def tearDown(self):
        import review_workflow_load
        review_workflow_load.clear_review_workflow_load()
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()

    def test_mark_ready_same_pr_blocked_after_terminal_mutation(self):
        _seed([RC_A])
        result = mcp_server.gitea_mark_final_review_decision(
            pr_number=5, action="approve", remote="prgs",
            expected_head_sha=HEAD_SHA)
        self.assertFalse(result["marked_ready"])
        self.assertTrue(
            any("#332" in r for r in result["reasons"]), result["reasons"])

    def test_mark_ready_other_pr_allowed_after_foreign_terminal(self):
        """#693: foreign open-PR terminal must not block mark_final on PR B."""
        _seed([RC_A])
        no_lease = {"block": False, "reasons": [], "mutation_allowed": True}
        with patch("mcp_server._list_pr_lease_comments", return_value=[]), \
                patch("mcp_server._pr_work_lease_reviewer_block",
                      return_value=no_lease), \
                patch.object(mcp_server, "gitea_check_pr_eligibility",
                             return_value={"head_sha": HEAD_SHA, "eligible": True}):
            result = mcp_server.gitea_mark_final_review_decision(
                pr_number=6, action="approve", remote="prgs",
                expected_head_sha=HEAD_SHA)
        self.assertTrue(result["marked_ready"], result.get("reasons"))


def _feedback(blocking, stale=False, success=True):
    return {
        "success": success,
        "has_blocking_change_requests": blocking,
        "review_feedback_stale": stale,
        "current_head_sha": HEAD_SHA,
    }


def _mark(action, pr_number=6, **kwargs):
    kwargs.setdefault("expected_head_sha", HEAD_SHA)
    no_lease_block = {"block": False, "reasons": [], "mutation_allowed": True}
    with patch("mcp_server._list_pr_lease_comments", return_value=[]), \
            patch("mcp_server._pr_work_lease_reviewer_block", return_value=no_lease_block):
        return mcp_server.gitea_mark_final_review_decision(
            pr_number=pr_number, action=action, remote="prgs", **kwargs
        )


class TestDuplicateRequestChangesSuppression(unittest.TestCase):
    def tearDown(self):
        import review_workflow_load
        review_workflow_load.clear_review_workflow_load()
        mcp_server._save_review_decision_lock(None)
        mcp_server.review_workflow_load.clear_review_workflow_load()

    def test_duplicate_request_changes_blocked_at_same_head(self):
        _seed()
        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                          return_value=_feedback(blocking=True, stale=False)):
            result = _mark("request_changes")
        self.assertFalse(result["marked_ready"])
        self.assertTrue(
            any("duplicate" in r for r in result["reasons"]),
            result["reasons"])

    def test_request_changes_allowed_when_prior_blocker_stale(self):
        _seed()
        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                          return_value=_feedback(blocking=True, stale=True)):
            result = _mark("request_changes")
        self.assertTrue(result["marked_ready"], result.get("reasons"))

    def test_request_changes_allowed_when_no_blocker(self):
        _seed()
        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                          return_value=_feedback(blocking=False)):
            result = _mark("request_changes")
        self.assertTrue(result["marked_ready"], result.get("reasons"))

    def test_request_changes_fails_closed_when_feedback_unavailable(self):
        _seed()
        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                          return_value=_feedback(blocking=False,
                                                 success=False)):
            result = _mark("request_changes")
        self.assertFalse(result["marked_ready"])
        self.assertTrue(
            any("could not verify" in r for r in result["reasons"]),
            result["reasons"])

    def test_approve_does_not_fetch_feedback(self):
        _seed()
        with patch.object(mcp_server, "gitea_get_pr_review_feedback",
                          side_effect=AssertionError("must not be called")):
            result = _mark("approve")
        self.assertTrue(result["marked_ready"], result.get("reasons"))


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