Lukas
30de86e77a
- Add src/server/database/init.py with complete initialization framework * Schema creation with idempotent table generation * Schema validation with detailed reporting * Schema versioning (v1.0.0) and migration support * Health checks with connectivity monitoring * Backup functionality for SQLite databases * Initial data seeding framework * Utility functions for database info and migration guides - Add comprehensive test suite (tests/unit/test_database_init.py) * 28 tests covering all functionality * 100% test pass rate * Integration tests and error handling - Update src/server/database/__init__.py * Export new initialization functions * Add schema version and expected tables constants - Fix syntax error in src/server/models/anime.py * Remove duplicate import statement - Update instructions.md * Mark database initialization task as complete Features: - Automatic schema creation and validation - Database health monitoring - Backup creation with timestamps - Production-ready with Alembic migration guidance - Async/await support throughout - Comprehensive error handling and logging Test Results: 69/69 database tests passing (100%)
496 lines
16 KiB
Python
496 lines
16 KiB
Python
"""Unit tests for database initialization module.
|
|
|
|
Tests cover:
|
|
- Database initialization
|
|
- Schema creation and validation
|
|
- Schema version management
|
|
- Initial data seeding
|
|
- Health checks
|
|
- Backup functionality
|
|
"""
|
|
import logging
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from src.server.database.base import Base
|
|
from src.server.database.init import (
|
|
CURRENT_SCHEMA_VERSION,
|
|
EXPECTED_TABLES,
|
|
check_database_health,
|
|
create_database_backup,
|
|
create_database_schema,
|
|
get_database_info,
|
|
get_migration_guide,
|
|
get_schema_version,
|
|
initialize_database,
|
|
seed_initial_data,
|
|
validate_database_schema,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_engine():
|
|
"""Create in-memory SQLite engine for testing."""
|
|
engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:",
|
|
echo=False,
|
|
poolclass=StaticPool,
|
|
)
|
|
yield engine
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
async def test_engine_with_tables(test_engine):
|
|
"""Create engine with tables already created."""
|
|
async with test_engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield test_engine
|
|
|
|
|
|
# =============================================================================
|
|
# Database Initialization Tests
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_database_success(test_engine):
|
|
"""Test successful database initialization."""
|
|
result = await initialize_database(
|
|
engine=test_engine,
|
|
create_schema=True,
|
|
validate_schema=True,
|
|
seed_data=False,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["schema_version"] == CURRENT_SCHEMA_VERSION
|
|
assert len(result["tables_created"]) == len(EXPECTED_TABLES)
|
|
assert result["validation_result"]["valid"] is True
|
|
assert result["health_check"]["healthy"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_database_without_schema_creation(test_engine_with_tables):
|
|
"""Test initialization without creating schema."""
|
|
result = await initialize_database(
|
|
engine=test_engine_with_tables,
|
|
create_schema=False,
|
|
validate_schema=True,
|
|
seed_data=False,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
assert result["schema_version"] == CURRENT_SCHEMA_VERSION
|
|
assert result["tables_created"] == []
|
|
assert result["validation_result"]["valid"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_database_with_seeding(test_engine):
|
|
"""Test initialization with data seeding."""
|
|
result = await initialize_database(
|
|
engine=test_engine,
|
|
create_schema=True,
|
|
validate_schema=True,
|
|
seed_data=True,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
# Seeding should complete without errors
|
|
# (even if no actual data is seeded for empty database)
|
|
|
|
|
|
# =============================================================================
|
|
# Schema Creation Tests
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_database_schema(test_engine):
|
|
"""Test creating database schema."""
|
|
tables = await create_database_schema(test_engine)
|
|
|
|
assert len(tables) == len(EXPECTED_TABLES)
|
|
assert set(tables) == EXPECTED_TABLES
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_database_schema_idempotent(test_engine_with_tables):
|
|
"""Test that creating schema is idempotent."""
|
|
# Tables already exist
|
|
tables = await create_database_schema(test_engine_with_tables)
|
|
|
|
# Should return existing tables, not create duplicates
|
|
assert len(tables) == len(EXPECTED_TABLES)
|
|
assert set(tables) == EXPECTED_TABLES
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_schema_uses_default_engine_when_none():
|
|
"""Test schema creation with None engine uses default."""
|
|
with patch("src.server.database.init.get_engine") as mock_get_engine:
|
|
# Create a real test engine
|
|
test_engine = create_async_engine(
|
|
"sqlite+aiosqlite:///:memory:",
|
|
echo=False,
|
|
poolclass=StaticPool,
|
|
)
|
|
mock_get_engine.return_value = test_engine
|
|
|
|
# This should call get_engine() and work with test engine
|
|
tables = await create_database_schema(engine=None)
|
|
assert len(tables) == len(EXPECTED_TABLES)
|
|
|
|
await test_engine.dispose()
|
|
|
|
|
|
# =============================================================================
|
|
# Schema Validation Tests
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_database_schema_valid(test_engine_with_tables):
|
|
"""Test validating a valid schema."""
|
|
result = await validate_database_schema(test_engine_with_tables)
|
|
|
|
assert result["valid"] is True
|
|
assert len(result["missing_tables"]) == 0
|
|
assert len(result["issues"]) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_database_schema_empty(test_engine):
|
|
"""Test validating an empty database."""
|
|
result = await validate_database_schema(test_engine)
|
|
|
|
assert result["valid"] is False
|
|
assert len(result["missing_tables"]) == len(EXPECTED_TABLES)
|
|
assert len(result["issues"]) > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_database_schema_partial(test_engine):
|
|
"""Test validating partially created schema."""
|
|
# Create only one table
|
|
async with test_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("""
|
|
CREATE TABLE anime_series (
|
|
id INTEGER PRIMARY KEY,
|
|
key VARCHAR(255) UNIQUE NOT NULL,
|
|
name VARCHAR(500) NOT NULL
|
|
)
|
|
""")
|
|
)
|
|
|
|
result = await validate_database_schema(test_engine)
|
|
|
|
assert result["valid"] is False
|
|
assert len(result["missing_tables"]) == len(EXPECTED_TABLES) - 1
|
|
assert "anime_series" not in result["missing_tables"]
|
|
|
|
|
|
# =============================================================================
|
|
# Schema Version Tests
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_schema_version_empty(test_engine):
|
|
"""Test getting schema version from empty database."""
|
|
version = await get_schema_version(test_engine)
|
|
assert version == "empty"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_schema_version_current(test_engine_with_tables):
|
|
"""Test getting schema version from current schema."""
|
|
version = await get_schema_version(test_engine_with_tables)
|
|
assert version == CURRENT_SCHEMA_VERSION
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_schema_version_unknown(test_engine):
|
|
"""Test getting schema version from unknown schema."""
|
|
# Create some random tables
|
|
async with test_engine.begin() as conn:
|
|
await conn.execute(
|
|
text("CREATE TABLE random_table (id INTEGER PRIMARY KEY)")
|
|
)
|
|
|
|
version = await get_schema_version(test_engine)
|
|
assert version == "unknown"
|
|
|
|
|
|
# =============================================================================
|
|
# Data Seeding Tests
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seed_initial_data_empty_database(test_engine_with_tables):
|
|
"""Test seeding data into empty database."""
|
|
# Should complete without errors
|
|
await seed_initial_data(test_engine_with_tables)
|
|
|
|
# Verify database is still empty (no sample data)
|
|
async with test_engine_with_tables.connect() as conn:
|
|
result = await conn.execute(text("SELECT COUNT(*) FROM anime_series"))
|
|
count = result.scalar()
|
|
assert count == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_seed_initial_data_existing_data(test_engine_with_tables):
|
|
"""Test seeding skips if data already exists."""
|
|
# Add some data
|
|
async with test_engine_with_tables.begin() as conn:
|
|
await conn.execute(
|
|
text("""
|
|
INSERT INTO anime_series (key, name, site, folder)
|
|
VALUES ('test-key', 'Test Anime', 'https://test.com', '/test')
|
|
""")
|
|
)
|
|
|
|
# Seeding should skip
|
|
await seed_initial_data(test_engine_with_tables)
|
|
|
|
# Verify only one record exists
|
|
async with test_engine_with_tables.connect() as conn:
|
|
result = await conn.execute(text("SELECT COUNT(*) FROM anime_series"))
|
|
count = result.scalar()
|
|
assert count == 1
|
|
|
|
|
|
# =============================================================================
|
|
# Health Check Tests
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_database_health_healthy(test_engine_with_tables):
|
|
"""Test health check on healthy database."""
|
|
result = await check_database_health(test_engine_with_tables)
|
|
|
|
assert result["healthy"] is True
|
|
assert result["accessible"] is True
|
|
assert result["tables"] == len(EXPECTED_TABLES)
|
|
assert result["connectivity_ms"] > 0
|
|
assert len(result["issues"]) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_database_health_empty(test_engine):
|
|
"""Test health check on empty database."""
|
|
result = await check_database_health(test_engine)
|
|
|
|
assert result["healthy"] is False
|
|
assert result["accessible"] is True
|
|
assert result["tables"] == 0
|
|
assert len(result["issues"]) > 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_database_health_connection_error():
|
|
"""Test health check with connection error."""
|
|
mock_engine = AsyncMock(spec=AsyncEngine)
|
|
mock_engine.connect.side_effect = Exception("Connection failed")
|
|
|
|
result = await check_database_health(mock_engine)
|
|
|
|
assert result["healthy"] is False
|
|
assert result["accessible"] is False
|
|
assert len(result["issues"]) > 0
|
|
assert "Connection failed" in result["issues"][0]
|
|
|
|
|
|
# =============================================================================
|
|
# Backup Tests
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_database_backup_not_sqlite():
|
|
"""Test backup fails for non-SQLite databases."""
|
|
with patch("src.server.database.init.settings") as mock_settings:
|
|
mock_settings.database_url = "postgresql://localhost/test"
|
|
|
|
with pytest.raises(NotImplementedError):
|
|
await create_database_backup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_database_backup_file_not_found():
|
|
"""Test backup fails if database file doesn't exist."""
|
|
with patch("src.server.database.init.settings") as mock_settings:
|
|
mock_settings.database_url = "sqlite:///nonexistent.db"
|
|
|
|
with pytest.raises(RuntimeError, match="Database file not found"):
|
|
await create_database_backup()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_database_backup_success(tmp_path):
|
|
"""Test successful database backup."""
|
|
# Create a temporary database file
|
|
db_file = tmp_path / "test.db"
|
|
db_file.write_text("test data")
|
|
|
|
backup_file = tmp_path / "backup.db"
|
|
|
|
with patch("src.server.database.init.settings") as mock_settings:
|
|
mock_settings.database_url = f"sqlite:///{db_file}"
|
|
|
|
result = await create_database_backup(backup_path=backup_file)
|
|
|
|
assert result == backup_file
|
|
assert backup_file.exists()
|
|
assert backup_file.read_text() == "test data"
|
|
|
|
|
|
# =============================================================================
|
|
# Utility Function Tests
|
|
# =============================================================================
|
|
|
|
|
|
def test_get_database_info():
|
|
"""Test getting database configuration info."""
|
|
info = get_database_info()
|
|
|
|
assert "database_url" in info
|
|
assert "database_type" in info
|
|
assert "schema_version" in info
|
|
assert "expected_tables" in info
|
|
assert info["schema_version"] == CURRENT_SCHEMA_VERSION
|
|
assert set(info["expected_tables"]) == EXPECTED_TABLES
|
|
|
|
|
|
def test_get_migration_guide():
|
|
"""Test getting migration guide."""
|
|
guide = get_migration_guide()
|
|
|
|
assert isinstance(guide, str)
|
|
assert "Alembic" in guide
|
|
assert "alembic init" in guide
|
|
assert "alembic upgrade head" in guide
|
|
|
|
|
|
# =============================================================================
|
|
# Integration Tests
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_initialization_workflow(test_engine):
|
|
"""Test complete initialization workflow."""
|
|
# 1. Initialize database
|
|
result = await initialize_database(
|
|
engine=test_engine,
|
|
create_schema=True,
|
|
validate_schema=True,
|
|
seed_data=True,
|
|
)
|
|
|
|
assert result["success"] is True
|
|
|
|
# 2. Verify schema
|
|
validation = await validate_database_schema(test_engine)
|
|
assert validation["valid"] is True
|
|
|
|
# 3. Check version
|
|
version = await get_schema_version(test_engine)
|
|
assert version == CURRENT_SCHEMA_VERSION
|
|
|
|
# 4. Health check
|
|
health = await check_database_health(test_engine)
|
|
assert health["healthy"] is True
|
|
assert health["accessible"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reinitialize_existing_database(test_engine_with_tables):
|
|
"""Test reinitializing an existing database."""
|
|
# Should be idempotent - safe to call multiple times
|
|
result1 = await initialize_database(
|
|
engine=test_engine_with_tables,
|
|
create_schema=True,
|
|
validate_schema=True,
|
|
)
|
|
|
|
result2 = await initialize_database(
|
|
engine=test_engine_with_tables,
|
|
create_schema=True,
|
|
validate_schema=True,
|
|
)
|
|
|
|
assert result1["success"] is True
|
|
assert result2["success"] is True
|
|
assert result1["schema_version"] == result2["schema_version"]
|
|
|
|
|
|
# =============================================================================
|
|
# Error Handling Tests
|
|
# =============================================================================
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_database_with_creation_error():
|
|
"""Test initialization handles schema creation errors."""
|
|
mock_engine = AsyncMock(spec=AsyncEngine)
|
|
mock_engine.begin.side_effect = Exception("Creation failed")
|
|
|
|
with pytest.raises(RuntimeError, match="Failed to initialize database"):
|
|
await initialize_database(
|
|
engine=mock_engine,
|
|
create_schema=True,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_schema_with_connection_error():
|
|
"""Test schema creation handles connection errors."""
|
|
mock_engine = AsyncMock(spec=AsyncEngine)
|
|
mock_engine.begin.side_effect = Exception("Connection failed")
|
|
|
|
with pytest.raises(RuntimeError, match="Schema creation failed"):
|
|
await create_database_schema(mock_engine)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_validate_schema_with_inspection_error():
|
|
"""Test validation handles inspection errors gracefully."""
|
|
mock_engine = AsyncMock(spec=AsyncEngine)
|
|
mock_engine.connect.side_effect = Exception("Inspection failed")
|
|
|
|
result = await validate_database_schema(mock_engine)
|
|
|
|
assert result["valid"] is False
|
|
assert len(result["issues"]) > 0
|
|
assert "Inspection failed" in result["issues"][0]
|
|
|
|
|
|
# =============================================================================
|
|
# Constants Tests
|
|
# =============================================================================
|
|
|
|
|
|
def test_schema_constants():
|
|
"""Test that schema constants are properly defined."""
|
|
assert CURRENT_SCHEMA_VERSION == "1.0.0"
|
|
assert len(EXPECTED_TABLES) == 4
|
|
assert "anime_series" in EXPECTED_TABLES
|
|
assert "episodes" in EXPECTED_TABLES
|
|
assert "download_queue" in EXPECTED_TABLES
|
|
assert "user_sessions" in EXPECTED_TABLES
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|