"""Tests for MCP tools error handling.
Tests for error cases and edge conditions in MCP tools.
"""
import tempfile
from pathlib import Path
from unittest.mock import patch
import pytest
from mcp_task_aggregator.models import TodoSource, TodoStatus
from mcp_task_aggregator.storage import Database, TodoRepository
from mcp_task_aggregator.tools.server import list_tasks_impl, list_todos_impl, sync_tasks_impl
@pytest.fixture
def temp_db():
"""Create a temporary database for testing."""
with tempfile.TemporaryDirectory() as tmpdir:
db_path = Path(tmpdir) / "test.db"
db = Database(db_path)
yield db, db_path
db.close()
class TestErrorHandling:
"""Tests for error handling in MCP tools."""
def test_list_tasks_handles_database_error(self, temp_db):
"""Test list_tasks handles database errors gracefully."""
db, db_path = temp_db
with patch("mcp_task_aggregator.agents.sync_agent.get_settings") as mock_settings:
mock_settings.return_value.database_path = db_path
# Force an error by patching list method
with patch("mcp_task_aggregator.agents.sync_agent.SyncAgent.list_tasks") as mock_list:
mock_list.side_effect = Exception("Database error")
result = list_tasks_impl()
assert "error" in result
assert "Database error" in result["error"]
assert result["tasks"] == []
assert result["total"] == 0
def test_list_todos_handles_error(self, temp_db):
"""Test list_todos handles errors gracefully."""
db, db_path = temp_db
with patch("mcp_task_aggregator.agents.sync_agent.get_settings") as mock_settings:
mock_settings.return_value.database_path = db_path
with patch("mcp_task_aggregator.agents.sync_agent.SyncAgent.list_tasks") as mock_list:
mock_list.side_effect = Exception("Connection error")
result = list_todos_impl()
assert "error" in result
assert "Connection error" in result["error"]
def test_sync_tasks_handles_unexpected_error(self, temp_db):
"""Test sync_tasks handles unexpected errors."""
db, db_path = temp_db
with patch("mcp_task_aggregator.agents.sync_agent.get_settings") as mock_settings:
mock_settings.return_value.database_path = db_path
with patch("mcp_task_aggregator.agents.sync_agent.SyncAgent.sync_jira") as mock_sync:
mock_sync.side_effect = Exception("Unexpected error")
result = sync_tasks_impl(source="jira")
assert "error" in result
assert "Unexpected error" in result["error"]
assert result["total_synced"] == 0
class TestEdgeCases:
"""Tests for edge cases in MCP tools."""
def test_list_tasks_with_zero_limit(self, temp_db):
"""Test list_tasks with edge case limits."""
db, db_path = temp_db
repo = TodoRepository(db)
repo.create(content="Test task")
with patch("mcp_task_aggregator.agents.sync_agent.get_settings") as mock_settings:
mock_settings.return_value.database_path = db_path
# Query with limit=1
result = list_tasks_impl(limit=1)
assert len(result["tasks"]) == 1
assert result["limit"] == 1
def test_list_tasks_with_high_offset(self, temp_db):
"""Test list_tasks with offset beyond data."""
db, db_path = temp_db
repo = TodoRepository(db)
repo.create(content="Task 1")
repo.create(content="Task 2")
with patch("mcp_task_aggregator.agents.sync_agent.get_settings") as mock_settings:
mock_settings.return_value.database_path = db_path
# Query with offset beyond available data
result = list_tasks_impl(limit=10, offset=100)
assert result["tasks"] == []
assert result["offset"] == 100
def test_list_tasks_empty_database(self, temp_db):
"""Test list_tasks on empty database."""
db, db_path = temp_db
with patch("mcp_task_aggregator.agents.sync_agent.get_settings") as mock_settings:
mock_settings.return_value.database_path = db_path
result = list_tasks_impl()
assert result["tasks"] == []
assert result["total"] == 0
def test_list_tasks_filter_nonexistent_source(self, temp_db):
"""Test list_tasks filtering by nonexistent source."""
db, db_path = temp_db
repo = TodoRepository(db)
repo.create(content="Local task", source_system=TodoSource.LOCAL)
with patch("mcp_task_aggregator.agents.sync_agent.get_settings") as mock_settings:
mock_settings.return_value.database_path = db_path
# Filter by github source (no tasks)
result = list_tasks_impl(source_system="github")
assert result["tasks"] == []
def test_list_tasks_filter_by_multiple_criteria(self, temp_db):
"""Test list_tasks with multiple filters."""
db, db_path = temp_db
repo = TodoRepository(db)
repo.create(
content="Matching task",
status=TodoStatus.TODO,
priority=5,
source_system=TodoSource.LOCAL,
tags=["urgent"],
)
repo.create(
content="Non-matching task",
status=TodoStatus.DONE,
priority=1,
source_system=TodoSource.JIRA,
source_id="TEST-1",
)
with patch("mcp_task_aggregator.agents.sync_agent.get_settings") as mock_settings:
mock_settings.return_value.database_path = db_path
# Filter by multiple criteria
result = list_tasks_impl(
source_system="local",
status="todo",
priority=4,
tags=["urgent"],
)
assert len(result["tasks"]) == 1
assert result["tasks"][0]["content"] == "Matching task"
def test_list_todos_returns_only_local(self, temp_db):
"""Test list_todos filters out external sources."""
db, db_path = temp_db
repo = TodoRepository(db)
repo.create(content="Local 1", source_system=TodoSource.LOCAL)
repo.create(content="Local 2", source_system=TodoSource.LOCAL)
repo.create(content="Jira", source_system=TodoSource.JIRA, source_id="TEST-1")
repo.create(content="GitHub", source_system=TodoSource.GITHUB, source_id="123")
with patch("mcp_task_aggregator.agents.sync_agent.get_settings") as mock_settings:
mock_settings.return_value.database_path = db_path
result = list_todos_impl()
assert len(result["tasks"]) == 2
for task in result["tasks"]:
assert task["source_system"] == "local"
class TestSyncVariations:
"""Tests for sync variations."""
def test_sync_with_full_refresh_flag(self, temp_db):
"""Test sync with full_refresh flag."""
db, db_path = temp_db
with patch("mcp_task_aggregator.agents.sync_agent.get_settings") as mock_settings:
mock_settings.return_value.database_path = db_path
mock_settings.return_value.jira_configured = False
result = sync_tasks_impl(source="jira", full_refresh=True)
# Should fail because not configured, but respects full_refresh param
assert result["total_synced"] == 0
assert len(result["summaries"]) == 1
