"""Tests for database layer.
Task 1.3.1: 2-8 focused tests for database operations.
"""
import tempfile
from pathlib import Path
import pytest
from mcp_task_aggregator.models import TodoSource, TodoStatus
from mcp_task_aggregator.storage import Database, SyncLogRepository, TodoRepository
@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.close()
@pytest.fixture
def todo_repo(temp_db):
"""Create a TodoRepository with temp database."""
return TodoRepository(temp_db)
@pytest.fixture
def sync_log_repo(temp_db):
"""Create a SyncLogRepository with temp database."""
return SyncLogRepository(temp_db)
class TestDatabase:
"""Tests for Database class."""
def test_database_initialization(self, temp_db):
"""Test database initializes with correct schema."""
# Check that tables exist
tables = temp_db.fetchall("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")
table_names = [row["name"] for row in tables]
assert "todos" in table_names
assert "tags" in table_names
assert "todo_tags" in table_names
assert "sync_log" in table_names
assert "agent_metadata" in table_names
assert "schema_version" in table_names
def test_schema_version_recorded(self, temp_db):
"""Test schema version is recorded."""
row = temp_db.fetchone("SELECT version FROM schema_version")
assert row is not None
assert row["version"] >= 1
class TestTodoRepository:
"""Tests for TodoRepository CRUD operations."""
def test_create_task(self, todo_repo):
"""Test creating a basic task."""
task = todo_repo.create(
content="Test task",
priority=3,
)
assert task.id is not None
assert task.content == "Test task"
assert task.priority == 3
assert task.status == TodoStatus.TODO
assert task.source_system == TodoSource.LOCAL
def test_create_task_with_tags(self, todo_repo):
"""Test creating a task with tags."""
task = todo_repo.create(
content="Tagged task",
tags=["important", "backend"],
)
assert task.id is not None
assert len(task.tags) == 2
tag_names = [t.name for t in task.tags]
assert "important" in tag_names
assert "backend" in tag_names
def test_get_by_id(self, todo_repo):
"""Test retrieving task by ID."""
created = todo_repo.create(content="Find me")
found = todo_repo.get_by_id(created.id)
assert found is not None
assert found.id == created.id
assert found.content == "Find me"
def test_get_by_source_id(self, todo_repo):
"""Test retrieving task by source system and ID."""
task = todo_repo.create(
content="Jira task",
source_system=TodoSource.JIRA,
source_id="AGENTOPS-123",
)
found = todo_repo.get_by_source_id(TodoSource.JIRA, "AGENTOPS-123")
assert found is not None
assert found.id == task.id
assert found.source_id == "AGENTOPS-123"
def test_update_task(self, todo_repo):
"""Test updating a task."""
task = todo_repo.create(content="Original content")
updated = todo_repo.update(
task.id,
content="Updated content",
status=TodoStatus.IN_PROGRESS,
priority=5,
)
assert updated is not None
assert updated.content == "Updated content"
assert updated.status == TodoStatus.IN_PROGRESS
assert updated.priority == 5
def test_upsert_creates_new(self, todo_repo):
"""Test upsert creates new task when not exists."""
task, was_created = todo_repo.upsert_by_source_id(
source_system=TodoSource.JIRA,
source_id="NEW-123",
content="New task from Jira",
priority=3,
tags=["jira"],
)
assert was_created is True
assert task.id is not None
assert task.content == "New task from Jira"
assert task.source_system == TodoSource.JIRA
assert task.source_id == "NEW-123"
def test_upsert_updates_existing(self, todo_repo):
"""Test upsert updates existing task."""
# First create
task1, created1 = todo_repo.upsert_by_source_id(
source_system=TodoSource.JIRA,
source_id="EXIST-456",
content="Original content",
sync_hash="hash1",
)
assert created1 is True
# Then upsert with different content
task2, created2 = todo_repo.upsert_by_source_id(
source_system=TodoSource.JIRA,
source_id="EXIST-456",
content="Updated content",
sync_hash="hash2",
)
assert created2 is False
assert task2.id == task1.id # Same task
assert task2.content == "Updated content"
def test_list_with_filters(self, todo_repo):
"""Test listing tasks with filters."""
# Create tasks with different statuses
todo_repo.create(content="Task 1", status=TodoStatus.TODO)
todo_repo.create(content="Task 2", status=TodoStatus.IN_PROGRESS)
todo_repo.create(content="Task 3", status=TodoStatus.DONE)
# Filter by status
in_progress = todo_repo.list(status=TodoStatus.IN_PROGRESS)
assert len(in_progress) == 1
assert in_progress[0].content == "Task 2"
def test_delete_task(self, todo_repo):
"""Test deleting a task."""
task = todo_repo.create(content="Delete me")
task_id = task.id
result = todo_repo.delete(task_id)
assert result is True
found = todo_repo.get_by_id(task_id)
assert found is None
class TestSyncLogRepository:
"""Tests for SyncLogRepository."""
def test_create_and_update_log(self, sync_log_repo):
"""Test creating and updating a sync log."""
log_id = sync_log_repo.create_log(
source_system="jira",
sync_type="full",
)
assert log_id is not None
# Update the log
sync_log_repo.update_log(
log_id,
status="completed",
tasks_synced=10,
tasks_created=5,
tasks_updated=5,
)
# Get latest
latest = sync_log_repo.get_latest_sync("jira")
assert latest is not None
assert latest["tasks_synced"] == 10
assert latest["tasks_created"] == 5
assert latest["status"] == "completed"
def test_list_logs(self, sync_log_repo):
"""Test listing sync logs."""
# Create multiple logs
for i in range(3):
log_id = sync_log_repo.create_log("jira", "full")
sync_log_repo.update_log(log_id, status="completed", tasks_synced=i + 1)
logs = sync_log_repo.list_logs(source_system="jira", limit=10)
assert len(logs) == 3
# Check that we have all expected tasks_synced values (order may vary due to same timestamp)
synced_counts = {log["tasks_synced"] for log in logs}
assert synced_counts == {1, 2, 3}
