#!/usr/bin/env python3 """ Test that database schema (table names) never changes unexpectedly. Renaming or removing tables will cause data loss for existing users. These tests catch accidental schema changes before they're deployed. """ import pytest # Expected table names - DO NOT REMOVE ANY # Adding new tables is fine, but removing/renaming breaks existing databases EXPECTED_TABLES = { # Auth (in auth.db, not user dbs) "users", # Settings "api_keys", "settings", "user_settings", # Queue "queue_status", "task_metadata", # Research "research", "research_history", "research_resources", "research_strategies", "research_tasks", "search_queries", "search_results", "queued_researches", "user_active_researches", # Reports "reports", "report_sections", # Library "collections", "collection_folders", "collection_folder_files", "documents", "document_blobs", "document_chunks", "document_collections", "download_queue", "library_statistics", "rag_document_status", "rag_indices", "source_types", "upload_batches", # Download tracking "download_tracker", "download_duplicates", "download_attempts", # Metrics "token_usage", "model_usage", "research_ratings", "search_calls", # News "news_cards", "news_interests", "news_subscriptions", "news_user_preferences", "news_user_ratings", "subscription_folders", "user_news_search_history", # File integrity "file_integrity_records", "file_verification_failures", # Providers "provider_models", # Rate limiting "rate_limit_attempts", "rate_limit_estimates", # Domain classification "domain_classifications", # Logs "app_logs", "journals", # Papers (deduplicated academic papers) "papers", "paper_appearances", # Benchmark "benchmark_configs", "benchmark_progress", "benchmark_results", "benchmark_runs", # Chat "chat_sessions", "chat_messages", "chat_progress_steps", # Zotero "zotero_sync_state", "zotero_item_map", } class TestSchemaStability: """ Verify that database table names haven't changed. Renaming or removing tables will cause existing user databases to lose data or fail to open properly. """ def test_no_tables_removed(self): """ Ensure no expected tables have been removed. Removing a table definition will cause data loss when users upgrade, as SQLAlchemy won't know how to access that data. """ from local_deep_research.database.models import Base # Get all actual table names from the models actual_tables = set(Base.metadata.tables.keys()) # Check that all expected tables still exist missing_tables = EXPECTED_TABLES - actual_tables assert not missing_tables, ( f"CRITICAL: Database tables have been removed!\n" f"Missing tables: {missing_tables}\n\n" "Removing tables will cause data loss for existing users.\n" "If you intentionally removed these tables, you need a migration plan.\n" "Otherwise, REVERT THIS CHANGE." ) def test_no_tables_renamed(self): """ Detect if tables might have been renamed. If new tables appear and expected tables are missing, it's likely a rename which will cause data loss. """ from local_deep_research.database.models import Base actual_tables = set(Base.metadata.tables.keys()) missing_tables = EXPECTED_TABLES - actual_tables if missing_tables: # Check if there are new tables that might be renames new_tables = actual_tables - EXPECTED_TABLES if new_tables: pytest.fail( f"Possible table rename detected!\n" f"Missing: {missing_tables}\n" f"New: {new_tables}\n\n" "If you renamed tables, existing data will be lost.\n" "You need a migration to copy data from old to new tables." ) def test_new_tables_are_documented(self): """ Ensure any new tables are added to EXPECTED_TABLES. This is a reminder to update this test when adding new tables. New tables should be added to EXPECTED_TABLES to track them. """ from local_deep_research.database.models import Base actual_tables = set(Base.metadata.tables.keys()) new_tables = actual_tables - EXPECTED_TABLES # These are okay - just a reminder to update the test if new_tables: pytest.fail( f"New tables detected that aren't in EXPECTED_TABLES:\n" f"{new_tables}\n\n" "Please add these to EXPECTED_TABLES in this test file.\n" "This ensures they'll be protected from accidental removal." ) class TestCriticalColumns: """ Verify that critical columns in key tables haven't been removed. These are columns that store important user data. """ def test_user_settings_has_required_columns(self): """Verify UserSettings table has all required columns.""" from local_deep_research.database.models import UserSettings required_columns = {"id", "key", "value", "category"} actual_columns = set(UserSettings.__table__.columns.keys()) missing = required_columns - actual_columns assert not missing, ( f"UserSettings is missing required columns: {missing}\n" "This will break user settings storage." ) def test_research_has_required_columns(self): """Verify Research table has all required columns.""" from local_deep_research.database.models.research import Research required_columns = {"id", "query", "status", "mode", "created_at"} actual_columns = set(Research.__table__.columns.keys()) missing = required_columns - actual_columns assert not missing, ( f"Research is missing required columns: {missing}\n" "This will break research history." ) def test_api_keys_has_required_columns(self): """Verify APIKey table has all required columns.""" from local_deep_research.database.models import APIKey required_columns = {"id", "provider", "key", "is_active"} actual_columns = set(APIKey.__table__.columns.keys()) missing = required_columns - actual_columns assert not missing, ( f"APIKey is missing required columns: {missing}\n" "This will break API key storage." ) def test_journal_has_exact_column_set(self): """Journal is an LLM-only cache; the column set is deliberately minimal. Lock it down so an accidental add/drop in the model gets caught without a matching migration. The journal-quality redesign intentionally excluded the bibliometric columns (issn, h_index, impact_factor, ...) that would have served a Tier 2/3 cache; those values live in the read-only reference DB instead. Re-adding any of them to the model without a new migration would cause the schema drift this test catches. """ from local_deep_research.database.models.logs import Journal expected = { "id", "name", "name_lower", "quality", "score_source", "quality_model", "quality_analysis_time", } actual = set(Journal.__table__.columns.keys()) missing = expected - actual extra = actual - expected assert not missing, ( f"Journal model is missing required columns: {sorted(missing)}" ) assert not extra, ( f"Journal model has unexpected columns: {sorted(extra)}. " "If you intentionally added a column, also update this test " "and the journal-quality migration (and its downgrade)." ) def test_chat_tables_have_exact_column_sets(self): """Lock down the column sets for the three chat tables introduced in migration 0010. Drift detection — accidental rename, removal, or new column without a migration would surface as a clear diff. Mirrors the journals pattern above. Currently `phase` on ChatProgressStep has no behavior tests pinning it; this test catches a silent drop. """ from local_deep_research.database.models.chat import ( ChatMessage, ChatProgressStep, ChatSession, ) expected = { ChatSession: { "id", "title", "status", "accumulated_context", "created_at", "message_count", }, ChatMessage: { "id", "session_id", "research_id", "role", "message_type", "content", "sequence_number", "created_at", }, ChatProgressStep: { "id", "research_id", "session_id", "phase", "content", "sequence_number", "created_at", }, } for model, cols in expected.items(): actual = set(model.__table__.columns.keys()) missing = cols - actual extra = actual - cols assert not missing, ( f"{model.__name__} is missing required columns: " f"{sorted(missing)}" ) assert not extra, ( f"{model.__name__} has unexpected columns: {sorted(extra)}. " "If intentional, update this test and migration 0010 " "(or open a follow-up migration)." ) if __name__ == "__main__": pytest.main([__file__, "-v"])