Gitea-Tools/tests/test_merger_lease_acquire.py

"""Merger lease acquisition tests (#718)."""

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 gitea_mcp_server as mcp_server
import reviewer_pr_lease as leases


class TestMergerLeaseAcquireTool(unittest.TestCase):
    def setUp(self):
        mcp_server._preflight_whoami_called = True
        mcp_server._preflight_capability_called = True
        mcp_server._preflight_resolved_role = "merger"
        leases.clear_session_lease()

    @patch("gitea_mcp_server.api_request")
    @patch("gitea_mcp_server._auth", return_value="token pass")
    @patch("gitea_mcp_server._resolve", return_value=("dadeschools", "org", "repo"))
    @patch("gitea_mcp_server.get_profile", return_value={"username": "merger-user", "profile_name": "prgs-merger", "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], "forbidden_operations": []})
    @patch("gitea_mcp_server._fetch_pr_comments", return_value=[])
    @patch("gitea_mcp_server._verify_role_mutation_workspace")
    def test_acquire_merger_lease_success(self, _verify, _comments, _profile, _resolve, _auth, mock_api):
        """Merger lease acquisition succeeds when no reviewer lease exists."""
        # api_request is called for PR state, then for POSTing comment
        mock_api.side_effect = [
            {"state": "open"},  # PR is open
            {"id": 456},        # Posted comment ID
        ]
        
        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
            res = mcp_server.gitea_acquire_merger_pr_lease(
                pr_number=718,
                worktree="branches/issue-718",
                session_id="merger-session",
            )
            
            self.assertTrue(res["success"])
            self.assertTrue(res["acquired"])
            self.assertEqual(res["comment_id"], 456)
            self.assertEqual(res["session_id"], "merger-session")
            
            # Verify lease body includes merger identity
            args, kwargs = mock_api.call_args_list[-1]
            self.assertEqual(args[0], "POST")
            body = args[3]["body"]
            self.assertIn("reviewer_identity: merger-user", body)
            self.assertIn("profile: prgs-merger", body)
            self.assertIn("phase: claimed", body)
            
            # Post-mutation readback: lease is recorded in session
            session_lease = leases._SESSION_LEASE
            self.assertIsNotNone(session_lease)
            self.assertEqual(session_lease["phase"], "claimed")
            self.assertEqual(session_lease["session_id"], "merger-session")
            
            # Verify lease provenance uses SOURCE_ACQUIRE_MERGER
            provenance = session_lease.get("lease_provenance") or {}
            self.assertEqual(provenance.get("source"), "gitea_acquire_merger_pr_lease")

    @patch("gitea_mcp_server.api_request")
    @patch("gitea_mcp_server._auth", return_value="token pass")
    @patch("gitea_mcp_server._resolve", return_value=("dadeschools", "org", "repo"))
    @patch("gitea_mcp_server.get_profile", return_value={"username": "merger-user", "profile_name": "prgs-merger", "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], "forbidden_operations": []})
    @patch("gitea_mcp_server._fetch_pr_comments")
    @patch("gitea_mcp_server._verify_role_mutation_workspace")
    def test_acquire_merger_lease_fails_if_active_exists(self, _verify, _comments, _profile, _resolve, _auth, mock_api):
        """Acquisition fails closed if another session holds an active lease."""
        active_body = leases.format_lease_body(
            repo="org/repo",
            pr_number=718,
            issue_number=None,
            reviewer_identity="other-user",
            profile="prgs-reviewer",
            session_id="other-session",
            worktree="branches/review-1",
            phase="claimed",
            candidate_head="a" * 40,
            target_branch="master",
            target_branch_sha="b" * 40,
            last_activity=datetime.now(timezone.utc),
        )
        _comments.return_value = [{"id": 1, "body": active_body, "user": {"login": "other-user"}}]
        
        # PR is open
        mock_api.return_value = {"state": "open"}
        
        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
            res = mcp_server.gitea_acquire_merger_pr_lease(
                pr_number=718,
                worktree="branches/issue-718",
                session_id="merger-session",
            )
            
            self.assertFalse(res["success"])
            self.assertFalse(res["acquired"])
            self.assertIn("already has active reviewer lease", " ".join(res["reasons"]))
            
            # No POST made
            post_calls = [c for c in mock_api.call_args_list if c[0][0] == "POST"]
            self.assertEqual(len(post_calls), 0)
            
            # No partial state recorded
            self.assertIsNone(leases._SESSION_LEASE)

    @patch("gitea_mcp_server.get_profile", return_value={"username": "merger-user", "profile_name": "prgs-merger", "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], "forbidden_operations": []})
    @patch("gitea_mcp_server._verify_role_mutation_workspace")
    def test_acquire_merger_lease_fails_workspace_binding(self, _verify, _profile):
        """Fails closed if workspace verification throws RuntimeError."""
        _verify.side_effect = RuntimeError("Branches-only mutation guard")
        
        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
            res = mcp_server.gitea_acquire_merger_pr_lease(
                pr_number=718,
                worktree="branches/issue-718",
                session_id="merger-session",
            )
            
            self.assertFalse(res["success"])
            self.assertFalse(res["acquired"])
            self.assertIn("Branches-only mutation guard", res["reasons"][0])
            self.assertIsNone(leases._SESSION_LEASE)

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