"""Tests for MCP tools."""
import json
from unittest.mock import MagicMock
import pytest
from pr_review_mcp.gh_api import GitHubAPI, GitHubAPIError
from pr_review_mcp.tools import (
handle_list_review_threads,
handle_reply_and_resolve,
handle_reply_to_review_thread,
handle_resolve_review_thread,
)
class TestToolHandlers:
"""Test tool handler functions."""
@pytest.mark.asyncio
async def test_handle_list_review_threads(self):
"""Test list_review_threads handler."""
# Mock API
mock_api = MagicMock(spec=GitHubAPI)
mock_api.list_review_threads.return_value = [
{
"id": "PRRT_1",
"isResolved": False,
"path": "src/test.py",
"line": 42,
"startLine": None,
"diffSide": "RIGHT",
"comments": {
"nodes": [
{
"id": "PRRC_1",
"author": {"login": "reviewer"},
"body": "This needs improvement",
"createdAt": "2025-12-10T12:00:00Z",
}
]
},
}
]
arguments = {
"owner": "test-owner",
"repo": "test-repo",
"pull_number": 123,
"unresolved_only": True,
}
result = await handle_list_review_threads(mock_api, arguments)
assert len(result) == 1
assert result[0].type == "text"
# Parse JSON response
data = json.loads(result[0].text)
assert data["pull_request"] == "test-owner/test-repo#123"
assert data["thread_count"] == 1
assert data["threads"][0]["id"] == "PRRT_1"
assert data["threads"][0]["file"] == "src/test.py"
assert data["threads"][0]["first_comment"]["author"] == "reviewer"
@pytest.mark.asyncio
async def test_handle_reply_to_review_thread(self):
"""Test reply_to_review_thread handler."""
# Mock API
mock_api = MagicMock(spec=GitHubAPI)
mock_api.get_pr_id.return_value = "PR_123"
mock_api.add_thread_reply.return_value = {
"id": "PRRC_new",
"body": "Fixed!",
"createdAt": "2025-12-10T13:00:00Z",
"author": {"login": "contributor"},
}
arguments = {
"owner": "test-owner",
"repo": "test-repo",
"pull_number": 123,
"thread_id": "PRRT_1",
"body": "Fixed!",
}
result = await handle_reply_to_review_thread(mock_api, arguments)
assert len(result) == 1
assert result[0].type == "text"
# Parse JSON response
data = json.loads(result[0].text)
assert data["success"] is True
assert data["comment"]["id"] == "PRRC_new"
assert data["comment"]["body"] == "Fixed!"
assert data["comment"]["author"] == "contributor"
# Verify API calls
mock_api.get_pr_id.assert_called_once_with("test-owner", "test-repo", 123)
mock_api.add_thread_reply.assert_called_once_with("PR_123", "PRRT_1", "Fixed!")
@pytest.mark.asyncio
async def test_handle_resolve_review_thread(self):
"""Test resolve_review_thread handler."""
# Mock API
mock_api = MagicMock(spec=GitHubAPI)
mock_api.resolve_thread.return_value = {"id": "PRRT_1", "isResolved": True}
arguments = {"thread_id": "PRRT_1"}
result = await handle_resolve_review_thread(mock_api, arguments)
assert len(result) == 1
assert result[0].type == "text"
# Parse JSON response
data = json.loads(result[0].text)
assert data["success"] is True
assert data["thread"]["id"] == "PRRT_1"
assert data["thread"]["is_resolved"] is True
# Verify API call
mock_api.resolve_thread.assert_called_once_with("PRRT_1")
@pytest.mark.asyncio
async def test_handle_reply_and_resolve(self):
"""Test reply_and_resolve handler."""
# Mock API
mock_api = MagicMock(spec=GitHubAPI)
mock_api.get_pr_id.return_value = "PR_123"
mock_api.add_thread_reply.return_value = {
"id": "PRRC_new",
"body": "Done!",
"createdAt": "2025-12-10T14:00:00Z",
"author": {"login": "contributor"},
}
mock_api.resolve_thread.return_value = {"id": "PRRT_1", "isResolved": True}
arguments = {
"owner": "test-owner",
"repo": "test-repo",
"pull_number": 123,
"thread_id": "PRRT_1",
"body": "Done!",
}
result = await handle_reply_and_resolve(mock_api, arguments)
assert len(result) == 1
assert result[0].type == "text"
# Parse JSON response
data = json.loads(result[0].text)
assert data["success"] is True
assert data["comment"]["id"] == "PRRC_new"
assert data["comment"]["body"] == "Done!"
assert data["thread"]["id"] == "PRRT_1"
assert data["thread"]["is_resolved"] is True
# Verify API calls
mock_api.get_pr_id.assert_called_once_with("test-owner", "test-repo", 123)
mock_api.add_thread_reply.assert_called_once_with("PR_123", "PRRT_1", "Done!")
mock_api.resolve_thread.assert_called_once_with("PRRT_1")
@pytest.mark.asyncio
async def test_handle_list_review_threads_empty(self):
"""Test list_review_threads with no threads."""
# Mock API
mock_api = MagicMock(spec=GitHubAPI)
mock_api.list_review_threads.return_value = []
arguments = {
"owner": "test-owner",
"repo": "test-repo",
"pull_number": 123,
"unresolved_only": True,
}
result = await handle_list_review_threads(mock_api, arguments)
assert len(result) == 1
data = json.loads(result[0].text)
assert data["thread_count"] == 0
assert data["threads"] == []
@pytest.mark.asyncio
async def test_handle_with_api_error(self):
"""Test handler with GitHub API error."""
# Mock API that raises an error
mock_api = MagicMock(spec=GitHubAPI)
mock_api.list_review_threads.side_effect = GitHubAPIError("API request failed")
arguments = {
"owner": "test-owner",
"repo": "test-repo",
"pull_number": 123,
"unresolved_only": True,
}
# The handler should catch the error and return an error message
# Note: This test assumes error handling is done at a higher level
# If error handling is in the handler, we need to test that
with pytest.raises(GitHubAPIError):
await handle_list_review_threads(mock_api, arguments)
