"""Tests for MCP tools.
Task 1.5.1: 2-8 focused tests for 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()
@pytest.fixture
def seeded_db(temp_db):
"""Create a database with seeded tasks."""
db, db_path = temp_db
repo = TodoRepository(db)
# Create local tasks
repo.create(content="Local task 1", status=TodoStatus.TODO, priority=3)
repo.create(content="Local task 2", status=TodoStatus.IN_PROGRESS, priority=5)
repo.create(content="Local task 3", status=TodoStatus.DONE, priority=1)
# Create Jira tasks
repo.create(
content="Jira task 1",
status=TodoStatus.TODO,
priority=4,
source_system=TodoSource.JIRA,
source_id="AGENTOPS-100",
tags=["jira", "backend"],
)
repo.create(
content="Jira task 2",
status=TodoStatus.IN_PROGRESS,
priority=2,
source_system=TodoSource.JIRA,
source_id="AGENTOPS-101",
tags=["jira", "frontend"],
)
return db, db_path
class TestListTasks:
"""Tests for list_tasks tool."""
def test_list_all_tasks(self, seeded_db):
"""Test listing all tasks without filters."""
db, db_path = seeded_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 "tasks" in result
assert len(result["tasks"]) == 5
assert result["total"] == 5
def test_list_tasks_by_source(self, seeded_db):
"""Test filtering tasks by source system."""
db, db_path = seeded_db
with patch("mcp_task_aggregator.agents.sync_agent.get_settings") as mock_settings:
mock_settings.return_value.database_path = db_path
# Filter by Jira source
result = list_tasks_impl(source_system="jira")
assert len(result["tasks"]) == 2
for task in result["tasks"]:
assert task["source_system"] == "jira"
def test_list_tasks_by_status(self, seeded_db):
"""Test filtering tasks by status."""
db, db_path = seeded_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(status="in_progress")
assert len(result["tasks"]) == 2
for task in result["tasks"]:
assert task["status"] == "in_progress"
def test_list_tasks_pagination(self, seeded_db):
"""Test pagination with limit and offset."""
db, db_path = seeded_db
with patch("mcp_task_aggregator.agents.sync_agent.get_settings") as mock_settings:
mock_settings.return_value.database_path = db_path
# First page
result1 = list_tasks_impl(limit=2, offset=0)
# Second page
result2 = list_tasks_impl(limit=2, offset=2)
assert len(result1["tasks"]) == 2
assert len(result2["tasks"]) == 2
assert result1["limit"] == 2
assert result1["offset"] == 0
assert result2["offset"] == 2
class TestListTodos:
"""Tests for list_todos tool."""
def test_list_todos_only_local(self, seeded_db):
"""Test that list_todos only returns local tasks."""
db, db_path = seeded_db
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"]) == 3
for task in result["tasks"]:
assert task["source_system"] == "local"
class TestSyncTasks:
"""Tests for sync_tasks tool."""
def test_sync_jira_not_configured(self, temp_db):
"""Test sync fails gracefully when Jira not configured."""
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
mock_settings.return_value.jira_url = ""
mock_settings.return_value.jira_email = ""
mock_settings.return_value.jira_api_token = ""
result = sync_tasks_impl(source="jira")
assert result["total_synced"] == 0
assert len(result["summaries"]) == 1
assert result["summaries"][0]["status"] == "failed"
def test_sync_unknown_source(self, temp_db):
"""Test sync with unknown source returns error."""
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 = sync_tasks_impl(source="unknown")
assert "error" in result
assert "unknown" in result["error"].lower()
def test_sync_all_no_sources_configured(self, temp_db):
"""Test sync all when no sources are configured."""
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
mock_settings.return_value.github_configured = False
mock_settings.return_value.linear_configured = False
mock_settings.return_value.markdown_configured = False
mock_settings.return_value.stm_configured = False
result = sync_tasks_impl(source="all")
# Should return empty results when nothing is configured
assert result["total_synced"] == 0
assert len(result["summaries"]) == 0
class TestTaskResponseFormat:
"""Tests for task response format."""
def test_task_response_has_required_fields(self, seeded_db):
"""Test that task responses include all required fields."""
db, db_path = seeded_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(limit=1)
assert len(result["tasks"]) == 1
task = result["tasks"][0]
# Check required fields
assert "id" in task
assert "content" in task
assert "status" in task
assert "priority" in task
assert "source_system" in task
assert "source_id" in task
assert "source_url" in task
assert "due_date" in task
assert "tags" in task
assert "created_at" in task
assert "updated_at" in task
