diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index c054d7b..d52263b 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -4289,6 +4289,161 @@ class TestIssue546Deadlock(unittest.TestCase): # verify comment phase is released self.assertIn("phase: released", posted_comments[0]["body"]) + def test_reviewer_pr_lease_gate_resolves_identity_with_host(self): + """Lease mutation gate must resolve identity using host, not remote alias.""" + import mcp_server + + resolved_hosts = [] + + def tracking_username(host): + resolved_hosts.append(host) + return None if host in {"prgs", "dadeschools"} else "reviewer-bot" + + with _install_owned_reviewer_lease( + 550, + session_id=_DEFAULT_LEASE_SESSION, + head_sha="abc123", + ), patch( + "mcp_server._resolve", + return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"), + ), patch( + "mcp_server._authenticated_username", + side_effect=tracking_username, + ): + reasons = mcp_server._reviewer_pr_lease_gate( + pr_number=550, + remote="prgs", + host=None, + org=None, + repo=None, + mutation="approve", + live_head_sha="abc123", + pinned_head_sha="abc123", + ) + + self.assertEqual(reasons, []) + self.assertIn("gitea.prgs.cc", resolved_hosts) + self.assertNotIn("prgs", resolved_hosts) + + def test_acquire_reviewer_pr_lease_resolves_identity_with_host(self): + """Acquire path must not pass a remote alias into identity lookup.""" + import mcp_server + + resolved_hosts = [] + posted = [] + + def tracking_username(host): + resolved_hosts.append(host) + return None if host in {"prgs", "dadeschools"} else "reviewer-bot" + + def mock_api(method, url, auth, data=None): + if "/pulls/" in url: + return { + "user": {"login": "author-user"}, + "state": "open", + "head": {"sha": "abc123"}, + "mergeable": True, + } + if "/comments" in url: + if method == "POST": + posted.append(data["body"]) + return {"id": 1003} + return [] + return {} + + with patch("mcp_server.api_request", side_effect=mock_api), patch( + "mcp_server._authenticated_username", + side_effect=tracking_username, + ), patch( + "mcp_server._resolve", + return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"), + ): + res = mcp_server.gitea_acquire_reviewer_pr_lease( + pr_number=550, + worktree="/workspace", + candidate_head="abc123", + remote="prgs", + ) + + self.assertTrue(res.get("success"), res) + self.assertTrue(res.get("acquired"), res) + self.assertEqual(res.get("comment_id"), 1003) + self.assertTrue(posted) + self.assertIn("gitea.prgs.cc", resolved_hosts) + self.assertNotIn("prgs", resolved_hosts) + + def test_adopt_merger_pr_lease_resolves_identity_with_host(self): + """Adopt path must not pass a remote alias into identity lookup.""" + import mcp_server + import reviewer_pr_lease + + reviewer_pr_lease.clear_session_lease() + head = "a" * 40 + resolved_hosts = [] + posted = [] + reviewer_comment = _reviewer_lease_comment( + 550, + session_id="reviewer-session", + head_sha=head, + reviewer="reviewer-bot", + ) + + def tracking_username(host): + resolved_hosts.append(host) + return None if host in {"prgs", "dadeschools"} else "merger-bot" + + def mock_api(method, url, auth, data=None): + if "/pulls/" in url and "/comments" not in url: + return { + "number": 550, + "user": {"login": "author-user"}, + "state": "open", + "head": {"sha": head}, + "mergeable": True, + "merged": False, + } + if "/comments" in url: + if method == "POST": + posted.append(data["body"]) + return {"id": 1004} + return [reviewer_comment] + return {} + + merger_profile = { + "profile_name": "prgs-merger", + "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"], + "forbidden_operations": [], + } + with patch("mcp_server.get_profile", return_value=merger_profile), patch( + "mcp_server.api_request", + side_effect=mock_api, + ), patch( + "mcp_server._authenticated_username", + side_effect=tracking_username, + ), patch( + "mcp_server._resolve", + return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"), + ), patch( + "mcp_server.gitea_get_pr_review_feedback", + return_value={ + "approval_at_current_head": True, + "latest_approved_head_sha": head, + }, + ): + res = mcp_server.gitea_adopt_merger_pr_lease( + pr_number=550, + worktree="/workspace", + expected_head_sha=head, + remote="prgs", + ) + + self.assertTrue(res.get("success"), res) + self.assertTrue(res.get("adopted"), res) + self.assertEqual(res.get("adoption_comment_id"), 1004) + self.assertTrue(posted) + self.assertIn("gitea.prgs.cc", resolved_hosts) + self.assertNotIn("prgs", resolved_hosts) + def test_release_reviewer_pr_lease_same_identity_cross_session(self): """Same reviewer identity may release a lease claimed by another session.