d5a95e9270
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
"""
|
|
Tests for database performance indexes and schema migrations.
|
|
"""
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, inspect
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from app.database import Base, _ensure_indexes
|
|
|
|
|
|
@pytest.fixture
|
|
def engine_with_tables():
|
|
"""Create an in-memory SQLite engine with all tables."""
|
|
engine = create_engine(
|
|
"sqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
# Import models to register them with Base
|
|
from app.models import ( # noqa: F401
|
|
ApplicationSettings,
|
|
FileProcessingStep,
|
|
FileRecord,
|
|
ProcessingLog,
|
|
)
|
|
|
|
Base.metadata.create_all(bind=engine)
|
|
return engine
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_ensure_indexes_creates_expected_indexes(engine_with_tables):
|
|
"""_ensure_indexes should create all performance indexes."""
|
|
engine = engine_with_tables
|
|
inspector = inspect(engine)
|
|
|
|
# Run the migration
|
|
_ensure_indexes(engine, inspector)
|
|
|
|
# Refresh inspector after DDL
|
|
inspector = inspect(engine)
|
|
|
|
expected = {
|
|
"files": {"ix_files_created_at", "ix_files_mime_type"},
|
|
"processing_logs": {"ix_processing_logs_file_id", "ix_processing_logs_timestamp"},
|
|
"file_processing_steps": {"ix_file_processing_steps_status"},
|
|
}
|
|
|
|
for table, idx_names in expected.items():
|
|
actual_indexes = {idx["name"] for idx in inspector.get_indexes(table)}
|
|
for name in idx_names:
|
|
assert name in actual_indexes, f"Index {name} missing from {table}; found {actual_indexes}"
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_ensure_indexes_idempotent(engine_with_tables):
|
|
"""_ensure_indexes should be safe to run multiple times."""
|
|
engine = engine_with_tables
|
|
inspector = inspect(engine)
|
|
|
|
_ensure_indexes(engine, inspector)
|
|
# Running again should not raise
|
|
inspector2 = inspect(engine)
|
|
_ensure_indexes(engine, inspector2)
|
|
|
|
|
|
@pytest.mark.unit
|
|
def test_model_indexes_declared():
|
|
"""Verify key indexes are declared in SQLAlchemy model metadata."""
|
|
from app.models import FileProcessingStep, FileRecord, ProcessingLog
|
|
|
|
# FileRecord.created_at should be indexed
|
|
assert FileRecord.__table__.c.created_at.index is True
|
|
# FileRecord.mime_type should be indexed
|
|
assert FileRecord.__table__.c.mime_type.index is True
|
|
# ProcessingLog.file_id should be indexed
|
|
assert ProcessingLog.__table__.c.file_id.index is True
|
|
# ProcessingLog.timestamp should be indexed
|
|
assert ProcessingLog.__table__.c.timestamp.index is True
|
|
# FileProcessingStep.status should be indexed
|
|
assert FileProcessingStep.__table__.c.status.index is True
|