"""Shared pytest fixtures and configuration.
Provides reusable fixtures for database setup, test data factories,
and mock configurations across all test modules.
"""
import tempfile
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock
import pytest
from mcp_task_aggregator.adapters import JiraConfig
from mcp_task_aggregator.config import Settings, reset_settings
from mcp_task_aggregator.models import (
JiraMetadata,
Tag,
Todo,
TodoSource,
TodoStatus,
)
from mcp_task_aggregator.storage import Database, SyncLogRepository, TagRepository, TodoRepository
# ============================================================================
# Database Fixtures
# ============================================================================
@pytest.fixture
def temp_db_path():
"""Create a temporary database file path."""
with tempfile.TemporaryDirectory() as tmpdir:
yield Path(tmpdir) / "test.db"
@pytest.fixture
def temp_db(temp_db_path):
"""Create a temporary database for testing."""
db = Database(temp_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 tag_repo(temp_db):
"""Create a TagRepository with temp database."""
return TagRepository(temp_db)
@pytest.fixture
def sync_log_repo(temp_db):
"""Create a SyncLogRepository with temp database."""
return SyncLogRepository(temp_db)
# ============================================================================
# Data Factories
# ============================================================================
@pytest.fixture
def todo_factory():
"""Factory for creating test Todo instances."""
def _create_todo(
content: str = "Test task",
status: TodoStatus = TodoStatus.TODO,
priority: int = 0,
source_system: TodoSource = TodoSource.LOCAL,
source_id: str | None = None,
tags: list[str] | None = None,
**kwargs: Any,
) -> Todo:
"""Create a test Todo with sensible defaults."""
return Todo(
content=content,
status=status,
priority=priority,
source_system=source_system,
source_id=source_id,
tags=[Tag(name=t) for t in (tags or [])],
**kwargs,
)
return _create_todo
@pytest.fixture
def jira_issue_factory():
"""Factory for creating test Jira issue data."""
def _create_jira_issue(
key: str = "TEST-1",
summary: str = "Test issue",
status: str = "To Do",
priority: str = "Medium",
**kwargs: Any,
) -> dict[str, Any]:
"""Create test Jira issue data with sensible defaults."""
defaults = {
"key": key,
"id": "10001",
"self": "https://test.atlassian.net/rest/api/3/issue/10001",
"summary": summary,
"description": "Test description",
"status": status,
"priority": priority,
"issuetype": "Task",
"created": "2024-01-15T10:00:00.000+0000",
"updated": "2024-01-15T10:00:00.000+0000",
"duedate": None,
"reporter": "Test User",
"reporter_email": "test@example.com",
"assignee": None,
"assignee_email": None,
"labels": [],
"components": [],
"project_key": "TEST",
"parent_key": None,
"sprint": None,
}
defaults.update(kwargs)
return defaults
return _create_jira_issue
@pytest.fixture
def jira_metadata_factory():
"""Factory for creating test JiraMetadata instances."""
def _create_metadata(
key: str = "TEST-1",
project_key: str = "TEST",
issue_type: str = "Task",
**kwargs: Any,
) -> JiraMetadata:
"""Create test JiraMetadata with sensible defaults."""
defaults = {
"key": key,
"project_key": project_key,
"issue_type": issue_type,
"reporter": None,
"assignee": None,
"labels": [],
"components": [],
"sprint": None,
"parent_key": None,
}
defaults.update(kwargs)
return JiraMetadata(**defaults)
return _create_metadata
# ============================================================================
# Configuration Fixtures
# ============================================================================
@pytest.fixture
def mock_settings():
"""Create a mock Settings instance for testing."""
with tempfile.TemporaryDirectory() as tmpdir:
settings = Settings(
database_path=Path(tmpdir) / "test.db",
jira_url="",
jira_email="",
jira_api_token="",
jira_project_key="TEST",
github_token="",
github_repos=[],
linear_api_key="",
log_level="DEBUG",
log_file=None,
sync_batch_size=100,
)
yield settings
reset_settings()
@pytest.fixture
def jira_config():
"""Create a test Jira configuration."""
return JiraConfig(
url="https://test.atlassian.net",
email="test@example.com",
api_token="test-token",
project_key="TEST",
)
@pytest.fixture
def mock_jira_settings(mock_settings):
"""Create mock settings with Jira configured."""
mock_settings.jira_url = "https://test.atlassian.net"
mock_settings.jira_email = "test@example.com"
mock_settings.jira_api_token = "test-token"
mock_settings.jira_project_key = "TEST"
return mock_settings
# ============================================================================
# Seeded Database Fixtures
# ============================================================================
@pytest.fixture
def seeded_db(temp_db, todo_repo):
"""Create a database with seeded test data."""
# Create local tasks
todo_repo.create(
content="Local task 1",
status=TodoStatus.TODO,
priority=3,
tags=["local", "urgent"],
)
todo_repo.create(
content="Local task 2",
status=TodoStatus.IN_PROGRESS,
priority=5,
tags=["local"],
)
todo_repo.create(
content="Local task 3",
status=TodoStatus.DONE,
priority=1,
)
# Create Jira tasks
todo_repo.create(
content="Jira task 1",
status=TodoStatus.TODO,
priority=4,
source_system=TodoSource.JIRA,
source_id="TEST-100",
source_url="https://test.atlassian.net/browse/TEST-100",
tags=["jira", "backend"],
)
todo_repo.create(
content="Jira task 2",
status=TodoStatus.IN_PROGRESS,
priority=2,
source_system=TodoSource.JIRA,
source_id="TEST-101",
source_url="https://test.atlassian.net/browse/TEST-101",
tags=["jira", "frontend"],
)
# Create blocked task
todo_repo.create(
content="Blocked task",
status=TodoStatus.BLOCKED,
priority=4,
source_system=TodoSource.LOCAL,
tags=["blocked", "needs-attention"],
)
return temp_db
@pytest.fixture
def seeded_sync_logs(sync_log_repo):
"""Create test sync logs."""
# Create a successful Jira sync
log_id = sync_log_repo.create_log("jira", "full")
sync_log_repo.update_log(
log_id,
status="completed",
tasks_synced=10,
tasks_created=5,
tasks_updated=5,
)
# Create a failed sync
log_id = sync_log_repo.create_log("github", "incremental")
sync_log_repo.update_log(
log_id,
status="failed",
tasks_synced=0,
tasks_created=0,
tasks_updated=0,
errors=[{"error": "Connection timeout"}],
)
return sync_log_repo
# ============================================================================
# Mock Fixtures
# ============================================================================
@pytest.fixture
def mock_jira_client():
"""Create a mock Jira client."""
mock_client = MagicMock()
mock_client.search_issues.return_value = []
return mock_client
@pytest.fixture
def sample_jira_issues(jira_issue_factory):
"""Create a set of sample Jira issues for testing."""
return [
jira_issue_factory(
key="TEST-1",
summary="First issue",
status="To Do",
priority="High",
labels=["backend", "urgent"],
),
jira_issue_factory(
key="TEST-2",
summary="Second issue",
status="In Progress",
priority="Medium",
assignee="John Doe",
assignee_email="john@example.com",
),
jira_issue_factory(
key="TEST-3",
summary="Third issue",
status="Done",
priority="Low",
),
]
# ============================================================================
# Utility Functions
# ============================================================================
@pytest.fixture
def assert_todo_equal():
"""Helper to assert two Todo objects are equal (ignoring timestamps)."""
def _assert_equal(todo1: Todo, todo2: Todo, ignore_fields: list[str] | None = None) -> None:
"""Assert two todos are equal, optionally ignoring specific fields."""
ignore = ignore_fields or []
ignore.extend(["id", "created_at", "updated_at", "uuid"])
for field in Todo.model_fields:
if field in ignore:
continue
assert getattr(todo1, field) == getattr(todo2, field), f"Field {field} differs"
return _assert_equal
# ============================================================================
# Cleanup
# ============================================================================
@pytest.fixture(autouse=True)
def reset_config_after_test():
"""Reset global settings after each test."""
yield
reset_settings()
