copilot-swe-agent[bot]
b79ba81e05
- Add database indexes on FileRecord.created_at, FileRecord.mime_type, ProcessingLog.file_id, ProcessingLog.timestamp, and FileProcessingStep.status for faster filtering, sorting, and joins. - Add _ensure_indexes() migration for existing databases. - Create app/utils/cache.py with fail-open Redis GET/SET/DELETE helpers. - Cache MIME types dropdown query in files view (120s TTL). - Optimize batch status query to load only needed columns. Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
"""
|
|
Tests for database performance indexes and schema migrations.
|
|
"""
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, inspect, text
|
|
from sqlalchemy.orm import sessionmaker
|
|
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
|