""" Comprehensive tests for Alembic migration functionality. Tests cover: - Fresh database migrations - Existing database handling (pre-Alembic databases) - Column migrations for existing tables - Idempotent migrations - Error scenarios - SQLCipher encrypted database migrations """ import os from unittest.mock import patch import pytest from sqlalchemy import create_engine, inspect, text from sqlalchemy.orm import sessionmaker from local_deep_research.database.alembic_runner import ( get_alembic_config, get_current_revision, get_head_revision, get_migrations_dir, needs_migration, run_migrations, stamp_database, ) from local_deep_research.database.initialize import initialize_database from local_deep_research.database.models import Base class TestAlembicRunner: """Tests for the alembic_runner module.""" @pytest.fixture def fresh_engine(self, tmp_path): """Create a fresh SQLite engine for testing.""" db_path = tmp_path / "fresh_test.db" engine = create_engine(f"sqlite:///{db_path}") yield engine engine.dispose() @pytest.fixture def existing_engine(self, tmp_path): """Create an engine with existing tables but no Alembic.""" db_path = tmp_path / "existing_test.db" engine = create_engine(f"sqlite:///{db_path}") # Create tables directly (simulating pre-Alembic database) Base.metadata.create_all(engine) yield engine engine.dispose() @pytest.fixture def partial_engine(self, tmp_path): """Create an engine with only some tables (partial initialization).""" db_path = tmp_path / "partial_test.db" engine = create_engine(f"sqlite:///{db_path}") # Create only a few tables from local_deep_research.database.models import Setting, QueueStatus Setting.__table__.create(engine, checkfirst=True) QueueStatus.__table__.create(engine, checkfirst=True) yield engine engine.dispose() def test_get_migrations_dir_exists(self): """Migrations directory should exist.""" migrations_dir = get_migrations_dir() assert migrations_dir.exists() assert (migrations_dir / "env.py").exists() assert (migrations_dir / "versions").exists() def test_get_head_revision_returns_value(self): """Should return the head revision.""" head = get_head_revision() assert head is not None assert isinstance(head, str) def test_get_alembic_config(self, fresh_engine): """Should create a valid Alembic config.""" config = get_alembic_config(fresh_engine) assert config is not None assert config.get_main_option("script_location") is not None assert "engine" not in config.attributes def test_fresh_database_has_no_revision(self, fresh_engine): """Fresh database should have no current revision.""" # Fresh database with no tables revision = get_current_revision(fresh_engine) assert revision is None def test_fresh_database_needs_migration(self, fresh_engine): """Fresh database should need migration.""" assert needs_migration(fresh_engine) def test_run_migrations_on_fresh_database(self, fresh_engine): """Migrations should create all tables on fresh database.""" run_migrations(fresh_engine) # raises on failure assert not needs_migration(fresh_engine) # Verify tables were created inspector = inspect(fresh_engine) tables = inspector.get_table_names() assert "alembic_version" in tables assert "settings" in tables assert "research" in tables assert "task_metadata" in tables def test_run_migrations_creates_all_expected_tables(self, fresh_engine): """Should create all core tables.""" run_migrations(fresh_engine) inspector = inspect(fresh_engine) tables = set(inspector.get_table_names()) # Core tables that should exist (not exhaustive) expected_tables = { "settings", "research", "research_history", "journals", "app_logs", "queued_researches", "token_usage", "research_ratings", "task_metadata", "queue_status", "alembic_version", } for table in expected_tables: assert table in tables, f"Expected table '{table}' not found" def test_existing_database_migration(self, existing_engine): """Existing database without Alembic should be migrated properly.""" # Verify no alembic_version yet inspector = inspect(existing_engine) tables = inspector.get_table_names() assert "alembic_version" not in tables assert len(tables) > 0 # Has other tables # Run migrations run_migrations(existing_engine) # raises on failure # Verify alembic_version now exists new_inspector = inspect(existing_engine) new_tables = new_inspector.get_table_names() assert "alembic_version" in new_tables # Verify current revision is set revision = get_current_revision(existing_engine) assert revision is not None def test_partial_database_migration(self, partial_engine): """Partial database should have missing tables created.""" # Verify only some tables exist inspector = inspect(partial_engine) initial_tables = set(inspector.get_table_names()) assert "settings" in initial_tables assert "research" not in initial_tables # Run migrations run_migrations(partial_engine) # raises on failure # Verify missing tables were created new_inspector = inspect(partial_engine) final_tables = set(new_inspector.get_table_names()) assert "settings" in final_tables # Still exists assert "research" in final_tables # Now created assert "task_metadata" in final_tables # Now created assert "alembic_version" in final_tables def test_stamp_database(self, fresh_engine): """Should stamp database without running migrations.""" # Create tables first Base.metadata.create_all(fresh_engine) # Stamp at head stamp_database(fresh_engine, "head") # Verify stamped revision = get_current_revision(fresh_engine) head = get_head_revision() assert revision == head def test_idempotent_migrations(self, fresh_engine): """Running migrations multiple times should be safe.""" # First run run_migrations(fresh_engine) # raises on failure inspector1 = inspect(fresh_engine) tables1 = set(inspector1.get_table_names()) revision1 = get_current_revision(fresh_engine) # Second run run_migrations(fresh_engine) # raises on failure inspector2 = inspect(fresh_engine) tables2 = set(inspector2.get_table_names()) revision2 = get_current_revision(fresh_engine) # Should be identical assert tables1 == tables2 assert revision1 == revision2 # Third run run_migrations(fresh_engine) # raises on failure def test_needs_migration_after_complete(self, fresh_engine): """After migrations, needs_migration should return False.""" run_migrations(fresh_engine) assert not needs_migration(fresh_engine) def test_run_migrations_skips_upgrade_when_at_head( self, fresh_engine, loguru_caplog ): """When the database is already at head, run_migrations must not call command.upgrade() — that would open a no-op write transaction and hold a RESERVED lock under isolation_level="IMMEDIATE", blocking concurrent readers for no benefit. """ # Migrate to head first run_migrations(fresh_engine) assert get_current_revision(fresh_engine) == get_head_revision() # Second call must short-circuit — command.upgrade() should not run, # and the short-circuit log line must be emitted (positive signal # that the guard fired, not that the function was gutted). # # Also pins that the orphan-cleanup and FK toggle introduced by # PR #4000 are NOT invoked on the short-circuit path. If the # guard is moved BELOW engine.connect() + _disable_fk_for_migration # in a future refactor, this test fails — the existing # command.upgrade mock alone would not catch that regression. with ( patch( "local_deep_research.database.alembic_runner.command.upgrade" ) as mock_upgrade, patch( "local_deep_research.database.alembic_runner." "_drop_orphan_alembic_temp_tables" ) as mock_drop_orphans, patch( "local_deep_research.database.alembic_runner." "_disable_fk_for_migration" ) as mock_disable_fk, ): with loguru_caplog.at_level("INFO"): run_migrations(fresh_engine) assert mock_upgrade.call_count == 0 assert mock_drop_orphans.call_count == 0 assert mock_disable_fk.call_count == 0 assert "skipping upgrade" in loguru_caplog.text def test_run_migrations_runs_upgrade_on_fresh_db(self, tmp_path): """Fresh DB (current revision None) must NOT short-circuit; command.upgrade() must run so tables and alembic_version get created. """ db_path = tmp_path / "fresh_for_upgrade.db" engine = create_engine(f"sqlite:///{db_path}") try: assert get_current_revision(engine) is None run_migrations(engine) # If the short-circuit had applied (it must not), revision # would still be None and alembic_version would not exist. assert get_current_revision(engine) == get_head_revision() assert "alembic_version" in inspect(engine).get_table_names() finally: engine.dispose() def test_imports_without_errors(self): """All migration modules import without side effects.""" env_path = get_migrations_dir() / "env.py" assert env_path.exists() class TestColumnMigrations: """Tests for column-level migrations.""" @pytest.fixture def old_schema_engine(self, tmp_path): """Create database with old schema (missing progress columns).""" db_path = tmp_path / "old_schema.db" engine = create_engine(f"sqlite:///{db_path}") # Create task_metadata table without progress columns # This simulates an old database before those columns were added with engine.begin() as conn: conn.execute( text( """ CREATE TABLE task_metadata ( task_id VARCHAR PRIMARY KEY, status VARCHAR NOT NULL, task_type VARCHAR NOT NULL, created_at DATETIME, started_at DATETIME, completed_at DATETIME, error_message VARCHAR, priority INTEGER DEFAULT 0, retry_count INTEGER DEFAULT 0, max_retries INTEGER DEFAULT 3 ) """ ) ) # Create a few more tables to simulate existing database conn.execute( text( """ CREATE TABLE settings ( id INTEGER PRIMARY KEY, key VARCHAR NOT NULL UNIQUE, value VARCHAR, type VARCHAR DEFAULT 'string' ) """ ) ) yield engine engine.dispose() def test_old_schema_missing_columns(self, old_schema_engine): """Verify old schema is missing expected columns.""" inspector = inspect(old_schema_engine) columns = { col["name"] for col in inspector.get_columns("task_metadata") } assert "task_id" in columns assert "status" in columns assert "progress_current" not in columns assert "progress_total" not in columns assert "progress_message" not in columns assert "metadata_json" not in columns def test_column_migration_adds_missing_columns(self, old_schema_engine): """Migration should add missing columns to existing tables.""" # Run migrations run_migrations(old_schema_engine) # raises on failure # Verify columns were added inspector = inspect(old_schema_engine) columns = { col["name"] for col in inspector.get_columns("task_metadata") } assert "progress_current" in columns assert "progress_total" in columns assert "progress_message" in columns assert "metadata_json" in columns def test_column_migration_preserves_data(self, old_schema_engine): """Migration should preserve existing data in tables.""" # Insert test data with old_schema_engine.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type) VALUES ('test-123', 'completed', 'research') """ ) ) # Run migrations run_migrations(old_schema_engine) # Verify data preserved with old_schema_engine.connect() as conn: result = conn.execute( text( "SELECT task_id, status FROM task_metadata WHERE task_id = 'test-123'" ) ).fetchone() assert result is not None assert result[0] == "test-123" assert result[1] == "completed" def test_column_migration_sets_defaults(self, old_schema_engine): """New columns should have correct default values.""" # Insert test data before migration with old_schema_engine.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type) VALUES ('test-456', 'processing', 'benchmark') """ ) ) # Run migrations run_migrations(old_schema_engine) # Check that new columns have defaults with old_schema_engine.connect() as conn: row = conn.execute( text( """ SELECT progress_current, progress_total, progress_message FROM task_metadata WHERE task_id = 'test-456' """ ) ).fetchone() # New columns should have default values or NULL # progress_current and progress_total default to 0 # progress_message is nullable assert row is not None assert row[0] == 0 # progress_current defaults to 0 assert row[1] == 0 # progress_total defaults to 0 class TestInitializeDatabaseIntegration: """Integration tests for initialize_database with Alembic.""" @pytest.fixture def temp_engine(self, tmp_path): """Create a temporary database engine.""" db_path = tmp_path / "integration_test.db" engine = create_engine(f"sqlite:///{db_path}") yield engine engine.dispose() def test_initialize_fresh_database(self, temp_engine): """initialize_database should work on fresh database.""" Session = sessionmaker(bind=temp_engine) with Session() as session: initialize_database(temp_engine, session) inspector = inspect(temp_engine) tables = inspector.get_table_names() assert "alembic_version" in tables assert "settings" in tables assert len(tables) > 20 # Many tables def test_initialize_existing_database(self, temp_engine): """initialize_database should work on existing database.""" # First initialization Session = sessionmaker(bind=temp_engine) with Session() as session: initialize_database(temp_engine, session) inspector1 = inspect(temp_engine) tables1 = set(inspector1.get_table_names()) # Second initialization (should be safe) with Session() as session: initialize_database(temp_engine, session) inspector2 = inspect(temp_engine) tables2 = set(inspector2.get_table_names()) assert tables1 == tables2 def test_initialize_preserves_data(self, temp_engine): """initialize_database should preserve existing data.""" Session = sessionmaker(bind=temp_engine) # First initialization with Session() as session: initialize_database(temp_engine, session) # Insert test data into queue_status (simpler schema) with temp_engine.begin() as conn: conn.execute( text( """ INSERT INTO queue_status (active_tasks, queued_tasks) VALUES (5, 10) """ ) ) # Second initialization with Session() as session: initialize_database(temp_engine, session) # Verify data preserved with temp_engine.connect() as conn: result = conn.execute( text("SELECT active_tasks, queued_tasks FROM queue_status") ).fetchone() assert result is not None assert result[0] == 5 assert result[1] == 10 class TestMigrationEdgeCases: """Tests for edge cases and error scenarios.""" @pytest.fixture def temp_engine(self, tmp_path): """Create a temporary database engine.""" db_path = tmp_path / "edge_case_test.db" engine = create_engine(f"sqlite:///{db_path}") yield engine engine.dispose() def test_migration_with_empty_versions_dir(self, temp_engine, tmp_path): """Should handle case when no migrations exist.""" # This tests the edge case where migrations dir is empty # The actual implementation should handle this gracefully with patch( "local_deep_research.database.alembic_runner.get_head_revision", return_value=None, ): run_migrations(temp_engine) # raises on failure def test_get_current_revision_on_corrupted_table(self, temp_engine): """Should handle corrupted alembic_version table.""" # Create a corrupted alembic_version table with temp_engine.begin() as conn: conn.execute( text("CREATE TABLE alembic_version (version_num VARCHAR)") ) # Don't insert any rows - table exists but is empty # Should not crash revision = get_current_revision(temp_engine) assert revision is None # Empty table means no revision def test_multiple_concurrent_migrations(self, tmp_path): """Multiple engines against same DB should work safely.""" db_path = tmp_path / "concurrent_test.db" engine1 = create_engine(f"sqlite:///{db_path}") engine2 = create_engine(f"sqlite:///{db_path}") try: # First migration run_migrations(engine1) # raises on failure # Second migration (should be idempotent) run_migrations(engine2) # raises on failure # Both should see same revision rev1 = get_current_revision(engine1) rev2 = get_current_revision(engine2) assert rev1 == rev2 finally: engine1.dispose() engine2.dispose() class TestSQLCipherMigrations: """Tests for SQLCipher encrypted database migrations.""" @pytest.fixture def sqlcipher_available(self): """Check if SQLCipher is available.""" import importlib.util if importlib.util.find_spec("sqlcipher3") is None: pytest.skip("SQLCipher not available") return True @pytest.fixture def encrypted_engine(self, tmp_path, sqlcipher_available): """Create an encrypted SQLCipher database.""" import sqlcipher3 db_path = tmp_path / "encrypted_test.db" password = "test_password_123" # Create encrypted connection def create_connection(): conn = sqlcipher3.connect(str(db_path), check_same_thread=False) cursor = conn.cursor() cursor.execute(f"PRAGMA key = '{password}'") cursor.execute("PRAGMA cipher_page_size = 4096") cursor.execute("PRAGMA kdf_iter = 256000") cursor.execute("PRAGMA cipher_hmac_algorithm = HMAC_SHA512") cursor.execute("PRAGMA cipher_kdf_algorithm = PBKDF2_HMAC_SHA512") cursor.close() return conn engine = create_engine("sqlite://", creator=create_connection) yield engine engine.dispose() def test_encrypted_database_migration(self, encrypted_engine): """Should be able to run migrations on encrypted database.""" run_migrations(encrypted_engine) # raises on failure # Verify tables created inspector = inspect(encrypted_engine) tables = inspector.get_table_names() assert "alembic_version" in tables assert "settings" in tables def test_encrypted_database_idempotent(self, encrypted_engine): """Multiple migrations on encrypted DB should be safe.""" run_migrations(encrypted_engine) # raises on failure run_migrations(encrypted_engine) # raises on failure rev1 = get_current_revision(encrypted_engine) head = get_head_revision() assert rev1 == head class TestAlembicVersionTable: """Tests for alembic_version table behavior.""" @pytest.fixture def migrated_engine(self, tmp_path): """Create a fully migrated database.""" db_path = tmp_path / "migrated_test.db" engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine) yield engine engine.dispose() def test_alembic_version_table_exists(self, migrated_engine): """Alembic version table should exist after migration.""" inspector = inspect(migrated_engine) assert "alembic_version" in inspector.get_table_names() def test_alembic_version_has_single_row(self, migrated_engine): """Alembic version table should have exactly one row.""" with migrated_engine.connect() as conn: result = conn.execute( text("SELECT COUNT(*) FROM alembic_version") ).fetchone() assert result[0] == 1 def test_alembic_version_matches_head(self, migrated_engine): """Alembic version should match head revision.""" with migrated_engine.connect() as conn: result = conn.execute( text("SELECT version_num FROM alembic_version") ).fetchone() assert result[0] == get_head_revision() def test_revision_updates_after_migration(self, tmp_path): """Revision should update when new migrations run.""" db_path = tmp_path / "revision_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Run migrations run_migrations(engine) revision = get_current_revision(engine) # Verify it's at head assert revision == get_head_revision() finally: engine.dispose() class TestMigrationChain: """Tests for migration chain and ordering.""" def test_migration_files_exist(self): """All expected migration files should exist.""" versions_dir = get_migrations_dir() / "versions" expected_files = [ "0001_initial_schema.py", "0002_add_task_progress_columns.py", "0003_add_research_indexes.py", "0004_migrate_legacy_app_settings.py", "0005_add_resource_document_id.py", "0006_journal_quality_system.py", "0007_backfill_missing_indexes.py", "0008_fix_research_strategy_fk.py", "0009_default_fetch_mode_summary.py", "0010_add_chat_tables.py", "0011_add_collection_is_public.py", "0012_add_collection_agent_enabled.py", "0013_remove_meta_search_engines.py", "0014_benchmark_run_version_and_snapshot.py", "0015_drop_document_notes.py", "0016_drop_orphaned_cache_tables.py", ] for filename in expected_files: filepath = versions_dir / filename assert filepath.exists(), f"Migration file {filename} not found" def test_migration_chain_is_valid(self): """Migration chain should be properly linked.""" from alembic.config import Config from alembic.script import ScriptDirectory migrations_dir = get_migrations_dir() config = Config() config.set_main_option("script_location", str(migrations_dir)) script = ScriptDirectory.from_config(config) # Get all revisions revisions = list(script.walk_revisions()) # Should have at least 3 revisions assert len(revisions) >= 3 # Head should be the latest migration assert script.get_current_head() == get_head_revision() # Verify chain link 0010 -> 0009 -> 0008 (the rest of the chain is checked below). rev_0010 = script.get_revision("0010") assert rev_0010.down_revision == "0009" rev_0009 = script.get_revision("0009") assert rev_0009.down_revision == "0008" rev_0008 = script.get_revision("0008") assert rev_0008.down_revision == "0007" rev_0007 = script.get_revision("0007") assert rev_0007.down_revision == "0006" rev_0006 = script.get_revision("0006") assert rev_0006.down_revision == "0005" rev_0005 = script.get_revision("0005") assert rev_0005.down_revision == "0004" rev_0004 = script.get_revision("0004") assert rev_0004.down_revision == "0003" rev_0003 = script.get_revision("0003") assert rev_0003.down_revision == "0002" rev_0002 = script.get_revision("0002") assert rev_0002.down_revision == "0001" rev_0001 = script.get_revision("0001") assert rev_0001.down_revision is None def test_run_to_specific_revision(self, tmp_path): """Should be able to run migrations to a specific revision.""" db_path = tmp_path / "specific_rev_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Run only to revision 0001 run_migrations(engine, target="0001") # raises on failure # Should be at 0001 revision = get_current_revision(engine) assert revision == "0001" # Should still need migration (not at head) assert needs_migration(engine) # Run to head run_migrations(engine, target="head") # raises on failure assert get_current_revision(engine) == get_head_revision() finally: engine.dispose() class TestUsersTableExclusion: """Tests to verify users table is correctly excluded from user databases.""" @pytest.fixture def user_db_engine(self, tmp_path): """Create a user database (not auth database).""" db_path = tmp_path / "user_db_test.db" engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine) yield engine engine.dispose() def test_users_table_not_in_user_database(self, user_db_engine): """Users table should NOT be created in user databases.""" inspector = inspect(user_db_engine) tables = inspector.get_table_names() # Users table should NOT exist in user databases # It only exists in the auth database assert "users" not in tables def test_user_specific_tables_exist(self, user_db_engine): """User-specific tables should exist.""" inspector = inspect(user_db_engine) tables = set(inspector.get_table_names()) # These tables should exist in user databases user_tables = { "settings", "research", "token_usage", "api_keys", "task_metadata", } for table in user_tables: assert table in tables, f"User table '{table}' not found" class TestCheckDatabaseSchemaWithAlembic: """Tests for check_database_schema function with Alembic.""" @pytest.fixture def migrated_engine(self, tmp_path): """Create a fully migrated database.""" db_path = tmp_path / "schema_check_test.db" engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine) yield engine engine.dispose() def test_check_schema_shows_no_missing_tables(self, migrated_engine): """After migration, no tables should be missing.""" from local_deep_research.database.initialize import ( check_database_schema, ) schema_info = check_database_schema(migrated_engine) # Should have many tables assert len(schema_info["tables"]) > 20 # Should have no missing tables (except users which is excluded) assert len(schema_info["missing_tables"]) == 0 def test_check_schema_excludes_users_table(self, migrated_engine): """check_database_schema should not report users as missing.""" from local_deep_research.database.initialize import ( check_database_schema, ) schema_info = check_database_schema(migrated_engine) # Users should not appear in missing tables assert "users" not in schema_info["missing_tables"] # Users should not appear in existing tables either assert "users" not in schema_info["tables"] class TestNeedsMigrationStates: """Tests for needs_migration in various database states.""" def test_needs_migration_empty_database(self, tmp_path): """Empty database should need migration.""" db_path = tmp_path / "empty_test.db" engine = create_engine(f"sqlite:///{db_path}") try: assert needs_migration(engine) finally: engine.dispose() def test_needs_migration_partial_tables(self, tmp_path): """Database with partial tables should need migration.""" db_path = tmp_path / "partial_tables_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Create only one table with engine.begin() as conn: conn.execute( text("CREATE TABLE settings (id INTEGER PRIMARY KEY)") ) assert needs_migration(engine) finally: engine.dispose() def test_needs_migration_old_revision(self, tmp_path): """Database at old revision should need migration.""" db_path = tmp_path / "old_revision_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Run only to 0001 run_migrations(engine, target="0001") # Should still need migration assert needs_migration(engine) finally: engine.dispose() def test_needs_migration_at_head(self, tmp_path): """Database at head should not need migration.""" db_path = tmp_path / "at_head_test.db" engine = create_engine(f"sqlite:///{db_path}") try: run_migrations(engine) assert not needs_migration(engine) finally: engine.dispose() def test_needs_migration_stamped_database(self, tmp_path): """Stamped database at head should not need migration.""" db_path = tmp_path / "stamped_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Create tables and stamp Base.metadata.create_all(engine) stamp_database(engine, "head") assert not needs_migration(engine) finally: engine.dispose() class TestDataIntegrityAcrossMigrations: """Tests to verify data integrity during migrations.""" @pytest.fixture def database_with_data(self, tmp_path): """Create a database with existing data.""" db_path = tmp_path / "data_integrity_test.db" engine = create_engine(f"sqlite:///{db_path}") # Run initial migration run_migrations(engine, target="0001") # Insert test data into multiple tables with engine.begin() as conn: # Insert into task_metadata conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type, priority) VALUES ('task-1', 'completed', 'research', 1), ('task-2', 'processing', 'benchmark', 2), ('task-3', 'queued', 'research', 3) """ ) ) # Insert into queue_status conn.execute( text( """ INSERT INTO queue_status (active_tasks, queued_tasks) VALUES (2, 5) """ ) ) yield engine engine.dispose() def test_data_preserved_after_column_migration(self, database_with_data): """Data should be preserved when adding columns.""" # Verify current state with database_with_data.connect() as conn: count = conn.execute( text("SELECT COUNT(*) FROM task_metadata") ).fetchone()[0] assert count == 3 # Run migration to add columns run_migrations(database_with_data) # Verify data still exists with database_with_data.connect() as conn: result = conn.execute( text( "SELECT task_id, status, priority FROM task_metadata ORDER BY task_id" ) ).fetchall() assert len(result) == 3 assert result[0][0] == "task-1" assert result[0][1] == "completed" assert result[0][2] == 1 def test_new_columns_have_defaults_for_existing_rows( self, database_with_data ): """New columns should have default values for existing rows.""" # Run migration run_migrations(database_with_data) # Check new column values with database_with_data.connect() as conn: result = conn.execute( text( """ SELECT task_id, progress_current, progress_total FROM task_metadata ORDER BY task_id """ ) ).fetchall() # All rows should have default values (0) for progress columns for row in result: # progress_current and progress_total should be 0 or NULL assert row[1] is None or row[1] == 0 assert row[2] is None or row[2] == 0 def test_multiple_tables_data_preserved(self, database_with_data): """Data in multiple tables should be preserved.""" run_migrations(database_with_data) with database_with_data.connect() as conn: # Check queue_status qs_result = conn.execute( text("SELECT active_tasks, queued_tasks FROM queue_status") ).fetchone() assert qs_result[0] == 2 assert qs_result[1] == 5 # Check task_metadata tm_count = conn.execute( text("SELECT COUNT(*) FROM task_metadata") ).fetchone()[0] assert tm_count == 3 class TestColumnMigrationIdempotency: """Tests for column migration idempotency.""" @pytest.fixture def old_schema_engine(self, tmp_path): """Create database with old schema.""" db_path = tmp_path / "column_idempotent_test.db" engine = create_engine(f"sqlite:///{db_path}") with engine.begin() as conn: conn.execute( text( """ CREATE TABLE task_metadata ( task_id VARCHAR PRIMARY KEY, status VARCHAR NOT NULL, task_type VARCHAR NOT NULL ) """ ) ) yield engine engine.dispose() def test_column_migration_idempotent(self, old_schema_engine): """Running column migration multiple times should be safe.""" # First run run_migrations(old_schema_engine) inspector1 = inspect(old_schema_engine) columns1 = { col["name"] for col in inspector1.get_columns("task_metadata") } # Second run (should not fail) run_migrations(old_schema_engine) # raises on failure inspector2 = inspect(old_schema_engine) columns2 = { col["name"] for col in inspector2.get_columns("task_metadata") } # Columns should be identical assert columns1 == columns2 def test_column_already_exists_no_error(self, tmp_path): """Adding column that already exists should not error.""" db_path = tmp_path / "column_exists_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Create table WITH the columns already with engine.begin() as conn: conn.execute( text( """ CREATE TABLE task_metadata ( task_id VARCHAR PRIMARY KEY, status VARCHAR NOT NULL, task_type VARCHAR NOT NULL, progress_current INTEGER DEFAULT 0, progress_total INTEGER DEFAULT 0, progress_message VARCHAR, metadata_json TEXT ) """ ) ) # Run migrations - should not fail run_migrations(engine) # raises on failure # Should be at head assert get_current_revision(engine) == get_head_revision() finally: engine.dispose() class TestMigrationErrorHandling: """Tests for migration error handling.""" def test_migration_raises_on_corrupted_schema(self, tmp_path): """Migration should raise when alembic_version has wrong schema.""" db_path = tmp_path / "error_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Create a conflicting table structure with engine.begin() as conn: # Create alembic_version with wrong schema conn.execute( text( "CREATE TABLE alembic_version (bad_column INTEGER PRIMARY KEY)" ) ) conn.execute(text("INSERT INTO alembic_version VALUES (999)")) with pytest.raises(Exception): run_migrations(engine) finally: engine.dispose() def test_migration_with_read_only_database(self, tmp_path): """Migration should handle read-only database gracefully.""" import stat db_path = tmp_path / "readonly_test.db" # Create and migrate database first engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine) engine.dispose() # Make database read-only os.chmod(db_path, stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH) engine = create_engine(f"sqlite:///{db_path}") try: # Migration on a read-only DB that's already at head should # be a no-op (no schema changes needed), so it must succeed # and the DB must remain at head. run_migrations(engine) assert get_current_revision(engine) == get_head_revision(), ( "Read-only DB at head should remain at head after no-op migration" ) finally: engine.dispose() # Restore permissions for cleanup os.chmod( db_path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH, ) class TestStampDatabaseBehavior: """Tests for stamp_database function.""" def test_stamp_at_specific_revision(self, tmp_path): """Should be able to stamp at a specific revision.""" db_path = tmp_path / "stamp_specific_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Create tables Base.metadata.create_all(engine) # Stamp at 0001 stamp_database(engine, "0001") # Should be at 0001 assert get_current_revision(engine) == "0001" # Should need migration (not at head) assert needs_migration(engine) finally: engine.dispose() def test_stamp_creates_alembic_version_table(self, tmp_path): """Stamp should create alembic_version table if missing.""" db_path = tmp_path / "stamp_creates_table_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Create some tables but not alembic_version with engine.begin() as conn: conn.execute( text("CREATE TABLE settings (id INTEGER PRIMARY KEY)") ) # Verify alembic_version doesn't exist inspector = inspect(engine) assert "alembic_version" not in inspector.get_table_names() # Stamp stamp_database(engine, "head") # Verify alembic_version now exists inspector = inspect(engine) assert "alembic_version" in inspector.get_table_names() finally: engine.dispose() def test_stamp_overwrites_existing_revision(self, tmp_path): """Stamp should overwrite existing revision.""" db_path = tmp_path / "stamp_overwrite_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Run migrations to 0001 run_migrations(engine, target="0001") assert get_current_revision(engine) == "0001" # Stamp at head stamp_database(engine, "head") assert get_current_revision(engine) == get_head_revision() finally: engine.dispose() class TestMigrationWithForeignKeys: """Tests for migrations with foreign key relationships.""" @pytest.fixture def migrated_engine(self, tmp_path): """Create a fully migrated database.""" db_path = tmp_path / "fk_test.db" engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine) yield engine engine.dispose() def test_foreign_key_tables_created(self, migrated_engine): """Tables with foreign keys should be created correctly.""" inspector = inspect(migrated_engine) tables = inspector.get_table_names() # Tables that have foreign key relationships fk_tables = [ "research_history", "document_collections", "document_chunks", ] for table in fk_tables: assert table in tables, f"FK table '{table}' not found" def test_can_insert_with_foreign_keys(self, migrated_engine): """Should be able to insert data respecting foreign keys.""" # Use research_history and research_resources which have a clear FK relationship with migrated_engine.begin() as conn: # Insert a research history record (uses TEXT columns, UUID-like id) conn.execute( text( """ INSERT INTO research_history (id, query, mode, status, created_at) VALUES ('test-uuid-001', 'Test Query', 'quick', 'completed', datetime('now')) """ ) ) # Insert research resource which references research_history conn.execute( text( """ INSERT INTO research_resources (research_id, title, url, created_at) VALUES ('test-uuid-001', 'Test Resource', 'http://example.com', datetime('now')) """ ) ) # Verify data and foreign key relationship with migrated_engine.connect() as conn: result = conn.execute( text( "SELECT COUNT(*) FROM research_resources WHERE research_id = 'test-uuid-001'" ) ).fetchone() assert result[0] == 1 class TestMigrationWithIndexes: """Tests for index creation during migrations.""" @pytest.fixture def migrated_engine(self, tmp_path): """Create a fully migrated database.""" db_path = tmp_path / "index_test.db" engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine) yield engine engine.dispose() def test_indexes_created(self, migrated_engine): """Important indexes should be created.""" inspector = inspect(migrated_engine) # Check indexes on settings table settings_indexes = inspector.get_indexes("settings") index_names = {idx["name"] for idx in settings_indexes} # Should have index on key column assert any("key" in name.lower() for name in index_names if name) def test_primary_keys_exist(self, migrated_engine): """Primary keys should exist on all tables.""" inspector = inspect(migrated_engine) tables_checked = 0 for table_name in inspector.get_table_names(): if table_name == "alembic_version": continue pk = inspector.get_pk_constraint(table_name) # Most tables should have a primary key # (some junction tables might not) # Verify we can at least retrieve the constraint assert pk is not None tables_checked += 1 # Verify we actually checked some tables assert tables_checked > 0 class TestMigrationRobustness: """Tests for migration robustness and recovery.""" def test_migration_after_crash_simulation(self, tmp_path): """Database should recover from simulated crash during migration.""" db_path = tmp_path / "crash_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Run partial migration run_migrations(engine, target="0001") # Simulate "crash" by just disposing without cleanup engine.dispose() # Reconnect and continue engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine) # raises on failure assert get_current_revision(engine) == get_head_revision() finally: engine.dispose() def test_migration_with_wal_mode(self, tmp_path): """Migration should work with WAL journal mode.""" db_path = tmp_path / "wal_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Enable WAL mode with engine.begin() as conn: conn.execute(text("PRAGMA journal_mode=WAL")) # Run migrations run_migrations(engine) # raises on failure # Verify inspector = inspect(engine) assert "alembic_version" in inspector.get_table_names() finally: engine.dispose() def test_migration_with_busy_timeout(self, tmp_path): """Migration should respect busy timeout settings.""" db_path = tmp_path / "busy_test.db" engine = create_engine( f"sqlite:///{db_path}", connect_args={"timeout": 30}, ) try: run_migrations(engine) # Verify migration actually ran to head, not just "didn't raise" assert get_current_revision(engine) == get_head_revision() finally: engine.dispose() class TestAllModelsCreated: """Tests to verify all expected models are created.""" @pytest.fixture def migrated_engine(self, tmp_path): """Create a fully migrated database.""" db_path = tmp_path / "all_models_test.db" engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine) yield engine engine.dispose() def test_research_models_created(self, migrated_engine): """Research-related tables should be created.""" inspector = inspect(migrated_engine) tables = set(inspector.get_table_names()) research_tables = [ "research", "research_history", "research_tasks", "search_queries", "search_results", ] for table in research_tables: assert table in tables, f"Research table '{table}' not found" def test_library_models_created(self, migrated_engine): """Library-related tables should be created.""" inspector = inspect(migrated_engine) tables = set(inspector.get_table_names()) library_tables = [ "collections", "documents", "document_collections", "source_types", ] for table in library_tables: assert table in tables, f"Library table '{table}' not found" def test_metrics_models_created(self, migrated_engine): """Metrics-related tables should be created.""" inspector = inspect(migrated_engine) tables = set(inspector.get_table_names()) metrics_tables = [ "token_usage", "model_usage", "research_ratings", ] for table in metrics_tables: assert table in tables, f"Metrics table '{table}' not found" def test_news_models_created(self, migrated_engine): """News-related tables should be created.""" inspector = inspect(migrated_engine) tables = set(inspector.get_table_names()) news_tables = [ "news_subscriptions", "news_cards", "news_interests", ] for table in news_tables: assert table in tables, f"News table '{table}' not found" def test_benchmark_models_created(self, migrated_engine): """Benchmark-related tables should be created.""" inspector = inspect(migrated_engine) tables = set(inspector.get_table_names()) benchmark_tables = [ "benchmark_runs", "benchmark_results", "benchmark_configs", ] for table in benchmark_tables: assert table in tables, f"Benchmark table '{table}' not found" class TestColumnTypes: """Tests to verify column types are correct after migration.""" @pytest.fixture def migrated_engine(self, tmp_path): """Create a fully migrated database.""" db_path = tmp_path / "column_types_test.db" engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine) yield engine engine.dispose() def test_task_metadata_columns(self, migrated_engine): """task_metadata should have correct column types.""" inspector = inspect(migrated_engine) columns = { col["name"]: col for col in inspector.get_columns("task_metadata") } # Check required columns exist assert "task_id" in columns assert "status" in columns assert "progress_current" in columns assert "progress_total" in columns assert "progress_message" in columns assert "metadata_json" in columns def test_settings_columns(self, migrated_engine): """settings should have correct column types.""" inspector = inspect(migrated_engine) columns = { col["name"]: col for col in inspector.get_columns("settings") } assert "id" in columns assert "key" in columns assert "value" in columns assert "name" in columns def test_datetime_columns_exist(self, migrated_engine): """Tables should have datetime columns where expected.""" inspector = inspect(migrated_engine) # Check research table has datetime columns research_columns = { col["name"] for col in inspector.get_columns("research") } assert "created_at" in research_columns # Check task_metadata has datetime columns task_columns = { col["name"] for col in inspector.get_columns("task_metadata") } assert "created_at" in task_columns class TestMigrationPerformance: """Performance-related tests for migrations.""" def test_migration_completes_in_reasonable_time(self, tmp_path): """Migration should complete within reasonable time.""" import time db_path = tmp_path / "perf_test.db" engine = create_engine(f"sqlite:///{db_path}") try: start = time.time() run_migrations(engine) elapsed = time.time() - start # Should complete in under 30 seconds even on slow systems assert elapsed < 30, ( f"Migration took {elapsed:.1f}s, expected < 30s" ) finally: engine.dispose() def test_migration_with_large_existing_data(self, tmp_path): """Migration should handle tables with existing data.""" db_path = tmp_path / "large_data_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Run initial migration run_migrations(engine, target="0001") # Insert many rows with engine.begin() as conn: for i in range(100): conn.execute( text( f""" INSERT INTO task_metadata (task_id, status, task_type) VALUES ('task-{i}', 'completed', 'research') """ ) ) # Run remaining migrations run_migrations(engine) # raises on failure # Verify all data preserved with engine.connect() as conn: count = conn.execute( text("SELECT COUNT(*) FROM task_metadata") ).fetchone()[0] assert count == 100 finally: engine.dispose() class TestDatabaseManagerIntegration: """Integration tests with DatabaseManager (encrypted databases).""" @pytest.fixture def sqlcipher_available(self): """Check if SQLCipher is available.""" import importlib.util if importlib.util.find_spec("sqlcipher3") is None: pytest.skip("SQLCipher not available") return True def test_migration_through_database_manager_flow( self, tmp_path, sqlcipher_available ): """Test migrations work through the full DatabaseManager flow.""" import sqlcipher3 db_path = tmp_path / "dm_flow_test.db" password = "test_password" # Simulate DatabaseManager creating a new database def create_connection(): conn = sqlcipher3.connect(str(db_path), check_same_thread=False) cursor = conn.cursor() cursor.execute(f"PRAGMA key = '{password}'") cursor.execute("PRAGMA cipher_page_size = 4096") cursor.execute("PRAGMA kdf_iter = 256000") cursor.execute("PRAGMA cipher_hmac_algorithm = HMAC_SHA512") cursor.execute("PRAGMA cipher_kdf_algorithm = PBKDF2_HMAC_SHA512") cursor.close() return conn engine = create_engine("sqlite://", creator=create_connection) try: # This simulates what initialize_database does run_migrations(engine) # raises on failure # Verify database is properly set up inspector = inspect(engine) tables = inspector.get_table_names() assert "alembic_version" in tables assert "settings" in tables assert "task_metadata" in tables # Verify we can query with engine.connect() as conn: result = conn.execute( text("SELECT version_num FROM alembic_version") ).fetchone() assert result[0] == get_head_revision() finally: engine.dispose() def test_reopening_encrypted_database(self, tmp_path, sqlcipher_available): """Test that reopening an encrypted database works correctly.""" import sqlcipher3 db_path = tmp_path / "reopen_test.db" password = "test_password" def create_connection(): conn = sqlcipher3.connect(str(db_path), check_same_thread=False) cursor = conn.cursor() cursor.execute(f"PRAGMA key = '{password}'") cursor.execute("PRAGMA cipher_page_size = 4096") cursor.execute("PRAGMA kdf_iter = 256000") cursor.execute("PRAGMA cipher_hmac_algorithm = HMAC_SHA512") cursor.execute("PRAGMA cipher_kdf_algorithm = PBKDF2_HMAC_SHA512") cursor.close() return conn # First open - create and migrate engine1 = create_engine("sqlite://", creator=create_connection) run_migrations(engine1) rev1 = get_current_revision(engine1) engine1.dispose() # Second open - should recognize existing migration engine2 = create_engine("sqlite://", creator=create_connection) run_migrations(engine2) # raises on failure rev2 = get_current_revision(engine2) assert rev1 == rev2 # Should not need migration assert not needs_migration(engine2) engine2.dispose() class TestMigrationWithDifferentEngineConfigs: """Test migrations with various engine configurations.""" def test_migration_with_echo_enabled(self, tmp_path): """Migration should work with SQL echo enabled.""" db_path = tmp_path / "echo_test.db" engine = create_engine(f"sqlite:///{db_path}", echo=False) try: run_migrations(engine) assert get_current_revision(engine) == get_head_revision() finally: engine.dispose() def test_migration_with_pool_pre_ping(self, tmp_path): """Migration should work with pool pre-ping enabled.""" db_path = tmp_path / "preping_test.db" engine = create_engine(f"sqlite:///{db_path}", pool_pre_ping=True) try: run_migrations(engine) assert get_current_revision(engine) == get_head_revision() finally: engine.dispose() def test_migration_with_static_pool(self, tmp_path): """Migration should work with StaticPool.""" from sqlalchemy.pool import StaticPool db_path = tmp_path / "static_pool_test.db" engine = create_engine( f"sqlite:///{db_path}", poolclass=StaticPool, connect_args={"check_same_thread": False}, ) try: run_migrations(engine) assert get_current_revision(engine) == get_head_revision() finally: engine.dispose() def test_migration_in_memory_database(self): """Migration should work with in-memory database.""" from sqlalchemy.pool import StaticPool engine = create_engine( "sqlite:///:memory:", poolclass=StaticPool, connect_args={"check_same_thread": False}, ) try: run_migrations(engine) # raises on failure inspector = inspect(engine) assert "alembic_version" in inspector.get_table_names() finally: engine.dispose() class TestDowngradeMigrations: """Tests for downgrade functionality.""" @pytest.fixture def migrated_engine(self, tmp_path): """Create a fully migrated database.""" db_path = tmp_path / "downgrade_test.db" engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine) yield engine engine.dispose() def test_downgrade_from_head_to_specific_revision(self, tmp_path): """Test downgrade from head to a specific revision. Starts at 0008 (last fully-reversible) rather than head; 0010 is intentionally non-reversible (SQLite ALTER TABLE limitations). """ from alembic import command db_path = tmp_path / "downgrade_specific_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Upgrade to 0008 (last reversible) config = get_alembic_config(engine) with engine.begin() as conn: config.attributes["connection"] = conn command.upgrade(config, "0008") assert get_current_revision(engine) == "0008" # Downgrade to 0001 with engine.begin() as conn: config.attributes["connection"] = conn command.downgrade(config, "0001") # Verify revision assert get_current_revision(engine) == "0001" # Verify progress columns were removed inspector = inspect(engine) columns = { col["name"] for col in inspector.get_columns("task_metadata") } assert "progress_current" not in columns assert "progress_total" not in columns finally: engine.dispose() def test_downgrade_preserves_core_data(self, tmp_path): """Test that downgrade preserves data in core columns. Starts at 0008 (last fully-reversible) rather than head; 0010 is intentionally non-reversible (SQLite ALTER TABLE limitations). """ from alembic import command db_path = tmp_path / "downgrade_data_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Upgrade to 0008 and insert data config = get_alembic_config(engine) with engine.begin() as conn: config.attributes["connection"] = conn command.upgrade(config, "0008") with engine.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type, priority) VALUES ('task-preserve-1', 'completed', 'research', 5) """ ) ) # Downgrade with engine.begin() as conn: config.attributes["connection"] = conn command.downgrade(config, "0001") # Verify core data preserved with engine.connect() as conn: result = conn.execute( text( "SELECT task_id, status, priority FROM task_metadata WHERE task_id = 'task-preserve-1'" ) ).fetchone() assert result is not None assert result[0] == "task-preserve-1" assert result[1] == "completed" assert result[2] == 5 finally: engine.dispose() def test_full_downgrade_to_empty_database(self, tmp_path): """Test full downgrade removes all tables. Starts from 0008 (last fully-reversible revision) rather than head, since 0010 is intentionally non-reversible (SQLite ALTER TABLE limitations against legacy unnamed constraints + FK-target columns; see NON_REVERSIBLE_REVISIONS). """ from alembic import command db_path = tmp_path / "full_downgrade_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Upgrade to 0008 (last reversible) instead of head config = get_alembic_config(engine) with engine.begin() as conn: config.attributes["connection"] = conn command.upgrade(config, "0008") # Verify tables exist inspector = inspect(engine) tables_before = set(inspector.get_table_names()) assert len(tables_before) > 10 # Downgrade to base (empty) with engine.begin() as conn: config.attributes["connection"] = conn command.downgrade(config, "base") # Verify most tables removed (only alembic_version may remain) inspector = inspect(engine) tables_after = set(inspector.get_table_names()) # After full downgrade, should have minimal tables assert len(tables_after) <= 1 # Only alembic_version or empty finally: engine.dispose() def test_downgrade_then_upgrade_roundtrip(self, tmp_path): """Test downgrade and re-upgrade works correctly. Starts from 0008 (last fully-reversible revision) rather than head, since 0010 is intentionally non-reversible. After the roundtrip we re-upgrade to head to verify forward path still works through the non-reversible step. """ from alembic import command db_path = tmp_path / "roundtrip_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Upgrade to 0008 (last reversible) config = get_alembic_config(engine) with engine.begin() as conn: config.attributes["connection"] = conn command.upgrade(config, "0008") # Insert data with engine.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type) VALUES ('roundtrip-task', 'queued', 'test') """ ) ) # Downgrade to 0001 with engine.begin() as conn: config.attributes["connection"] = conn command.downgrade(config, "0001") # Re-upgrade to head (passes through 0010 in forward direction) run_migrations(engine) assert get_current_revision(engine) == get_head_revision() # Verify data still exists with engine.connect() as conn: result = conn.execute( text( "SELECT task_id FROM task_metadata WHERE task_id = 'roundtrip-task'" ) ).fetchone() assert result is not None finally: engine.dispose() class TestMigrationWithExistingData: """Tests for migration with various data scenarios.""" @pytest.fixture def engine_at_0001(self, tmp_path): """Create a database at revision 0001 for data tests.""" db_path = tmp_path / "data_test.db" engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine, target="0001") yield engine engine.dispose() def test_migration_with_thousands_of_rows(self, tmp_path): """Test migration handles tables with thousands of rows.""" db_path = tmp_path / "large_data_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Run initial migration run_migrations(engine, target="0001") # Insert 5000 rows with engine.begin() as conn: for batch in range(50): values = ", ".join( [ f"('task-{batch}-{i}', 'completed', 'research')" for i in range(100) ] ) conn.execute( text( f""" INSERT INTO task_metadata (task_id, status, task_type) VALUES {values} """ ) ) # Verify count with engine.connect() as conn: count = conn.execute( text("SELECT COUNT(*) FROM task_metadata") ).fetchone()[0] assert count == 5000 # Run remaining migrations run_migrations(engine) # raises on failure # Verify all data preserved with engine.connect() as conn: count = conn.execute( text("SELECT COUNT(*) FROM task_metadata") ).fetchone()[0] assert count == 5000 # Verify new columns exist inspector = inspect(engine) columns = { col["name"] for col in inspector.get_columns("task_metadata") } assert "progress_current" in columns finally: engine.dispose() def test_migration_preserves_json_data(self, engine_at_0001): """Test migration preserves JSON data in columns.""" # Insert data with JSON - use research_history which has simpler constraints with engine_at_0001.begin() as conn: conn.execute( text( """ INSERT INTO research_history (id, query, mode, status, created_at, research_meta) VALUES ('json-test-id', 'test query', 'quick', 'completed', datetime('now'), '{"nested": {"key": "value"}, "array": [1, 2, 3]}') """ ) ) # Run migration run_migrations(engine_at_0001) # Verify JSON data preserved with engine_at_0001.connect() as conn: result = conn.execute( text( "SELECT research_meta FROM research_history WHERE id = 'json-test-id'" ) ).fetchone() assert result is not None import json data = json.loads(result[0]) assert data["nested"]["key"] == "value" assert data["array"] == [1, 2, 3] def test_migration_preserves_datetime_precision(self, engine_at_0001): """Test migration preserves datetime precision.""" # Insert data with precise datetime with engine_at_0001.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type, created_at) VALUES ('datetime-test', 'completed', 'research', '2025-06-15 14:30:45.123456') """ ) ) # Run migration run_migrations(engine_at_0001) # Verify datetime preserved with engine_at_0001.connect() as conn: result = conn.execute( text( "SELECT created_at FROM task_metadata WHERE task_id = 'datetime-test'" ) ).fetchone() assert result is not None # SQLite stores as string, verify it contains the microseconds datetime_str = str(result[0]) assert "14:30:45" in datetime_str def test_migration_with_null_values_in_nullable_columns( self, engine_at_0001 ): """Test migration handles NULL values in nullable columns.""" # Insert data with NULL values with engine_at_0001.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type, error_message, started_at, completed_at) VALUES ('null-test', 'queued', 'research', NULL, NULL, NULL) """ ) ) # Run migration run_migrations(engine_at_0001) # Verify NULL values preserved and new columns also NULL with engine_at_0001.connect() as conn: result = conn.execute( text( """ SELECT error_message, started_at, completed_at, progress_message FROM task_metadata WHERE task_id = 'null-test' """ ) ).fetchone() assert result is not None assert result[0] is None # error_message assert result[1] is None # started_at assert result[2] is None # completed_at assert result[3] is None # progress_message (new column) def test_migration_with_unicode_data(self, engine_at_0001): """Test migration preserves unicode and special characters.""" # Insert unicode data with engine_at_0001.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type, error_message) VALUES ('unicode-test', 'failed', 'research', 'Error: 日本語テスト 🔥 Ñoño émojis café') """ ) ) # Run migration run_migrations(engine_at_0001) # Verify unicode preserved with engine_at_0001.connect() as conn: result = conn.execute( text( "SELECT error_message FROM task_metadata WHERE task_id = 'unicode-test'" ) ).fetchone() assert result is not None assert "日本語テスト" in result[0] assert "🔥" in result[0] assert "café" in result[0] def test_migration_with_empty_strings(self, engine_at_0001): """Test migration handles empty string values.""" with engine_at_0001.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type, error_message) VALUES ('empty-string-test', 'completed', 'research', '') """ ) ) run_migrations(engine_at_0001) with engine_at_0001.connect() as conn: result = conn.execute( text( "SELECT error_message FROM task_metadata WHERE task_id = 'empty-string-test'" ) ).fetchone() assert result is not None assert result[0] == "" class TestConcurrentDatabaseAccess: """Tests for concurrent database access during migrations.""" def test_migration_while_database_is_being_read(self, tmp_path): """Test migration completes while concurrent reads occur.""" import threading import time db_path = tmp_path / "concurrent_read_test.db" engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine, target="0001") # Insert some test data with engine.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type) VALUES ('concurrent-test', 'completed', 'research') """ ) ) read_results = [] read_errors = [] def concurrent_reader(): """Read from database concurrently.""" try: for _ in range(10): with engine.connect() as conn: result = conn.execute( text("SELECT COUNT(*) FROM task_metadata") ).fetchone() read_results.append(result[0]) time.sleep(0.01) except Exception as e: read_errors.append(str(e)) try: # Start reader thread reader = threading.Thread(target=concurrent_reader) reader.start() # Run migration while reading run_migrations(engine) # raises on failure reader.join() # Verify reads succeeded assert len(read_errors) == 0, f"Read errors: {read_errors}" assert len(read_results) > 0 finally: engine.dispose() def test_two_engines_migrate_same_database(self, tmp_path): """Test two processes trying to migrate same database.""" db_path = tmp_path / "dual_migrate_test.db" engine1 = create_engine(f"sqlite:///{db_path}") engine2 = create_engine(f"sqlite:///{db_path}") try: # Both engines try to migrate run_migrations(engine1) # raises on failure run_migrations(engine2) # raises on failure # Both should see same revision rev1 = get_current_revision(engine1) rev2 = get_current_revision(engine2) assert rev1 == rev2 == get_head_revision() finally: engine1.dispose() engine2.dispose() def test_migration_with_sqlite_busy_timeout(self, tmp_path): """Test migration respects SQLite busy timeout.""" db_path = tmp_path / "busy_timeout_test.db" # Create engine with explicit busy timeout engine = create_engine( f"sqlite:///{db_path}", connect_args={"timeout": 30}, ) try: run_migrations(engine) # raises on failure # Verify database is correctly migrated assert get_current_revision(engine) == get_head_revision() finally: engine.dispose() def test_sequential_migrations_different_engines(self, tmp_path): """Test sequential migrations from different engine instances.""" db_path = tmp_path / "sequential_test.db" # First engine migrates to 0001 engine1 = create_engine(f"sqlite:///{db_path}") run_migrations(engine1, target="0001") engine1.dispose() # Second engine continues to head engine2 = create_engine(f"sqlite:///{db_path}") run_migrations(engine2) # raises on failure assert get_current_revision(engine2) == get_head_revision() engine2.dispose() # Third engine verifies no migration needed engine3 = create_engine(f"sqlite:///{db_path}") assert not needs_migration(engine3) engine3.dispose() class TestMigrationRecovery: """Tests for migration recovery scenarios.""" def test_recovery_after_partial_migration(self, tmp_path): """Test recovery after a partial migration state.""" db_path = tmp_path / "partial_recovery_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Create a partial state - some tables but not all with engine.begin() as conn: conn.execute( text( """ CREATE TABLE settings ( id INTEGER PRIMARY KEY, key VARCHAR UNIQUE, value VARCHAR, name VARCHAR, description VARCHAR, type VARCHAR ) """ ) ) conn.execute( text( """ CREATE TABLE task_metadata ( task_id VARCHAR PRIMARY KEY, status VARCHAR NOT NULL, task_type VARCHAR NOT NULL ) """ ) ) # Try to recover by running full migration run_migrations(engine) # raises on failure # Verify recovery succeeded inspector = inspect(engine) tables = inspector.get_table_names() assert "alembic_version" in tables assert "research" in tables # Should have been created finally: engine.dispose() def test_database_state_after_interrupted_migration(self, tmp_path): """Test database state is consistent after simulated interruption.""" db_path = tmp_path / "interrupted_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Run partial migration run_migrations(engine, target="0001") # Simulate "interruption" by just stopping here engine.dispose() # Reconnect and check state engine = create_engine(f"sqlite:///{db_path}") # Database should be at 0001 assert get_current_revision(engine) == "0001" # Should be able to continue run_migrations(engine) # raises on failure assert get_current_revision(engine) == get_head_revision() finally: engine.dispose() def test_re_running_migration_after_failure(self, tmp_path): """Test re-running migration after a previous failure.""" db_path = tmp_path / "rerun_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # First run succeeds run_migrations(engine) # raises on failure # Simulate database being in inconsistent state # by manually inserting conflicting data with engine.begin() as conn: # This shouldn't cause issues on re-run conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type) VALUES ('conflict-test', 'completed', 'research') """ ) ) # Re-running should still work run_migrations(engine) # raises on failure # Data should be preserved with engine.connect() as conn: result = conn.execute( text( "SELECT task_id FROM task_metadata WHERE task_id = 'conflict-test'" ) ).fetchone() assert result is not None finally: engine.dispose() def test_migration_with_foreign_key_violations_in_source(self, tmp_path): """Test migration handles pre-existing data correctly.""" db_path = tmp_path / "fk_data_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Run full migration first run_migrations(engine) # Insert valid data with engine.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type, priority) VALUES ('fk-test-1', 'completed', 'research', 1), ('fk-test-2', 'queued', 'benchmark', 2) """ ) ) # Re-run migration (should be idempotent) run_migrations(engine) # raises on failure # Verify data intact with engine.connect() as conn: count = conn.execute( text( "SELECT COUNT(*) FROM task_metadata WHERE task_id LIKE 'fk-test%'" ) ).fetchone()[0] assert count == 2 finally: engine.dispose() class TestSchemaValidation: """Tests for schema correctness after migration.""" @pytest.fixture def migrated_engine(self, tmp_path): """Create a fully migrated database.""" db_path = tmp_path / "schema_validation_test.db" engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine) yield engine engine.dispose() def test_all_foreign_key_relationships_valid(self, migrated_engine): """Test that all foreign key relationships reference existing tables.""" inspector = inspect(migrated_engine) existing_tables = set(inspector.get_table_names()) for table_name in existing_tables: if table_name == "alembic_version": continue fks = inspector.get_foreign_keys(table_name) for fk in fks: referred_table = fk.get("referred_table") if referred_table: assert referred_table in existing_tables, ( f"Table {table_name} has FK to non-existent table {referred_table}" ) def test_all_expected_indexes_exist(self, migrated_engine): """Test that important indexes exist.""" inspector = inspect(migrated_engine) # Check settings table has index on key if "settings" in inspector.get_table_names(): indexes = inspector.get_indexes("settings") index_columns = [] for idx in indexes: index_columns.extend(idx.get("column_names", [])) # Settings should have unique constraint or index on key # Check if 'key' has unique constraint unique_constraints = inspector.get_unique_constraints("settings") has_key_constraint = any( "key" in uc.get("column_names", []) for uc in unique_constraints ) has_key_index = "key" in index_columns assert has_key_constraint or has_key_index, ( "settings.key should have unique index" ) def test_column_types_match_expectations(self, migrated_engine): """Test that column types are as expected.""" inspector = inspect(migrated_engine) # Check task_metadata columns tm_columns = { col["name"]: col for col in inspector.get_columns("task_metadata") } # task_id should be string type assert ( "VARCHAR" in str(tm_columns["task_id"]["type"]).upper() or "TEXT" in str(tm_columns["task_id"]["type"]).upper() ) # progress_current should be integer assert ( "INTEGER" in str(tm_columns["progress_current"]["type"]).upper() or "INT" in str(tm_columns["progress_current"]["type"]).upper() ) def test_nullable_constraints_correct(self, migrated_engine): """Test nullable constraints are set correctly.""" inspector = inspect(migrated_engine) # Check task_metadata tm_columns = { col["name"]: col for col in inspector.get_columns("task_metadata") } # status should not be nullable assert tm_columns["status"]["nullable"] is False # error_message should be nullable assert tm_columns["error_message"]["nullable"] is True # progress_message should be nullable assert tm_columns["progress_message"]["nullable"] is True def test_primary_keys_defined_correctly(self, migrated_engine): """Test all tables have properly defined primary keys.""" inspector = inspect(migrated_engine) for table_name in inspector.get_table_names(): if table_name == "alembic_version": continue pk = inspector.get_pk_constraint(table_name) pk_columns = pk.get("constrained_columns", []) # Most tables should have a primary key # (some association tables might not) if table_name not in ["document_collections"]: # Junction tables assert len(pk_columns) > 0, ( f"Table {table_name} missing primary key" ) def test_all_model_tables_exist(self, migrated_engine): """Test that all model tables are created.""" inspector = inspect(migrated_engine) tables = set(inspector.get_table_names()) # Core tables that must exist required_tables = [ "settings", "task_metadata", "queue_status", "research", "research_history", "token_usage", "benchmark_runs", "news_subscriptions", ] for table in required_tables: assert table in tables, f"Required table '{table}' not found" class TestMigrationVersionConsistency: """Tests for migration version tracking consistency.""" @pytest.fixture def migrated_engine(self, tmp_path): """Create a fully migrated database.""" db_path = tmp_path / "version_consistency_test.db" engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine) yield engine engine.dispose() def test_alembic_version_never_has_multiple_rows(self, migrated_engine): """Test that alembic_version table never has multiple rows.""" with migrated_engine.connect() as conn: result = conn.execute( text("SELECT COUNT(*) FROM alembic_version") ).fetchone() assert result[0] == 1, "alembic_version should have exactly one row" def test_revision_chain_has_no_gaps(self): """Test that the revision chain is continuous with no gaps.""" from alembic.config import Config from alembic.script import ScriptDirectory migrations_dir = get_migrations_dir() config = Config() config.set_main_option("script_location", str(migrations_dir)) script = ScriptDirectory.from_config(config) # Get all revisions revisions = list(script.walk_revisions()) # Build the chain from head to base head = script.get_current_head() current = head chain = [] while current is not None: rev = script.get_revision(current) chain.append(current) current = rev.down_revision # Verify chain matches walk_revisions assert len(chain) == len(revisions), "Revision chain has gaps" def test_head_revision_matches_latest_migration_file(self): """Test that head revision matches the latest migration file.""" from alembic.config import Config from alembic.script import ScriptDirectory migrations_dir = get_migrations_dir() config = Config() config.set_main_option("script_location", str(migrations_dir)) script = ScriptDirectory.from_config(config) head = script.get_current_head() # Current head should be the latest migration assert head == get_head_revision(), ( f"Head revision mismatch: expected {get_head_revision()}, got {head}" ) def test_all_revision_files_have_valid_structure(self): """Test that all migration files have required functions.""" versions_dir = get_migrations_dir() / "versions" for py_file in versions_dir.glob("*.py"): if py_file.name.startswith("__"): continue content = py_file.read_text() # Check required elements assert "revision = " in content, f"{py_file.name} missing revision" assert "down_revision = " in content, ( f"{py_file.name} missing down_revision" ) assert "def upgrade(" in content, ( f"{py_file.name} missing upgrade function" ) assert "def downgrade(" in content, ( f"{py_file.name} missing downgrade function" ) def test_revision_ids_are_unique(self): """Test that all revision IDs are unique.""" from alembic.config import Config from alembic.script import ScriptDirectory migrations_dir = get_migrations_dir() config = Config() config.set_main_option("script_location", str(migrations_dir)) script = ScriptDirectory.from_config(config) revisions = list(script.walk_revisions()) revision_ids = [rev.revision for rev in revisions] assert len(revision_ids) == len(set(revision_ids)), ( "Duplicate revision IDs found" ) def test_stamp_updates_version_correctly(self, tmp_path): """Test that stamping updates version correctly.""" db_path = tmp_path / "stamp_version_test.db" engine = create_engine(f"sqlite:///{db_path}") try: Base.metadata.create_all(engine) # Stamp at 0001 stamp_database(engine, "0001") assert get_current_revision(engine) == "0001" with engine.connect() as conn: count = conn.execute( text("SELECT COUNT(*) FROM alembic_version") ).fetchone()[0] assert count == 1 # Stamp at head stamp_database(engine, "head") assert get_current_revision(engine) == get_head_revision() with engine.connect() as conn: count = conn.execute( text("SELECT COUNT(*) FROM alembic_version") ).fetchone()[0] assert count == 1 # Still only one row finally: engine.dispose() class TestEdgeCaseTableNames: """Tests for edge cases with table names and SQL handling.""" @pytest.fixture def migrated_engine(self, tmp_path): """Create a fully migrated database.""" db_path = tmp_path / "edge_case_names_test.db" engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine) yield engine engine.dispose() def test_tables_with_underscores_work(self, migrated_engine): """Test that tables with underscores in names work correctly.""" inspector = inspect(migrated_engine) tables = inspector.get_table_names() # Find tables with underscores underscore_tables = [t for t in tables if "_" in t] assert len(underscore_tables) > 0, "Should have tables with underscores" # Verify they can be queried for table in underscore_tables[:3]: # Test first 3 with migrated_engine.connect() as conn: # Should not raise conn.execute(text(f"SELECT * FROM {table} LIMIT 1")) def test_reserved_sql_keywords_handled(self, migrated_engine): """Test that any reserved SQL keywords are properly escaped.""" inspector = inspect(migrated_engine) # Check all tables can be queried for table_name in inspector.get_table_names(): with migrated_engine.connect() as conn: # This should work even if table name is a reserved word conn.execute(text(f'SELECT * FROM "{table_name}" LIMIT 1')) def test_column_names_with_special_chars(self, migrated_engine): """Test columns can be accessed correctly.""" inspector = inspect(migrated_engine) for table_name in inspector.get_table_names(): if table_name == "alembic_version": continue columns = inspector.get_columns(table_name) for col in columns[:5]: # Test first 5 columns per table col_name = col["name"] with migrated_engine.connect() as conn: # Should be able to select each column conn.execute( text(f'SELECT "{col_name}" FROM "{table_name}" LIMIT 1') ) def test_case_sensitivity_in_table_names(self, migrated_engine): """Test case handling in table names (SQLite is case-insensitive).""" inspector = inspect(migrated_engine) tables = inspector.get_table_names() # SQLite table names are case-insensitive by default if "settings" in tables: with migrated_engine.connect() as conn: # Both should work in SQLite conn.execute(text("SELECT * FROM settings LIMIT 1")) conn.execute(text("SELECT * FROM SETTINGS LIMIT 1")) def test_numeric_prefixed_table_handling(self, migrated_engine): """Test handling of tables that might have numeric-looking parts.""" # Alembic uses revision IDs like 0001, 0002 # Verify these don't cause issues from alembic.config import Config from alembic.script import ScriptDirectory migrations_dir = get_migrations_dir() config = Config() config.set_main_option("script_location", str(migrations_dir)) script = ScriptDirectory.from_config(config) # Should be able to get revisions with numeric IDs rev_0001 = script.get_revision("0001") rev_0002 = script.get_revision("0002") assert rev_0001 is not None assert rev_0002 is not None def test_empty_table_operations(self, migrated_engine): """Test operations on empty tables work correctly.""" inspector = inspect(migrated_engine) for table_name in inspector.get_table_names(): if table_name == "alembic_version": continue with migrated_engine.connect() as conn: # Count on empty table result = conn.execute( text(f'SELECT COUNT(*) FROM "{table_name}"') ).fetchone() # Should return 0 or some number, not error assert result[0] >= 0 def test_long_value_insertion(self, migrated_engine): """Test handling of long string values.""" # Create a very long string long_string = "x" * 10000 with migrated_engine.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type, error_message) VALUES ('long-value-test', 'failed', 'research', :msg) """ ), {"msg": long_string}, ) # Verify retrieval with migrated_engine.connect() as conn: result = conn.execute( text( "SELECT error_message FROM task_metadata WHERE task_id = 'long-value-test'" ) ).fetchone() assert result is not None assert len(result[0]) == 10000 class TestSQLCipherMigrationsComprehensive: """Comprehensive tests for SQLCipher encrypted database migrations. These tests cover scenarios specific to encrypted databases including: - Real file-based encrypted databases (not in-memory) - Data preservation during migration - Wrong password handling - Password change followed by migration """ @pytest.fixture def sqlcipher_available(self): """Check if SQLCipher is available.""" import importlib.util if importlib.util.find_spec("sqlcipher3") is None: pytest.skip("SQLCipher not available") return True @pytest.fixture def encrypted_file_engine(self, tmp_path, sqlcipher_available): """Create an encrypted file-based SQLCipher database (not in-memory).""" import sqlcipher3 db_path = tmp_path / "encrypted_file_test.db" password = "secure_test_password_123!" def create_connection(): conn = sqlcipher3.connect(str(db_path), check_same_thread=False) cursor = conn.cursor() cursor.execute(f"PRAGMA key = '{password}'") cursor.execute("PRAGMA cipher_page_size = 4096") cursor.execute("PRAGMA kdf_iter = 256000") cursor.execute("PRAGMA cipher_hmac_algorithm = HMAC_SHA512") cursor.execute("PRAGMA cipher_kdf_algorithm = PBKDF2_HMAC_SHA512") cursor.close() return conn engine = create_engine("sqlite://", creator=create_connection) yield {"engine": engine, "db_path": db_path, "password": password} engine.dispose() def test_migration_on_encrypted_file_database( self, encrypted_file_engine, sqlcipher_available ): """Test migrations work correctly on encrypted file-based databases.""" engine = encrypted_file_engine["engine"] db_path = encrypted_file_engine["db_path"] # Run migrations run_migrations(engine) # raises on failure # Verify tables created inspector = inspect(engine) tables = inspector.get_table_names() assert "alembic_version" in tables assert "settings" in tables assert "task_metadata" in tables assert "research" in tables # Verify the file was actually created (not just in-memory) assert db_path.exists() # Encrypted file should have some content assert db_path.stat().st_size > 0 def test_migration_preserves_encrypted_data( self, tmp_path, sqlcipher_available ): """Test that data is preserved during migration on encrypted database.""" import sqlcipher3 db_path = tmp_path / "data_preserve_encrypted.db" password = "data_preserve_password" def create_connection(): conn = sqlcipher3.connect(str(db_path), check_same_thread=False) cursor = conn.cursor() cursor.execute(f"PRAGMA key = '{password}'") cursor.execute("PRAGMA cipher_page_size = 4096") cursor.execute("PRAGMA kdf_iter = 256000") cursor.execute("PRAGMA cipher_hmac_algorithm = HMAC_SHA512") cursor.execute("PRAGMA cipher_kdf_algorithm = PBKDF2_HMAC_SHA512") cursor.close() return conn engine = create_engine("sqlite://", creator=create_connection) try: # Run initial migration to 0001 run_migrations(engine, target="0001") # Insert test data with engine.begin() as conn: for i in range(100): conn.execute( text( f""" INSERT INTO task_metadata (task_id, status, task_type, priority) VALUES ('encrypted-task-{i}', 'completed', 'research', {i}) """ ) ) # Verify data count before migration with engine.connect() as conn: count_before = conn.execute( text("SELECT COUNT(*) FROM task_metadata") ).fetchone()[0] assert count_before == 100 # Run remaining migrations run_migrations(engine) # raises on failure # Verify all data preserved after migration with engine.connect() as conn: count_after = conn.execute( text("SELECT COUNT(*) FROM task_metadata") ).fetchone()[0] assert count_after == 100 # Verify specific data integrity sample = conn.execute( text( "SELECT task_id, priority FROM task_metadata WHERE task_id = 'encrypted-task-50'" ) ).fetchone() assert sample is not None assert sample[1] == 50 finally: engine.dispose() def test_wrong_password_fails_gracefully( self, tmp_path, sqlcipher_available ): """Test that using wrong password on encrypted database fails gracefully.""" import sqlcipher3 db_path = tmp_path / "wrong_password_test.db" correct_password = "correct_password_123" wrong_password = "wrong_password_456" # Create database with correct password def create_connection_correct(): conn = sqlcipher3.connect(str(db_path), check_same_thread=False) cursor = conn.cursor() cursor.execute(f"PRAGMA key = '{correct_password}'") cursor.execute("PRAGMA cipher_page_size = 4096") cursor.close() return conn engine1 = create_engine("sqlite://", creator=create_connection_correct) run_migrations(engine1) engine1.dispose() # Try to open with wrong password def create_connection_wrong(): conn = sqlcipher3.connect(str(db_path), check_same_thread=False) cursor = conn.cursor() cursor.execute(f"PRAGMA key = '{wrong_password}'") cursor.execute("PRAGMA cipher_page_size = 4096") cursor.close() return conn engine2 = create_engine("sqlite://", creator=create_connection_wrong) # Trying to access the database should fail with pytest.raises(Exception): with engine2.connect() as conn: conn.execute(text("SELECT * FROM settings LIMIT 1")) engine2.dispose() def test_encrypted_database_reopen_preserves_revision( self, tmp_path, sqlcipher_available ): """Test that reopening encrypted database preserves migration revision.""" import sqlcipher3 db_path = tmp_path / "reopen_revision_test.db" password = "reopen_test_password" def create_connection(): conn = sqlcipher3.connect(str(db_path), check_same_thread=False) cursor = conn.cursor() cursor.execute(f"PRAGMA key = '{password}'") cursor.execute("PRAGMA cipher_page_size = 4096") cursor.execute("PRAGMA kdf_iter = 256000") cursor.close() return conn # First session - create and migrate engine1 = create_engine("sqlite://", creator=create_connection) run_migrations(engine1) revision1 = get_current_revision(engine1) engine1.dispose() # Second session - reopen and check engine2 = create_engine("sqlite://", creator=create_connection) revision2 = get_current_revision(engine2) assert revision1 == revision2 assert revision2 == get_head_revision() # Should not need migration assert not needs_migration(engine2) engine2.dispose() def test_encrypted_migration_with_large_data( self, tmp_path, sqlcipher_available ): """Test migration on encrypted database with realistic data volume.""" import sqlcipher3 db_path = tmp_path / "large_encrypted_test.db" password = "large_data_password" def create_connection(): conn = sqlcipher3.connect(str(db_path), check_same_thread=False) cursor = conn.cursor() cursor.execute(f"PRAGMA key = '{password}'") cursor.execute("PRAGMA cipher_page_size = 4096") cursor.execute("PRAGMA kdf_iter = 256000") cursor.close() return conn engine = create_engine("sqlite://", creator=create_connection) try: # Run initial migration run_migrations(engine, target="0001") # Insert 1000 rows (realistic for a user database) with engine.begin() as conn: for batch in range(10): values = ", ".join( [ f"('large-task-{batch}-{i}', 'completed', 'research')" for i in range(100) ] ) conn.execute( text( f""" INSERT INTO task_metadata (task_id, status, task_type) VALUES {values} """ ) ) # Run column migration (adds columns to existing table) run_migrations(engine) # raises on failure # Verify data preserved with engine.connect() as conn: count = conn.execute( text("SELECT COUNT(*) FROM task_metadata") ).fetchone()[0] assert count == 1000 # Verify new columns exist and have defaults inspector = inspect(engine) columns = { col["name"] for col in inspector.get_columns("task_metadata") } assert "progress_current" in columns assert "progress_total" in columns finally: engine.dispose() class TestMigrationPerformanceBenchmarks: """Performance benchmark tests for migrations. These tests establish baselines for migration performance to catch any regressions in batch mode operations on large tables. """ @pytest.mark.slow def test_batch_migration_performance_small_table(self, tmp_path): """Batch column migration should complete quickly on small tables (1K rows).""" import time db_path = tmp_path / "perf_small_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Run initial migration run_migrations(engine, target="0001") # Insert 1,000 rows with engine.begin() as conn: for batch in range(10): values = ", ".join( [ f"('small-perf-{batch}-{i}', 'completed', 'research')" for i in range(100) ] ) conn.execute( text( f""" INSERT INTO task_metadata (task_id, status, task_type) VALUES {values} """ ) ) # Time the batch column migration start = time.time() run_migrations(engine) # raises on failure elapsed = time.time() - start # 1K rows should migrate in under 5 seconds assert elapsed < 5, ( f"Small table migration took {elapsed:.2f}s, expected < 5s" ) # Verify data preserved with engine.connect() as conn: count = conn.execute( text("SELECT COUNT(*) FROM task_metadata") ).fetchone()[0] assert count == 1000 finally: engine.dispose() @pytest.mark.slow def test_batch_migration_performance_medium_table(self, tmp_path): """Batch column migration should complete reasonably on medium tables (10K rows).""" import time db_path = tmp_path / "perf_medium_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Run initial migration run_migrations(engine, target="0001") # Insert 10,000 rows with engine.begin() as conn: for batch in range(100): values = ", ".join( [ f"('medium-perf-{batch}-{i}', 'completed', 'research')" for i in range(100) ] ) conn.execute( text( f""" INSERT INTO task_metadata (task_id, status, task_type) VALUES {values} """ ) ) # Verify row count with engine.connect() as conn: count = conn.execute( text("SELECT COUNT(*) FROM task_metadata") ).fetchone()[0] assert count == 10000 # Time the batch column migration start = time.time() run_migrations(engine) # raises on failure elapsed = time.time() - start # 10K rows should migrate in under 30 seconds assert elapsed < 30, ( f"Medium table migration took {elapsed:.2f}s, expected < 30s" ) # Verify data preserved with engine.connect() as conn: count = conn.execute( text("SELECT COUNT(*) FROM task_metadata") ).fetchone()[0] assert count == 10000 # Verify new columns have correct defaults with engine.connect() as conn: sample = conn.execute( text( """ SELECT progress_current, progress_total FROM task_metadata LIMIT 1 """ ) ).fetchone() # New columns should have defaults assert sample[0] == 0 or sample[0] is None assert sample[1] == 0 or sample[1] is None finally: engine.dispose() def test_initial_migration_performance(self, tmp_path): """Initial schema creation should be fast.""" import time db_path = tmp_path / "initial_perf_test.db" engine = create_engine(f"sqlite:///{db_path}") try: start = time.time() run_migrations(engine) # raises on failure elapsed = time.time() - start # Fresh migration should complete in under 10 seconds assert elapsed < 10, ( f"Initial migration took {elapsed:.2f}s, expected < 10s" ) # Verify all tables created inspector = inspect(engine) tables = inspector.get_table_names() assert len(tables) > 20, "Should create many tables" finally: engine.dispose() class TestConcurrentMigrationProcesses: """Tests for concurrent migration scenarios. Tests what happens when multiple processes/threads try to migrate the same database. Note: SQLite has limited concurrency support, so we test with staggered timing and sequential engine usage. """ def test_sequential_engine_migrations(self, tmp_path): """Test sequential migrations from different engine instances. This tests the realistic scenario where different processes/sessions try to migrate the same database at different times. SQLite uses file-level locking which means truly parallel migrations would cause lock contention. """ db_path = tmp_path / "sequential_engine_test.db" for i in range(5): engine = create_engine(f"sqlite:///{db_path}") try: run_migrations(engine) # raises on failure finally: engine.dispose() # Verify database is correctly migrated engine = create_engine(f"sqlite:///{db_path}") try: assert get_current_revision(engine) == get_head_revision() assert not needs_migration(engine) finally: engine.dispose() def test_migration_during_active_reads(self, tmp_path): """Test migration completes correctly while reads are happening.""" import threading import time db_path = tmp_path / "active_reads_test.db" # Create and migrate database first engine = create_engine(f"sqlite:///{db_path}") run_migrations(engine, target="0001") # Insert data with engine.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type) VALUES ('read-test', 'completed', 'research') """ ) ) read_counts = [] read_errors = [] migration_done = threading.Event() def reader_thread(): """Continuously read from database.""" read_engine = create_engine(f"sqlite:///{db_path}") try: while not migration_done.is_set(): with read_engine.connect() as conn: result = conn.execute( text("SELECT COUNT(*) FROM task_metadata") ).fetchone() read_counts.append(result[0]) time.sleep(0.01) except Exception as e: read_errors.append(str(e)) finally: read_engine.dispose() # Start reader thread reader = threading.Thread(target=reader_thread) reader.start() # Run migration while reads are happening time.sleep(0.05) # Let some reads happen first run_migrations(engine) # raises on failure migration_done.set() reader.join() engine.dispose() # Verify results assert len(read_errors) == 0, f"Read errors: {read_errors}" assert len(read_counts) > 0, "Should have completed some reads" def test_no_corruption_on_sequential_migrations(self, tmp_path): """Test that sequential migrations from different engines don't corrupt the database.""" db_path = tmp_path / "no_corruption_test.db" # Run migrations sequentially from different engine instances for i in range(3): engine = create_engine(f"sqlite:///{db_path}") try: run_migrations(engine) # raises on failure finally: engine.dispose() # Verify integrity check_engine = create_engine(f"sqlite:///{db_path}") try: with check_engine.connect() as conn: # Run quick integrity check result = conn.execute(text("PRAGMA quick_check")).fetchone() assert result[0] == "ok" # Verify alembic_version has exactly one row version_count = conn.execute( text("SELECT COUNT(*) FROM alembic_version") ).fetchone()[0] assert version_count == 1 finally: check_engine.dispose() class TestDowngradeVerification: """Tests to verify and document downgrade behavior. These tests document the expected behavior of downgrade migrations, including destructive operations. """ def test_downgrade_0002_removes_progress_columns(self, tmp_path): """Test that downgrading from 0002 removes the progress columns. Starts at 0008 (last fully-reversible) rather than head; 0010 is intentionally non-reversible (SQLite ALTER TABLE limitations). """ from alembic import command db_path = tmp_path / "downgrade_0002_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Upgrade to 0008 (last reversible) config = get_alembic_config(engine) with engine.begin() as conn: config.attributes["connection"] = conn command.upgrade(config, "0008") # Verify progress columns exist inspector = inspect(engine) columns = { col["name"] for col in inspector.get_columns("task_metadata") } assert "progress_current" in columns assert "progress_total" in columns assert "progress_message" in columns assert "metadata_json" in columns # Downgrade to 0001 with engine.begin() as conn: config.attributes["connection"] = conn command.downgrade(config, "0001") # Verify progress columns removed inspector = inspect(engine) columns = { col["name"] for col in inspector.get_columns("task_metadata") } assert "progress_current" not in columns assert "progress_total" not in columns assert "progress_message" not in columns assert "metadata_json" not in columns # Core columns should still exist assert "task_id" in columns assert "status" in columns finally: engine.dispose() def test_downgrade_0001_is_destructive_warning(self, tmp_path): """WARNING: Test documents that downgrade from 0001 to base drops ALL tables. This test verifies the destructive nature of the 0001 downgrade. In production, this should NEVER be run without a backup. Starts at 0008 (last fully-reversible) rather than head; 0010 is intentionally non-reversible (SQLite ALTER TABLE limitations). """ from alembic import command db_path = tmp_path / "destructive_downgrade_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Upgrade to 0008 (last reversible) config = get_alembic_config(engine) with engine.begin() as conn: config.attributes["connection"] = conn command.upgrade(config, "0008") # Count tables before downgrade inspector = inspect(engine) tables_before = set(inspector.get_table_names()) assert len(tables_before) > 10, "Should have many tables" assert "settings" in tables_before assert "research" in tables_before # Downgrade to base (DESTRUCTIVE) with engine.begin() as conn: config.attributes["connection"] = conn command.downgrade(config, "base") # Verify most tables dropped inspector = inspect(engine) tables_after = set(inspector.get_table_names()) # Only alembic_version may remain (or nothing) assert len(tables_after) <= 1, ( f"Expected most tables dropped, got {tables_after}" ) finally: engine.dispose() def test_downgrade_preserves_data_in_kept_columns(self, tmp_path): """Test that downgrade preserves data in columns that are kept. Starts at 0008 (last fully-reversible) rather than head; 0010 is intentionally non-reversible (SQLite ALTER TABLE limitations). """ from alembic import command db_path = tmp_path / "downgrade_preserve_data_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Upgrade to 0008 (last reversible) config = get_alembic_config(engine) with engine.begin() as conn: config.attributes["connection"] = conn command.upgrade(config, "0008") # Insert data using both old and new columns with engine.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type, priority, progress_current, progress_total) VALUES ('downgrade-data-test', 'completed', 'research', 42, 10, 100) """ ) ) # Downgrade to 0001 (removes progress columns but keeps core columns) with engine.begin() as conn: config.attributes["connection"] = conn command.downgrade(config, "0001") # Verify core data preserved with engine.connect() as conn: result = conn.execute( text( """ SELECT task_id, status, task_type, priority FROM task_metadata WHERE task_id = 'downgrade-data-test' """ ) ).fetchone() assert result is not None assert result[0] == "downgrade-data-test" assert result[1] == "completed" assert result[2] == "research" assert result[3] == 42 # Priority preserved finally: engine.dispose() def test_upgrade_after_downgrade_restores_columns(self, tmp_path): """Test that upgrading after downgrade correctly restores columns. Starts at 0008 (last fully-reversible) rather than head; 0010 is intentionally non-reversible (SQLite ALTER TABLE limitations). Re-upgrade to head still verifies forward path through 0010. """ from alembic import command db_path = tmp_path / "upgrade_after_downgrade_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Upgrade to 0008 (last reversible) config = get_alembic_config(engine) with engine.begin() as conn: config.attributes["connection"] = conn command.upgrade(config, "0008") # Insert data with engine.begin() as conn: conn.execute( text( """ INSERT INTO task_metadata (task_id, status, task_type) VALUES ('roundtrip-test', 'completed', 'research') """ ) ) # Downgrade to 0001 with engine.begin() as conn: config.attributes["connection"] = conn command.downgrade(config, "0001") # Upgrade back to head (passes through 0010 in forward direction) run_migrations(engine) # Verify columns restored inspector = inspect(engine) columns = { col["name"] for col in inspector.get_columns("task_metadata") } assert "progress_current" in columns assert "progress_total" in columns # Verify data preserved with engine.connect() as conn: result = conn.execute( text( "SELECT task_id, status FROM task_metadata WHERE task_id = 'roundtrip-test'" ) ).fetchone() assert result is not None assert result[0] == "roundtrip-test" finally: engine.dispose() # ============================================================================= # Security Tests # ============================================================================= class TestDatabaseFileSecurity: """Tests for database file security properties.""" @pytest.mark.skipif( os.name == "nt", reason="File permissions work differently on Windows" ) def test_database_file_not_world_readable(self, tmp_path): """Database files should not be world-readable after creation.""" db_path = tmp_path / "permission_test.db" engine = create_engine(f"sqlite:///{db_path}") try: run_migrations(engine) # Check file permissions - verify file exists and is readable # Note: SQLite doesn't set restrictive permissions by default, # but this test documents the current behavior and verifies umask is applied assert db_path.exists(), "Database file should exist" file_mode = os.stat(db_path).st_mode # The file should have been created with current umask # This assertion documents that permissions are set assert file_mode & 0o600, "Database should be readable by owner" finally: engine.dispose() def test_encrypted_db_unreadable_without_password(self, tmp_path): """Verify encrypted DB cannot be accessed without correct password.""" pytest.importorskip("sqlcipher3") from local_deep_research.database.sqlcipher_utils import ( create_sqlcipher_connection, set_sqlcipher_key, apply_sqlcipher_pragmas, ) from local_deep_research.database.sqlcipher_compat import ( get_sqlcipher_module, ) db_path = tmp_path / "encrypted_test.db" password = "correct_password_123" # Create encrypted database conn = create_sqlcipher_connection(str(db_path), password) cursor = conn.cursor() cursor.execute( "CREATE TABLE test_data (id INTEGER PRIMARY KEY, value TEXT)" ) cursor.execute("INSERT INTO test_data (value) VALUES ('secret')") conn.commit() conn.close() # Try to open with wrong password sqlcipher3 = get_sqlcipher_module() wrong_conn = sqlcipher3.connect(str(db_path)) wrong_cursor = wrong_conn.cursor() set_sqlcipher_key(wrong_cursor, "wrong_password") apply_sqlcipher_pragmas(wrong_cursor, creation_mode=False) with pytest.raises(Exception): # Should fail to read data with wrong password wrong_cursor.execute("SELECT * FROM test_data") wrong_cursor.fetchall() wrong_conn.close() def test_encrypted_db_readable_with_correct_password(self, tmp_path): """Verify encrypted DB can be accessed with correct password.""" pytest.importorskip("sqlcipher3") from local_deep_research.database.sqlcipher_utils import ( create_sqlcipher_connection, ) db_path = tmp_path / "encrypted_correct_test.db" password = "test_password_456" # Create encrypted database with test data conn = create_sqlcipher_connection(str(db_path), password) cursor = conn.cursor() cursor.execute( "CREATE TABLE test_data (id INTEGER PRIMARY KEY, value TEXT)" ) cursor.execute("INSERT INTO test_data (value) VALUES ('accessible')") conn.commit() conn.close() # Reopen with correct password conn2 = create_sqlcipher_connection(str(db_path), password) cursor2 = conn2.cursor() cursor2.execute("SELECT value FROM test_data") result = cursor2.fetchone() conn2.close() assert result is not None assert result[0] == "accessible" class TestErrorMessageSanitization: """Tests for error message sanitization to prevent information leakage.""" def test_migration_error_does_not_expose_full_path(self, tmp_path): """Error messages should not expose full filesystem paths.""" # Test with a path that would be sensitive if exposed sensitive_path = tmp_path / "sensitive_user_data" / "private.db" sensitive_path.parent.mkdir(parents=True, exist_ok=True) # Create a read-only directory scenario engine = create_engine(f"sqlite:///{sensitive_path}") try: # This test verifies that if errors occur, they don't leak paths # in a way that could be exploited run_migrations(engine) # If we get here, migration worked - verify database exists assert sensitive_path.exists() finally: engine.dispose() def test_invalid_revision_error_is_sanitized(self, tmp_path): """Invalid revision errors should not expose internal details.""" db_path = tmp_path / "revision_error_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # First run valid migrations run_migrations(engine) # Try to migrate to a non-existent revision # This should fail gracefully without exposing internals try: run_migrations(engine, target="non_existent_revision_xyz") except Exception as e: error_msg = str(e) # Error message should not contain sensitive paths or details # that could be used for reconnaissance assert "/home/" not in error_msg or "non_existent" in error_msg finally: engine.dispose() def test_connection_error_no_credential_exposure(self): """Connection errors should not expose credentials in messages.""" # Create an engine with invalid credentials in URL # (SQLite doesn't use credentials, but the pattern is important) # Test that error handling doesn't echo back sensitive info try: # This should fail but not expose the path in a dangerous way engine = create_engine("sqlite:///nonexistent_path/db.sqlite") with engine.connect() as conn: conn.execute(text("SELECT 1")) except Exception as e: error_msg = str(e).lower() # Should not contain common credential patterns assert "password" not in error_msg assert "secret" not in error_msg assert "key=" not in error_msg def test_alembic_runner_logs_sanitized(self, tmp_path, caplog): """Verify that alembic runner logs don't expose sensitive data.""" db_path = tmp_path / "log_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # DEBUG level value is 10 in the standard hierarchy with caplog.at_level(10): run_migrations(engine) # Check log messages don't contain sensitive patterns for record in caplog.records: msg = record.getMessage().lower() # Should not log passwords or keys assert ( "password" not in msg or "wrong" in msg ) # test messages ok assert "secret" not in msg assert "credential" not in msg finally: engine.dispose() class TestCryptographicConsistency: """Tests for cryptographic consistency in SQLCipher operations.""" def test_rekey_uses_pbkdf2_like_set_key(self, tmp_path): """Verify rekey uses same PBKDF2 derivation as set_key for consistency.""" pytest.importorskip("sqlcipher3") from local_deep_research.database.sqlcipher_utils import ( create_sqlcipher_connection, set_sqlcipher_rekey, set_sqlcipher_key, apply_sqlcipher_pragmas, ) from local_deep_research.database.sqlcipher_compat import ( get_sqlcipher_module, ) db_path = tmp_path / "rekey_consistency_test.db" original_password = "original_password_123" new_password = "new_password_456" # Create encrypted database with original password conn = create_sqlcipher_connection(str(db_path), original_password) cursor = conn.cursor() cursor.execute( "CREATE TABLE test_data (id INTEGER PRIMARY KEY, value TEXT)" ) cursor.execute("INSERT INTO test_data (value) VALUES ('secret_data')") conn.commit() # Rekey to new password set_sqlcipher_rekey(cursor, new_password) conn.commit() conn.close() # Verify we can open with new password using set_sqlcipher_key # This proves rekey used the same PBKDF2 derivation sqlcipher3 = get_sqlcipher_module() conn2 = sqlcipher3.connect(str(db_path)) cursor2 = conn2.cursor() set_sqlcipher_key(cursor2, new_password) apply_sqlcipher_pragmas(cursor2, creation_mode=False) # Should be able to read data with new password cursor2.execute("SELECT value FROM test_data") result = cursor2.fetchone() conn2.close() assert result is not None assert result[0] == "secret_data" def test_rekey_derived_key_matches_set_key_derived_key(self): """Verify _get_key_from_password produces consistent results.""" from local_deep_research.database.sqlcipher_utils import ( _get_key_from_password, ) password = "test_password_xyz" salt = os.urandom(16) kdf_iterations = 256000 # Call twice to ensure consistency (cached result should match) key1 = _get_key_from_password(password, salt, kdf_iterations) key2 = _get_key_from_password(password, salt, kdf_iterations) assert key1 == key2 assert len(key1) > 0 # Verify it's a bytes object (PBKDF2 output) assert isinstance(key1, bytes) def _insert_witness_then_fail(config, target): """side_effect for patched ``command.upgrade``. Inserts a witness row through the transaction's connection, then raises. If ``engine.begin()``'s rollback works, the row must not be present after the raise — this is what makes the rollback tests non-tautological. A plain ``side_effect=RuntimeError(...)`` fires before any DB write, so the post-failure "revision preserved" assertion would be trivially true. Uses DML (INSERT) rather than DDL (CREATE TABLE) because pysqlite does not reliably roll back DDL statements on default SQLite. The caller must pre-create a ``rollback_witness`` table before patching ``command.upgrade`` with this side_effect. """ conn = config.attributes["connection"] conn.execute(text("INSERT INTO rollback_witness (x) VALUES (1)")) raise RuntimeError("simulated failure") class TestMigrationErrorSanitization: """Tests for migration error message sanitization.""" def test_migration_error_logs_type_not_full_path(self, tmp_path, caplog): """Verify migration errors log exception type, not full paths.""" # Create a scenario that will cause a migration error db_path = tmp_path / "error_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Run migrations first run_migrations(engine) # Now corrupt the alembic_version to cause an error with engine.begin() as conn: conn.execute(text("DELETE FROM alembic_version")) conn.execute( text( "INSERT INTO alembic_version (version_num) VALUES ('invalid')" ) ) # Clear logs and try to migrate again caplog.clear() # This should raise due to the invalid version with caplog.at_level(10): # DEBUG level with pytest.raises(Exception): run_migrations(engine) # Check that any error logs don't contain sensitive home paths for record in caplog.records: msg = record.getMessage() # Should not contain home directory paths in error messages if "error" in msg.lower() or "failed" in msg.lower(): # Error messages should be sanitized assert "/home/" not in msg or "Migration" in msg finally: engine.dispose() def test_migration_raises_on_failure_and_preserves_revision(self, tmp_path): """Failed migration must raise, leave DB at previous revision, AND roll back any writes made inside the transaction. Uses a fresh DB so the short-circuit for already-at-head does not bypass the patched failure. A fresh DB has current revision None, which falls through to command.upgrade(). The patched side_effect inserts a witness row before raising, so the rollback assertion is a real rollback proof — not a tautology. """ db_path = tmp_path / "failure_test.db" engine = create_engine(f"sqlite:///{db_path}") try: # Pre-create the witness table outside the transaction the # patched call will run in. DML (INSERT) rollback is # reliable on default SQLite; DDL (CREATE TABLE) is not. with engine.begin() as conn: conn.execute(text("CREATE TABLE rollback_witness (x INTEGER)")) rev_before = get_current_revision(engine) assert rev_before is None with patch( "local_deep_research.database.alembic_runner.command.upgrade", side_effect=_insert_witness_then_fail, ): with pytest.raises(RuntimeError, match="simulated failure"): run_migrations(engine) assert get_current_revision(engine) == rev_before with engine.begin() as conn: rows = list( conn.execute(text("SELECT x FROM rollback_witness")) ) assert rows == [] finally: engine.dispose() def test_migration_failure_preserves_concrete_prior_revision( self, tmp_path ): """Failed upgrade from a concrete prior revision must roll back cleanly — the DB must still report that concrete revision, and any write made inside the transaction must be undone. Passes an explicit revision target (not "head") so the at-head short-circuit does not skip the patched upgrade call. The patched side_effect inserts a witness row before raising, so the rollback assertion actually exercises rollback. """ db_path = tmp_path / "failure_concrete_test.db" engine = create_engine(f"sqlite:///{db_path}") try: run_migrations(engine) rev_before = get_current_revision(engine) assert rev_before is not None # Pre-create the witness table outside the transaction the # patched call will run in. with engine.begin() as conn: conn.execute(text("CREATE TABLE rollback_witness (x INTEGER)")) with patch( "local_deep_research.database.alembic_runner.command.upgrade", side_effect=_insert_witness_then_fail, ): with pytest.raises(RuntimeError, match="simulated failure"): run_migrations(engine, target=rev_before) assert get_current_revision(engine) == rev_before with engine.begin() as conn: rows = list( conn.execute(text("SELECT x FROM rollback_witness")) ) assert rows == [] finally: engine.dispose() class TestMigrationPathSecurity: """Tests for migration directory path security.""" def test_migrations_dir_within_package(self): """Verify migrations directory is within expected package path.""" migrations_dir = get_migrations_dir() real_path = migrations_dir.resolve() # Should be within the database package directory assert "local_deep_research" in str(real_path) assert "database" in str(real_path) assert "migrations" in str(real_path) def test_migrations_dir_is_not_symlink_by_default(self): """Verify the migrations directory is not a symlink by default.""" migrations_dir = get_migrations_dir() # The directory itself should not be a symlink # (this documents expected behavior, actual symlink detection is in the function) assert migrations_dir.exists() @pytest.mark.skipif( os.name == "nt", reason="Symlink creation requires admin on Windows" ) def test_symlink_attack_detection(self, tmp_path): """Verify symlink attacks are detected (mocked test).""" from pathlib import Path # Create a mock symlinked path scenario malicious_target = tmp_path / "malicious_migrations" malicious_target.mkdir() def mock_get_migrations_dir(): """ Simulates what get_migrations_dir() does but with a path that resolves outside the expected package boundary. """ # The actual function should detect this real_path = malicious_target expected_parent = Path(__file__).parent.resolve() if not str(real_path).startswith(str(expected_parent)): raise ValueError( "Invalid migrations path (possible symlink attack)" ) return malicious_target # Verify that a simulated attack would be caught with pytest.raises(ValueError) as exc_info: mock_get_migrations_dir() assert "symlink attack" in str(exc_info.value) class TestMigrationFilePermissions: """Tests for migration file permission validation.""" @pytest.mark.skipif( os.name == "nt", reason="File permissions work differently on Windows" ) def test_migration_files_not_world_writable(self): """Verify actual migration files are not world-writable.""" migrations_dir = get_migrations_dir() versions_dir = migrations_dir / "versions" if not versions_dir.exists(): pytest.skip("No versions directory exists yet") for migration_file in versions_dir.glob("*.py"): st = migration_file.stat() # Check world-writable bit is NOT set assert not (st.st_mode & 0o002), ( f"Migration file {migration_file.name} is world-writable" ) @pytest.mark.skipif( os.name == "nt", reason="File permissions work differently on Windows" ) def test_world_writable_migration_rejected(self, tmp_path): """Verify world-writable migration files are rejected.""" from local_deep_research.database.alembic_runner import ( _validate_migrations_permissions, ) # Create a mock migrations directory structure mock_migrations = tmp_path / "migrations" mock_versions = mock_migrations / "versions" mock_versions.mkdir(parents=True) # Create a world-writable migration file bad_migration = mock_versions / "001_bad_migration.py" bad_migration.write_text("# malicious content") os.chmod(bad_migration, 0o666) # noqa: S103 — intentionally testing permission validation # Should raise ValueError with pytest.raises(ValueError) as exc_info: _validate_migrations_permissions(mock_migrations) assert "world-writable" in str(exc_info.value) assert "001_bad_migration.py" in str(exc_info.value) @pytest.mark.skipif( os.name == "nt", reason="File permissions work differently on Windows" ) def test_secure_migration_permissions_accepted(self, tmp_path): """Verify migrations with secure permissions pass validation.""" from local_deep_research.database.alembic_runner import ( _validate_migrations_permissions, ) # Create a mock migrations directory with secure files mock_migrations = tmp_path / "migrations" mock_versions = mock_migrations / "versions" mock_versions.mkdir(parents=True) # Create secure migration files good_migration = mock_versions / "001_good_migration.py" good_migration.write_text("# secure content") os.chmod( good_migration, 0o644 ) # Owner rw, group r, other r (not writable) # Should not raise _validate_migrations_permissions(mock_migrations) def test_permission_check_skipped_on_windows(self, tmp_path, monkeypatch): """Verify permission checks are skipped on Windows.""" from local_deep_research.database.alembic_runner import ( _validate_migrations_permissions, ) # Mock os.name to be 'nt' (Windows) monkeypatch.setattr(os, "name", "nt") # Create a mock migrations directory (permissions don't matter) mock_migrations = tmp_path / "migrations" mock_versions = mock_migrations / "versions" mock_versions.mkdir(parents=True) bad_migration = mock_versions / "001_migration.py" bad_migration.write_text("# content") # Should not raise even if file would be "insecure" on Unix _validate_migrations_permissions(mock_migrations) @pytest.mark.skipif( os.name == "nt", reason="File permissions work differently on Windows" ) def test_world_writable_versions_dir_rejected(self, tmp_path): """Reject a world-writable versions/ directory even when files inside are secure. Regression: a Docker image landed with mode 0o777 on the directory itself (files were 0o644). Every per-user login then tripped this check, swallowed the resulting ValueError, and silently left the DB at its previous Alembic revision — surfacing later as ``no such table: papers`` on academic-source saves. """ from local_deep_research.database.alembic_runner import ( _validate_migrations_permissions, ) mock_migrations = tmp_path / "migrations" mock_versions = mock_migrations / "versions" mock_versions.mkdir(parents=True) # File itself is fine — the only insecure thing is the directory. secure_migration = mock_versions / "001_migration.py" secure_migration.write_text("# content") os.chmod(secure_migration, 0o644) os.chmod(mock_versions, 0o777) # noqa: S103 — intentionally testing rejection with pytest.raises(ValueError) as exc_info: _validate_migrations_permissions(mock_migrations) msg = str(exc_info.value) assert "world-writable" in msg assert "versions" in msg @pytest.mark.skipif( os.name == "nt", reason="File permissions work differently on Windows" ) def test_shipped_versions_dir_not_world_writable(self): """The actual shipped versions/ directory must not be world-writable. Companion to ``test_migration_files_not_world_writable`` (which only covers the files). The runtime check rejects either; both need to be guarded against packaging regressions. """ migrations_dir = get_migrations_dir() versions_dir = migrations_dir / "versions" if not versions_dir.exists(): pytest.skip("No versions directory exists yet") st = versions_dir.stat() assert not (st.st_mode & 0o002), ( f"versions/ directory is world-writable (mode={oct(st.st_mode)}). " "If this fires in CI after a packaging change, normalise the " "perms in Dockerfile/build, not by weakening this test." ) # ============================================================================= # Backward Compatibility Tests # ============================================================================= class TestPreAlembicDatabaseUpgrade: """ Simulate real-world upgrade scenarios where databases were created with the old Base.metadata.create_all() + _run_migrations() approach and must be seamlessly upgraded to Alembic-managed schema. """ @pytest.fixture def pre_alembic_db_full(self, tmp_path): """ Simulate a pre-Alembic database: all tables created via Base.metadata.create_all(), progress columns already added by the old _run_migrations() path. No alembic_version table. """ db_path = tmp_path / "pre_alembic_full.db" engine = create_engine(f"sqlite:///{db_path}") # This is exactly what the old initialize_database() did Base.metadata.create_all(engine, checkfirst=True) yield engine engine.dispose() @pytest.fixture def pre_alembic_db_old(self, tmp_path): """ Simulate a very old pre-Alembic database: all tables created via Base.metadata.create_all(), but progress columns NOT yet added (simulates a database from before _run_migrations existed). """ db_path = tmp_path / "pre_alembic_old.db" engine = create_engine(f"sqlite:///{db_path}") Base.metadata.create_all(engine, checkfirst=True) # Remove progress columns to simulate old schema with engine.begin() as conn: inspector = inspect(conn) if inspector.has_table("task_metadata"): cols = { c["name"] for c in inspector.get_columns("task_metadata") } for col in [ "progress_current", "progress_total", "progress_message", "metadata_json", ]: if col in cols: # SQLite doesn't support DROP COLUMN before 3.35, # so recreate the table without those columns pass # Simpler approach: create task_metadata manually without progress cols with engine.begin() as conn: conn.execute(text("DROP TABLE IF EXISTS task_metadata")) conn.execute( text(""" CREATE TABLE task_metadata ( id INTEGER PRIMARY KEY, task_id VARCHAR, status VARCHAR, task_type VARCHAR, created_at DATETIME, completed_at DATETIME, duration_seconds FLOAT, error_message VARCHAR ) """) ) yield engine engine.dispose() @pytest.fixture def pre_alembic_db_with_data(self, tmp_path): """ Simulate a pre-Alembic database with realistic user data that must survive the migration. """ db_path = tmp_path / "pre_alembic_data.db" engine = create_engine(f"sqlite:///{db_path}") Base.metadata.create_all(engine, checkfirst=True) # Insert realistic data using ORM (handles all NOT NULL defaults) from local_deep_research.database.models import Setting from local_deep_research.database.models.settings import SettingType Session = sessionmaker(bind=engine) with Session() as session: session.add( Setting( key="llm.provider", value="ollama", type=SettingType.LLM, name="LLM Provider", category="llm", ) ) session.add( Setting( key="search.engine", value="duckduckgo", type=SettingType.SEARCH, name="Search Engine", category="search", ) ) session.commit() # Use raw SQL for simpler tables without complex defaults with engine.begin() as conn: # Research history conn.execute( text(""" INSERT INTO research_history (id, query, mode, status, created_at, research_meta) VALUES ('res-001', 'quantum computing', 'deep', 'completed', datetime('now'), '{"iterations": 5}'), ('res-002', 'machine learning', 'quick', 'completed', datetime('now'), '{"iterations": 1}') """) ) # Task metadata with progress columns (old _run_migrations added these) conn.execute( text(""" INSERT INTO task_metadata (task_id, status, task_type, progress_current, progress_total, progress_message) VALUES ('task-001', 'completed', 'research', 5, 5, 'Done'), ('task-002', 'running', 'research', 2, 10, 'Iteration 2/10') """) ) yield engine engine.dispose() def test_full_pre_alembic_db_upgrades_cleanly(self, pre_alembic_db_full): """A complete pre-Alembic database upgrades without errors.""" engine = pre_alembic_db_full # Verify no alembic_version table yet inspector = inspect(engine) assert "alembic_version" not in inspector.get_table_names() # Run the new initialize_database (which calls run_migrations) initialize_database(engine) # Verify Alembic now tracks the database new_inspector = inspect(engine) assert "alembic_version" in new_inspector.get_table_names() # Verify at head revision assert get_current_revision(engine) == get_head_revision() assert not needs_migration(engine) def test_old_db_missing_progress_cols_gets_them_added( self, pre_alembic_db_old ): """ A very old database missing progress columns gets them added via Alembic migration 0002. """ engine = pre_alembic_db_old # Verify columns are missing inspector = inspect(engine) cols = {c["name"] for c in inspector.get_columns("task_metadata")} assert "progress_current" not in cols # Run migrations run_migrations(engine) # raises on failure # Verify progress columns now exist new_inspector = inspect(engine) new_cols = { c["name"] for c in new_inspector.get_columns("task_metadata") } assert "progress_current" in new_cols assert "progress_total" in new_cols assert "progress_message" in new_cols assert "metadata_json" in new_cols def test_data_survives_full_upgrade_path(self, pre_alembic_db_with_data): """ All user data (settings, research history, task metadata) survives the upgrade from pre-Alembic to Alembic-managed. """ engine = pre_alembic_db_with_data # Run the full upgrade initialize_database(engine) # Verify settings preserved with engine.connect() as conn: settings = conn.execute( text("SELECT key, value FROM settings ORDER BY key") ).fetchall() settings_dict = {row[0]: row[1] for row in settings} assert settings_dict["llm.provider"] in ("ollama", '"ollama"') assert settings_dict["search.engine"] in ( "duckduckgo", '"duckduckgo"', ) # Verify research history preserved with engine.connect() as conn: research = conn.execute( text( "SELECT id, query, status FROM research_history ORDER BY id" ) ).fetchall() assert len(research) == 2 assert research[0][1] == "quantum computing" assert research[1][1] == "machine learning" # Verify task metadata preserved (including progress columns) with engine.connect() as conn: tasks = conn.execute( text(""" SELECT task_id, status, progress_current, progress_total, progress_message FROM task_metadata ORDER BY task_id """) ).fetchall() assert len(tasks) == 2 assert tasks[0][0] == "task-001" assert tasks[0][2] == 5 # progress_current assert tasks[0][3] == 5 # progress_total assert tasks[0][4] == "Done" assert tasks[1][0] == "task-002" assert tasks[1][2] == 2 assert tasks[1][4] == "Iteration 2/10" def test_upgrade_then_reinitialize_is_idempotent( self, pre_alembic_db_with_data ): """ After upgrading a pre-Alembic DB, calling initialize_database again should be a no-op (idempotent). """ engine = pre_alembic_db_with_data # First upgrade initialize_database(engine) rev1 = get_current_revision(engine) inspector1 = inspect(engine) tables1 = set(inspector1.get_table_names()) # Count data with engine.connect() as conn: settings_count = conn.execute( text("SELECT COUNT(*) FROM settings") ).fetchone()[0] research_count = conn.execute( text("SELECT COUNT(*) FROM research_history") ).fetchone()[0] # Second initialization (should be no-op) initialize_database(engine) rev2 = get_current_revision(engine) inspector2 = inspect(engine) tables2 = set(inspector2.get_table_names()) # Verify nothing changed assert rev1 == rev2 assert tables1 == tables2 with engine.connect() as conn: assert ( conn.execute(text("SELECT COUNT(*) FROM settings")).fetchone()[ 0 ] == settings_count ) assert ( conn.execute( text("SELECT COUNT(*) FROM research_history") ).fetchone()[0] == research_count ) def test_schema_matches_after_upgrade_vs_fresh(self, tmp_path): """ The schema of an upgraded pre-Alembic database should match the schema of a freshly created Alembic database (minus the users table which pre-Alembic databases had). """ # Create a fresh Alembic database fresh_path = tmp_path / "fresh.db" fresh_engine = create_engine(f"sqlite:///{fresh_path}") run_migrations(fresh_engine) # Create a pre-Alembic database and upgrade it legacy_path = tmp_path / "legacy.db" legacy_engine = create_engine(f"sqlite:///{legacy_path}") Base.metadata.create_all(legacy_engine, checkfirst=True) run_migrations(legacy_engine) try: fresh_inspector = inspect(fresh_engine) legacy_inspector = inspect(legacy_engine) fresh_tables = set(fresh_inspector.get_table_names()) legacy_tables = set(legacy_inspector.get_table_names()) # Legacy DB may have 'users' table (pre-Alembic created all tables) # Fresh Alembic DB skips 'users' (auth-only) legacy_tables.discard("users") assert fresh_tables == legacy_tables, ( f"Table mismatch.\n" f" Only in fresh: {fresh_tables - legacy_tables}\n" f" Only in legacy: {legacy_tables - fresh_tables}" ) # Compare columns for each shared table (excluding alembic_version) for table_name in fresh_tables - {"alembic_version"}: fresh_cols = { c["name"] for c in fresh_inspector.get_columns(table_name) } legacy_cols = { c["name"] for c in legacy_inspector.get_columns(table_name) } assert fresh_cols == legacy_cols, ( f"Column mismatch in table '{table_name}'.\n" f" Only in fresh: {fresh_cols - legacy_cols}\n" f" Only in legacy: {legacy_cols - fresh_cols}" ) finally: fresh_engine.dispose() legacy_engine.dispose() def test_users_table_not_in_fresh_alembic_db(self, tmp_path): """ Verify that a fresh Alembic-managed database does NOT contain the 'users' table (it's auth-only and created separately). """ db_path = tmp_path / "fresh_no_users.db" engine = create_engine(f"sqlite:///{db_path}") try: run_migrations(engine) inspector = inspect(engine) tables = inspector.get_table_names() assert "users" not in tables finally: engine.dispose() def test_pre_alembic_db_with_users_table_keeps_it(self, tmp_path): """ A pre-Alembic database that has a 'users' table (from the old create_all) should keep it after migration - Alembic doesn't drop tables it doesn't manage. """ db_path = tmp_path / "legacy_with_users.db" engine = create_engine(f"sqlite:///{db_path}") try: # Old behavior: create_all creates everything including users Base.metadata.create_all(engine, checkfirst=True) inspector = inspect(engine) assert "users" in inspector.get_table_names() # Upgrade with Alembic run_migrations(engine) # users table should still be there (not dropped) new_inspector = inspect(engine) assert "users" in new_inspector.get_table_names() finally: engine.dispose() def test_multiple_sequential_upgrades(self, tmp_path): """ Simulate the lifecycle: create DB → upgrade to 0001 → insert data → upgrade to 0002 → verify data + new columns. """ db_path = tmp_path / "sequential.db" engine = create_engine(f"sqlite:///{db_path}") try: # Step 1: Migrate to 0001 only run_migrations(engine, target="0001") # raises on failure assert get_current_revision(engine) == "0001" # Step 2: Insert data with engine.begin() as conn: conn.execute( text(""" INSERT INTO task_metadata (task_id, status, task_type) VALUES ('seq-001', 'running', 'research'), ('seq-002', 'completed', 'research') """) ) # Use ORM for settings (many NOT NULL columns with Python defaults) from local_deep_research.database.models import Setting from local_deep_research.database.models.settings import SettingType Session = sessionmaker(bind=engine) with Session() as session: session.add( Setting( key="test.key", value="test_value", type=SettingType.APP, name="Test Key", category="test", ) ) session.commit() # Step 3: Upgrade to head (0005) run_migrations(engine, target="head") # raises on failure assert get_current_revision(engine) == get_head_revision() # Step 4: Verify data survived with engine.connect() as conn: tasks = conn.execute( text("SELECT task_id FROM task_metadata ORDER BY task_id") ).fetchall() assert len(tasks) == 2 assert tasks[0][0] == "seq-001" settings = conn.execute( text("SELECT value FROM settings WHERE key = 'test.key'") ).fetchone() assert settings[0] in ("test_value", '"test_value"') # Step 5: Verify new columns exist # Note: existing rows get NULL for new columns (SQLite batch alter # adds columns but doesn't backfill; server_default only applies # to future inserts) with engine.connect() as conn: row = conn.execute( text(""" SELECT progress_current, progress_total FROM task_metadata WHERE task_id = 'seq-001' """) ).fetchone() assert row is not None # Columns exist and are queryable # Verify new inserts can use the columns with engine.begin() as conn: conn.execute( text(""" INSERT INTO task_metadata (task_id, status, task_type, progress_current, progress_total) VALUES ('seq-003', 'pending', 'research', 0, 5) """) ) with engine.connect() as conn: row = conn.execute( text(""" SELECT progress_current, progress_total FROM task_metadata WHERE task_id = 'seq-003' """) ).fetchone() assert row[0] == 0 assert row[1] == 5 # Step 6: Another run_migrations should be no-op run_migrations(engine) # raises on failure assert get_current_revision(engine) == get_head_revision() finally: engine.dispose() class TestUpgradeFromBuggyV16xUserDatabase: """Regression for #3697. Pre-fix ``encrypted_db.create_user_database`` only emitted ``CreateTable``, never ``CreateIndex``. Combined with v1.6.0 enabling ``PRAGMA foreign_keys = ON``, every existing user DB has a mismatched FK (``download_attempts.url_hash`` → ``download_tracker.url_hash`` with no UNIQUE backing). On any DML touching those tables SQLite raises ``foreign key mismatch``, which would abort migration 0007 itself before it could repair the schema. Migration 0007 must therefore disable FK enforcement around its scrub + UNIQUE-index creation. """ @pytest.fixture def buggy_v16x_engine(self, tmp_path): """Build a DB that mirrors the pre-fix state of an existing user: tables exist, but ``download_tracker.url_hash`` has no UNIQUE backing, the alembic_version is at 0006, and FK enforcement is on for every connection (mirroring ``apply_performance_pragmas``).""" import sqlite3 from sqlalchemy import event db_path = tmp_path / "buggy_v16x.db" raw = sqlite3.connect(db_path) raw.execute("PRAGMA foreign_keys = OFF") raw.executescript( """ CREATE TABLE alembic_version (version_num VARCHAR(32) PRIMARY KEY); INSERT INTO alembic_version VALUES ('0006'); CREATE TABLE download_tracker ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT NOT NULL, url_hash VARCHAR(64) NOT NULL, first_resource_id INTEGER, is_downloaded BOOLEAN, file_hash VARCHAR(64), file_path TEXT, file_name VARCHAR(255), file_size INTEGER, is_accessible BOOLEAN, first_seen TIMESTAMP, downloaded_at TIMESTAMP, last_checked TIMESTAMP, library_document_id INTEGER ); CREATE TABLE download_attempts ( id INTEGER PRIMARY KEY AUTOINCREMENT, url_hash VARCHAR(64) NOT NULL, attempt_number INTEGER, succeeded BOOLEAN, attempted_at TIMESTAMP, FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash) ); CREATE TABLE download_duplicates ( id INTEGER PRIMARY KEY AUTOINCREMENT, url_hash VARCHAR(64) NOT NULL, resource_id INTEGER, research_id VARCHAR(36), added_at TIMESTAMP, FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash) ); CREATE TABLE research_history ( id VARCHAR(36) PRIMARY KEY, query TEXT NOT NULL, mode TEXT NOT NULL, status TEXT NOT NULL, created_at TEXT NOT NULL, completed_at TEXT, duration_seconds INTEGER, report_path TEXT, report_content TEXT, research_meta TEXT, progress_log TEXT, progress INTEGER, title TEXT ); """ ) # Seed: duplicate parent rows (HASH1) + valid child + orphan child. raw.execute( "INSERT INTO download_tracker (id, url, url_hash, first_resource_id) " "VALUES (1, 'a', 'HASH1', 1), (2, 'a-dup', 'HASH1', 2), (3, 'b', 'HASH2', 3)" ) raw.execute( "INSERT INTO download_attempts (url_hash, attempt_number) " "VALUES ('HASH1', 1), ('HASH_ORPHAN', 1)" ) raw.execute( "INSERT INTO download_duplicates (url_hash, resource_id, research_id) " "VALUES ('HASH_ORPHAN', 99, 'r1')" ) raw.commit() raw.close() engine = create_engine(f"sqlite:///{db_path}") @event.listens_for(engine, "connect") def _enable_fk(dbapi_conn, _): dbapi_conn.execute("PRAGMA foreign_keys = ON") yield engine engine.dispose() def test_migration_0007_succeeds_against_pre_fix_user_db( self, buggy_v16x_engine ): """Without the FK toggle in 0007's upgrade(), the duplicate scrub raises ``foreign key mismatch`` and the migration aborts at 0006.""" run_migrations(buggy_v16x_engine) # 0008 rides along on the same chain — assert head, not a literal. assert get_current_revision(buggy_v16x_engine) == get_head_revision() with buggy_v16x_engine.connect() as conn: assert conn.execute(text("PRAGMA foreign_keys")).scalar() == 1 assert ( conn.execute(text("PRAGMA foreign_key_check")).fetchall() == [] ) indexes = { row[1] for row in conn.execute( text("PRAGMA index_list(download_tracker)") ).fetchall() } assert "uq_download_tracker_url_hash" in indexes tracker_ids = sorted( r[0] for r in conn.execute( text("SELECT id FROM download_tracker") ).fetchall() ) # Survivor of HASH1 group is min(id)=1; HASH2 unaffected. assert tracker_ids == [1, 3] attempt_hashes = { r[0] for r in conn.execute( text("SELECT url_hash FROM download_attempts") ).fetchall() } assert attempt_hashes == {"HASH1"} # orphan removed duplicate_count = conn.execute( text("SELECT COUNT(*) FROM download_duplicates") ).scalar() assert duplicate_count == 0 # orphan removed def test_fk_enforcement_active_after_migration(self, buggy_v16x_engine): """Once 0007 finishes, the repaired FK must actually reject inserts with a non-existent ``url_hash`` — proving FK was re-enabled and the UNIQUE backing is recognized.""" from sqlalchemy.exc import IntegrityError run_migrations(buggy_v16x_engine) with buggy_v16x_engine.begin() as conn: conn.execute( text( "INSERT INTO download_attempts (url_hash, attempt_number) " "VALUES ('HASH1', 2)" ) ) with pytest.raises(IntegrityError): with buggy_v16x_engine.begin() as conn: conn.execute( text( "INSERT INTO download_attempts (url_hash, attempt_number) " "VALUES ('NEW_ORPHAN', 1)" ) ) class TestUpgradeFromBuggyV16xUserDbProductionEngine: """Regression for #3990 — multi-migration upgrade fails with FK mismatch. The existing ``TestUpgradeFromBuggyV16xUserDatabase`` covers the same schema corruption but stamps at revision 0006, so migration 0007 is the first to run any DML in the upgrade transaction — which lets 0007's own ``PRAGMA foreign_keys = OFF`` actually take effect (no auto-begun transaction yet). Real production users (issue #3990) upgrade from 0001 to head. Migrations 0002–0006 run DML before 0007, auto-beginning the sqlite3 driver transaction. ``PRAGMA foreign_keys`` is silently a no-op once a transaction is active (per sqlite.org/pragma.html#pragma_foreign_keys), so 0007's defensive PRAGMA never lands and the orphan-scrub DELETE fails with ``foreign key mismatch``. The fix (in ``alembic_runner.run_migrations``) issues PRAGMA OFF *before* opening the migration transaction. This test reproduces the production failure exactly: ``isolation_level=""`` (matching the sqlcipher3 engine in ``encrypted_db.py``) + FK ON at connect via the same event handler ``apply_performance_pragmas`` installs. """ @pytest.fixture def buggy_v16x_production_engine(self, tmp_path): """Mirror the production engine: isolation_level="" + FK ON at connect, with the buggy v1.6.x schema stamped at revision 0005. Stamping at 0005 (not 0006) means migration 0006's data backfill (``UPDATE journals SET name_lower = ...``) runs DML before 0007, auto-beginning the driver transaction and freezing FK in the connect-time ON state for the rest of the upgrade. """ import sqlite3 from sqlalchemy import event db_path = tmp_path / "buggy_v16x_prod.db" # Seed the buggy v1.6.x schema with FK off (raw connection, no # FK target validation needed — the schema deliberately reflects # the pre-fix shape with no UNIQUE backing on download_tracker.url_hash). raw = sqlite3.connect(db_path) raw.execute("PRAGMA foreign_keys = OFF") raw.executescript( """ CREATE TABLE alembic_version (version_num VARCHAR(32) PRIMARY KEY); INSERT INTO alembic_version VALUES ('0005'); CREATE TABLE download_tracker ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT NOT NULL, url_hash VARCHAR(64) NOT NULL, first_resource_id INTEGER, is_downloaded BOOLEAN, file_hash VARCHAR(64), file_path TEXT, file_name VARCHAR(255), file_size INTEGER, is_accessible BOOLEAN, first_seen TIMESTAMP, downloaded_at TIMESTAMP, last_checked TIMESTAMP, library_document_id INTEGER ); CREATE TABLE download_attempts ( id INTEGER PRIMARY KEY AUTOINCREMENT, url_hash VARCHAR(64) NOT NULL, attempt_number INTEGER, succeeded BOOLEAN, attempted_at TIMESTAMP, FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash) ); CREATE TABLE download_duplicates ( id INTEGER PRIMARY KEY AUTOINCREMENT, url_hash VARCHAR(64) NOT NULL, resource_id INTEGER, research_id VARCHAR(36), added_at TIMESTAMP, FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash) ); CREATE TABLE journals ( id INTEGER PRIMARY KEY AUTOINCREMENT, name VARCHAR(500) NOT NULL, quality INTEGER, quality_analysis_time TIMESTAMP ); INSERT INTO journals (name, quality) VALUES ('Nature', 100), ('NATURE', 95); """ ) raw.execute( "INSERT INTO download_tracker (id, url, url_hash, first_resource_id) " "VALUES (1, 'a', 'HASH1', 1), (2, 'a-dup', 'HASH1', 2), (3, 'b', 'HASH2', 3)" ) raw.execute( "INSERT INTO download_attempts (url_hash, attempt_number) " "VALUES ('HASH1', 1), ('HASH_ORPHAN', 1)" ) raw.execute( "INSERT INTO download_duplicates (url_hash, resource_id, research_id) " "VALUES ('HASH_ORPHAN', 99, 'r1')" ) raw.commit() raw.close() # Production-shape engine: deferred isolation_level + FK ON at connect. def _create_conn(): conn = sqlite3.connect( str(db_path), isolation_level="", check_same_thread=False, ) conn.execute("PRAGMA foreign_keys = ON") return conn engine = create_engine("sqlite://", creator=_create_conn) @event.listens_for(engine, "connect") def _enable_fk(dbapi_conn, _): # Defensive: also fire on any pooled re-checkouts that bypass # the creator (matches encrypted_db's apply_performance_pragmas). dbapi_conn.execute("PRAGMA foreign_keys = ON") yield engine engine.dispose() def test_run_migrations_succeeds_through_full_chain( self, buggy_v16x_production_engine ): """Without the runner-level FK toggle, this fails with ``foreign key mismatch`` at 0007's orphan scrub.""" run_migrations(buggy_v16x_production_engine) assert ( get_current_revision(buggy_v16x_production_engine) == get_head_revision() ) with buggy_v16x_production_engine.connect() as conn: # FK is back ON for the next checkout (engine was disposed). assert conn.execute(text("PRAGMA foreign_keys")).scalar() == 1 # No FK violations remain in the repaired DB. assert ( conn.execute(text("PRAGMA foreign_key_check")).fetchall() == [] ) # Orphan rows were scrubbed. attempt_hashes = { r[0] for r in conn.execute( text("SELECT url_hash FROM download_attempts") ).fetchall() } assert attempt_hashes == {"HASH1"} duplicate_count = conn.execute( text("SELECT COUNT(*) FROM download_duplicates") ).scalar() assert duplicate_count == 0 class TestOrphanAlembicTempTableCleanup: """Regression for #3817 — ``table _alembic_tmp_journals already exists``. ``op.batch_alter_table`` rebuilds a table by creating ``_alembic_tmp_``, copying data, dropping the original, and renaming. On a clean run alembic drops the temp table automatically. If a previous attempt failed in a way that bypassed transaction rollback (e.g., an older runner that auto-committed each migration), the temp table persists. The next ``batch_alter_table`` on the same parent fails with ``table _alembic_tmp_* already exists`` — even if the broader transaction would roll it back, alembic checks for pre-existing temp tables before creating its own. The fix drops orphan ``_alembic_tmp_*`` tables in ``alembic_runner.run_migrations`` before opening the migration transaction. """ @pytest.fixture def db_with_orphan_temp_table(self, tmp_path): """Build a buggy v1.6.x DB at revision 0005 with an orphan ``_alembic_tmp_journals`` table left over from a prior crash.""" import sqlite3 db_path = tmp_path / "with_orphan_tmp.db" raw = sqlite3.connect(db_path) raw.execute("PRAGMA foreign_keys = OFF") raw.executescript( """ CREATE TABLE alembic_version (version_num VARCHAR(32) PRIMARY KEY); INSERT INTO alembic_version VALUES ('0005'); CREATE TABLE download_tracker ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT, url_hash VARCHAR(64) NOT NULL, first_resource_id INTEGER, is_downloaded BOOLEAN, file_hash VARCHAR(64), file_path TEXT, file_name VARCHAR(255), file_size INTEGER, is_accessible BOOLEAN, first_seen TIMESTAMP, downloaded_at TIMESTAMP, last_checked TIMESTAMP, library_document_id INTEGER ); CREATE TABLE download_attempts ( id INTEGER PRIMARY KEY AUTOINCREMENT, url_hash VARCHAR(64) NOT NULL, attempt_number INTEGER, succeeded BOOLEAN, attempted_at TIMESTAMP, FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash) ); CREATE TABLE download_duplicates ( id INTEGER PRIMARY KEY AUTOINCREMENT, url_hash VARCHAR(64) NOT NULL, resource_id INTEGER, research_id VARCHAR(36), added_at TIMESTAMP, FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash) ); CREATE TABLE journals ( id INTEGER PRIMARY KEY AUTOINCREMENT, name VARCHAR(500) NOT NULL, quality INTEGER, quality_analysis_time TIMESTAMP ); -- The smoking gun: a leftover batch_alter_table temp table -- from a prior crashed migration attempt. The schema can be -- arbitrary — alembic only checks the name. CREATE TABLE _alembic_tmp_journals ( id INTEGER PRIMARY KEY, stale_marker TEXT ); """ ) raw.commit() raw.close() def _create_conn(): conn = sqlite3.connect( str(db_path), isolation_level="", check_same_thread=False ) conn.execute("PRAGMA foreign_keys = ON") return conn engine = create_engine("sqlite://", creator=_create_conn) yield engine engine.dispose() def test_run_migrations_drops_orphan_temp_table( self, db_with_orphan_temp_table ): """Without the cleanup, migration 0006's batch_alter_table fails with ``table _alembic_tmp_journals already exists``.""" # Sanity check: the orphan is present before the run. with db_with_orphan_temp_table.connect() as conn: tables = { r[0] for r in conn.execute( text("SELECT name FROM sqlite_master WHERE type='table'") ).fetchall() } assert "_alembic_tmp_journals" in tables run_migrations(db_with_orphan_temp_table) assert ( get_current_revision(db_with_orphan_temp_table) == get_head_revision() ) # The orphan was dropped; alembic's own temp tables (if any from # this run) were also cleaned up by alembic itself. with db_with_orphan_temp_table.connect() as conn: tables_after = { r[0] for r in conn.execute( text("SELECT name FROM sqlite_master WHERE type='table'") ).fetchall() } assert not any(t.startswith("_alembic_tmp_") for t in tables_after) @pytest.fixture def db_with_multiple_orphan_temp_tables(self, tmp_path): """Buggy v1.6.x DB at revision 0005 with THREE orphan ``_alembic_tmp_*`` tables — exercises the loop in ``_drop_orphan_alembic_temp_tables`` past index 0. A regression that replaces the iteration body with a single ``break`` (or otherwise short-circuits) would leave the trailing orphans behind and fail this test, while ``test_run_migrations_drops_orphan_temp_table`` with its single orphan would still pass.""" import sqlite3 db_path = tmp_path / "with_multi_orphans.db" raw = sqlite3.connect(db_path) raw.execute("PRAGMA foreign_keys = OFF") raw.executescript( """ CREATE TABLE alembic_version (version_num VARCHAR(32) PRIMARY KEY); INSERT INTO alembic_version VALUES ('0005'); CREATE TABLE download_tracker ( id INTEGER PRIMARY KEY AUTOINCREMENT, url TEXT, url_hash VARCHAR(64) NOT NULL, first_resource_id INTEGER, is_downloaded BOOLEAN, file_hash VARCHAR(64), file_path TEXT, file_name VARCHAR(255), file_size INTEGER, is_accessible BOOLEAN, first_seen TIMESTAMP, downloaded_at TIMESTAMP, last_checked TIMESTAMP, library_document_id INTEGER ); CREATE TABLE download_attempts ( id INTEGER PRIMARY KEY AUTOINCREMENT, url_hash VARCHAR(64) NOT NULL, attempt_number INTEGER, succeeded BOOLEAN, attempted_at TIMESTAMP, FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash) ); CREATE TABLE download_duplicates ( id INTEGER PRIMARY KEY AUTOINCREMENT, url_hash VARCHAR(64) NOT NULL, resource_id INTEGER, research_id VARCHAR(36), added_at TIMESTAMP, FOREIGN KEY (url_hash) REFERENCES download_tracker(url_hash) ); CREATE TABLE journals ( id INTEGER PRIMARY KEY AUTOINCREMENT, name VARCHAR(500) NOT NULL, quality INTEGER, quality_analysis_time TIMESTAMP ); CREATE TABLE _alembic_tmp_journals ( id INTEGER PRIMARY KEY, stale_marker TEXT ); CREATE TABLE _alembic_tmp_research_history ( id INTEGER PRIMARY KEY, stale_marker TEXT ); CREATE TABLE _alembic_tmp_settings ( id INTEGER PRIMARY KEY, stale_marker TEXT ); """ ) raw.commit() raw.close() def _create_conn(): conn = sqlite3.connect( str(db_path), isolation_level="", check_same_thread=False ) conn.execute("PRAGMA foreign_keys = ON") return conn engine = create_engine("sqlite://", creator=_create_conn) yield engine engine.dispose() def test_run_migrations_drops_multiple_orphan_temp_tables( self, db_with_multiple_orphan_temp_tables ): """The cleanup must process every match — not just the first.""" with db_with_multiple_orphan_temp_tables.connect() as conn: seeded = { r[0] for r in conn.execute( text( "SELECT name FROM sqlite_master " "WHERE type='table' AND name LIKE '_alembic_tmp_%'" ) ).fetchall() } assert seeded == { "_alembic_tmp_journals", "_alembic_tmp_research_history", "_alembic_tmp_settings", } run_migrations(db_with_multiple_orphan_temp_tables) assert ( get_current_revision(db_with_multiple_orphan_temp_tables) == get_head_revision() ) with db_with_multiple_orphan_temp_tables.connect() as conn: remaining = [ r[0] for r in conn.execute( text( "SELECT name FROM sqlite_master " "WHERE type='table' AND name LIKE '_alembic_tmp_%'" ) ).fetchall() ] assert remaining == [] def test_drop_orphan_temp_tables_no_op_when_none_present( self, tmp_path, loguru_caplog ): """Unit test on the cleanup helper itself: when the DB has no ``_alembic_tmp_*`` tables, the function must return without DDL and without emitting the ``Found N orphan alembic temp table(s)`` warning. This pins the early-return guard at the top of ``_drop_orphan_alembic_temp_tables`` so a future refactor that drops the guard (e.g. unconditional logging) would be caught immediately.""" from local_deep_research.database.alembic_runner import ( _drop_orphan_alembic_temp_tables, ) db_path = tmp_path / "clean.db" engine = create_engine(f"sqlite:///{db_path}") try: with engine.connect() as conn: conn.exec_driver_sql( "CREATE TABLE real_table (id INTEGER PRIMARY KEY)" ) with loguru_caplog.at_level("WARNING"): loguru_caplog.clear() _drop_orphan_alembic_temp_tables(conn) assert ( "orphan alembic temp table" not in loguru_caplog.text.lower() ) # The real table is still there — we didn't touch anything. tables = { r[0] for r in conn.execute( text( "SELECT name FROM sqlite_master WHERE type='table'" ) ).fetchall() } assert "real_table" in tables finally: engine.dispose() class TestPreAlembicDatabaseHotfix: """Regression tests for bug #3747 — pre-Alembic DB login failure. Real users whose database was created before v1.4.0 (2026-03-25, when Alembic was introduced) have schema tables but no alembic_version row. Before this hotfix, run_migrations() called command.upgrade() from scratch, exposing legacy column shapes to migration 0007's index backfill and `download_tracker` scrub. The fixture uses raw-SQL legacy schema (NOT Base.metadata.create_all) so it reflects what a real pre-2026-03-21 user database actually looks like: `settings` omits the modern `category` column, `download_tracker` has the legacy shape without a UNIQUE constraint on `url_hash`, etc. """ @pytest.fixture def pre_alembic_engine(self, tmp_path): engine = create_engine(f"sqlite:///{tmp_path}/pre_alembic.db") with engine.begin() as conn: # Legacy `settings` shape (no `category`, no `description`, # no `ui_element`, no enum extras — just the bare columns # that existed at project inception, 2025-06-29). conn.execute( text( """ CREATE TABLE settings ( id INTEGER PRIMARY KEY, key VARCHAR(255) NOT NULL UNIQUE, value JSON, type VARCHAR(50) NOT NULL, name VARCHAR(255) NOT NULL, visible BOOLEAN DEFAULT 1 NOT NULL, editable BOOLEAN DEFAULT 1 NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ) """ ) ) conn.execute( text( """ CREATE TABLE research_history ( id VARCHAR(36) PRIMARY KEY, query TEXT NOT NULL, mode VARCHAR(50), status VARCHAR(50), created_at DATETIME ) """ ) ) # Legacy `download_tracker` shape: url_hash without the # UNIQUE backing that migration 0007 expects to find. This # is the table that the unfixed migration path actually # trips over via 0007's index backfill / orphan scrub. conn.execute( text( """ CREATE TABLE download_tracker ( id INTEGER PRIMARY KEY, url TEXT NOT NULL, url_hash VARCHAR(64) NOT NULL, first_resource_id INTEGER, is_downloaded BOOLEAN DEFAULT 0 ) """ ) ) # Sanity: fixture really is pre-Alembic-shaped. inspector = inspect(engine) assert "alembic_version" not in inspector.get_table_names() assert "category" not in { c["name"] for c in inspector.get_columns("settings") } yield engine engine.dispose() def test_pre_alembic_db_reaches_head_and_stamp_branch_engaged( self, pre_alembic_engine, loguru_caplog ): """Core regression: pre-Alembic DB → run_migrations() → head, AND the BUG-3747 stamp branch is what got it there (not a coincidence).""" with loguru_caplog.at_level("WARNING"): run_migrations(pre_alembic_engine) assert get_current_revision(pre_alembic_engine) == get_head_revision() assert ( "alembic_version" in inspect(pre_alembic_engine).get_table_names() ) # Without the hotfix this log line never appears — its presence # proves we got to head via the stamp path, not by accident. assert "BUG-3747: pre-Alembic database detected" in loguru_caplog.text def test_pre_alembic_migration_is_idempotent(self, pre_alembic_engine): """Re-running run_migrations() after the hotfix is a no-op.""" run_migrations(pre_alembic_engine) rev1 = get_current_revision(pre_alembic_engine) run_migrations(pre_alembic_engine) assert get_current_revision(pre_alembic_engine) == rev1 def test_fresh_db_does_not_enter_stamp_branch( self, tmp_path, loguru_caplog ): """An empty DB must run upgrade from 0001, not the stamp branch.""" engine = create_engine(f"sqlite:///{tmp_path}/fresh.db") try: with loguru_caplog.at_level("WARNING"): run_migrations(engine) assert get_current_revision(engine) == get_head_revision() assert "BUG-3747" not in loguru_caplog.text finally: engine.dispose() def test_concurrent_stamp_is_neutralized(self, pre_alembic_engine): """stamp_database() must be safe to call twice (race-tolerant).""" stamp_database(pre_alembic_engine, "0001") # Second call simulates a racing concurrent caller — must NOT raise. stamp_database(pre_alembic_engine, "0001") assert get_current_revision(pre_alembic_engine) == "0001" def test_unrelated_operational_error_still_propagates(self, tmp_path): """The race-tolerance guard must NOT swallow non-race errors.""" from alembic import command from sqlalchemy.exc import OperationalError engine = create_engine(f"sqlite:///{tmp_path}/will_error.db") original_stamp = command.stamp def _raise_unrelated(*_args, **_kwargs): raise OperationalError( "SELECT something", {}, Exception("disk I/O error") ) try: command.stamp = _raise_unrelated with pytest.raises(OperationalError, match="disk I/O error"): stamp_database(engine, "0001") finally: command.stamp = original_stamp engine.dispose() def test_auth_db_shape_users_only_is_refused(self, tmp_path): """Engine with ONLY a `users` table must be refused (auth DB shape).""" engine = create_engine(f"sqlite:///{tmp_path}/auth.db") try: with engine.begin() as conn: conn.execute( text( "CREATE TABLE users " "(id INTEGER PRIMARY KEY, username VARCHAR)" ) ) with pytest.raises(RuntimeError, match="auth database"): run_migrations(engine) finally: engine.dispose() def test_auth_db_shape_users_plus_alembic_version_is_refused( self, tmp_path ): """Auth DB that's been (mis-)stamped must still be refused.""" engine = create_engine(f"sqlite:///{tmp_path}/auth_stamped.db") try: with engine.begin() as conn: conn.execute( text( "CREATE TABLE users " "(id INTEGER PRIMARY KEY, username VARCHAR)" ) ) conn.execute( text( "CREATE TABLE alembic_version " "(version_num VARCHAR(32) NOT NULL PRIMARY KEY)" ) ) with pytest.raises(RuntimeError, match="auth database"): run_migrations(engine) finally: engine.dispose() def test_pre_alembic_user_db_with_users_table_is_NOT_refused( self, tmp_path ): """Pre-Alembic user DBs contain `users` (created by the old `Base.metadata.create_all()` path before migration 0001 added the explicit skip). They must be allowed through, not refused as auth DBs. Uses `Base.metadata.create_all()` to simulate the legacy code path exactly — same approach as the existing pre-Alembic tests (see `test_pre_alembic_db_with_users_table_keeps_it`). """ engine = create_engine(f"sqlite:///{tmp_path}/pre_alembic_user.db") try: Base.metadata.create_all(engine, checkfirst=True) # Sanity: the simulated pre-Alembic DB really does contain # `users` (the modern 0001 migration would have skipped it). assert "users" in inspect(engine).get_table_names() # Must NOT raise as an auth-DB false positive. run_migrations(engine) assert get_current_revision(engine) == get_head_revision() finally: engine.dispose() def test_only_alembic_version_table_is_treated_as_fresh(self, tmp_path): """A bare alembic_version (no schema, no users) is OK — runs upgrade.""" engine = create_engine(f"sqlite:///{tmp_path}/bare.db") try: with engine.begin() as conn: conn.execute( text( "CREATE TABLE alembic_version " "(version_num VARCHAR(32) NOT NULL " "PRIMARY KEY)" ) ) run_migrations(engine) assert get_current_revision(engine) == get_head_revision() finally: engine.dispose() class TestMigrationAppliesSchemaChanges: """Tests proving the migration machinery actually changes schema and ORM works. These tests go beyond table/column name checks to verify: - Alembic Operations API works on our engine - Column properties (nullable, defaults) are correct after migration - ORM CRUD works after the production entry point - Downgrade/upgrade roundtrip preserves column properties """ @pytest.fixture def migrated_to_head(self, tmp_path): """Engine with all migrations applied.""" engine = create_engine(f"sqlite:///{tmp_path / 'head.db'}") run_migrations(engine, target="head") yield engine engine.dispose() def test_operations_api_can_modify_schema(self, migrated_to_head): """Alembic Operations API can create and drop a table on our engine.""" from alembic.migration import MigrationContext from alembic.operations import Operations from sqlalchemy import Column, Integer, String with migrated_to_head.connect() as conn: ctx = MigrationContext.configure(conn) op = Operations(ctx) # Before: test table does NOT exist inspector = inspect(migrated_to_head) assert "_test_migration_probe" not in inspector.get_table_names() # Apply: create test table op.create_table( "_test_migration_probe", Column("id", Integer, primary_key=True), Column("name", String(50)), ) # After: test table EXISTS with correct columns inspector = inspect(migrated_to_head) assert "_test_migration_probe" in inspector.get_table_names() cols = { c["name"] for c in inspector.get_columns("_test_migration_probe") } assert cols == {"id", "name"} # Cleanup: drop it op.drop_table("_test_migration_probe") inspector = inspect(migrated_to_head) assert "_test_migration_probe" not in inspector.get_table_names() def test_0011_adds_is_public_to_existing_collections(self, tmp_path): """Upgrade path: a DB created BEFORE 0011 (collections without is_public) gains the column on upgrade, and a pre-existing collection row is backfilled to private (False) via the server_default — existing users' collections don't become NULL/garbage or accidentally public. """ from alembic import command engine = create_engine(f"sqlite:///{tmp_path / 'pre_0011.db'}") try: cfg = get_alembic_config(engine) # Simulate a pre-0011 DB: a collections table created by an older # version of the code (WITHOUT is_public), with the alembic # version stamped at 0010 so only 0011 runs. (The migration chain # can't reproduce this — 0001 builds from the *current* model, # which already has is_public — so we hand-create the old schema, # exactly as the 0002 test does.) with engine.begin() as conn: conn.execute( text( """ CREATE TABLE collections ( id VARCHAR(36) PRIMARY KEY, name VARCHAR(255) NOT NULL, description TEXT, collection_type VARCHAR(50), is_default BOOLEAN, created_at DATETIME, updated_at DATETIME ) """ ) ) conn.execute( text( "INSERT INTO collections (id, name) " "VALUES ('legacy-c1', 'Legacy')" ) ) with engine.begin() as conn: cfg.attributes["connection"] = conn command.stamp(cfg, "0010") cols = { c["name"] for c in inspect(engine).get_columns("collections") } assert "is_public" not in cols # Upgrade to head — 0011 adds is_public. with engine.begin() as conn: cfg.attributes["connection"] = conn command.upgrade(cfg, "head") col_map = { c["name"]: c for c in inspect(engine).get_columns("collections") } assert "is_public" in col_map assert col_map["is_public"]["nullable"] is True # The pre-existing row is backfilled to private (falsy). with engine.connect() as conn: val = conn.execute( text( "SELECT is_public FROM collections WHERE id='legacy-c1'" ) ).scalar() assert not val, ( f"existing row not private after upgrade: {val!r}" ) finally: engine.dispose() def test_0002_adds_columns_with_properties_to_old_schema(self, tmp_path): """Migration 0002 adds progress columns with correct properties to a pre-Alembic schema. The 0001 migration creates all tables from Base.metadata (which includes progress columns). The 0002 migration is designed for pre-Alembic databases where task_metadata was created WITHOUT progress columns. This test simulates that scenario: stamp at 0001, manually create the old schema without progress columns, then run 0002 and verify column properties. """ engine = create_engine(f"sqlite:///{tmp_path / 'old_schema_props.db'}") try: # Create task_metadata WITHOUT progress columns (simulating old schema) with engine.begin() as conn: conn.execute( text( """ CREATE TABLE task_metadata ( task_id VARCHAR PRIMARY KEY, status VARCHAR NOT NULL, task_type VARCHAR NOT NULL, created_at DATETIME, started_at DATETIME, completed_at DATETIME, error_message VARCHAR, priority INTEGER DEFAULT 0, retry_count INTEGER DEFAULT 0, max_retries INTEGER DEFAULT 3 ) """ ) ) # Run the full migration — 0001 will create remaining tables, # and 0002 will add the missing progress columns run_migrations(engine) # AFTER: verify columns exist with correct properties inspector = inspect(engine) col_map = { c["name"]: c for c in inspector.get_columns("task_metadata") } assert "progress_current" in col_map assert "progress_total" in col_map assert "progress_message" in col_map assert "metadata_json" in col_map # Verify nullable constraints assert col_map["progress_message"]["nullable"] is True assert col_map["metadata_json"]["nullable"] is True # Verify defaults work via INSERT without specifying progress columns with engine.begin() as conn: conn.execute( text( "INSERT INTO task_metadata (task_id, status, task_type) " "VALUES ('default-test', 'queued', 'research')" ) ) row = conn.execute( text( "SELECT progress_current, progress_total FROM task_metadata " "WHERE task_id='default-test'" ) ).fetchone() assert row[0] == 0 # default assert row[1] == 0 # default finally: engine.dispose() def test_orm_crud_after_initialize_database(self, tmp_path): """ORM CRUD works after initialize_database() — the production entry point.""" from local_deep_research.database.models import TaskMetadata, Setting from local_deep_research.database.models.settings import SettingType engine = create_engine(f"sqlite:///{tmp_path / 'crud.db'}") try: Session = sessionmaker(bind=engine) # Production entry point with Session() as session: initialize_database(engine, session) # Verify migrations ran assert get_current_revision(engine) == get_head_revision() with Session() as session: # CREATE — TaskMetadata with 0002 progress columns task = TaskMetadata( task_id="orm-test-001", status="queued", task_type="research", progress_current=5, progress_total=100, progress_message="Searching...", metadata_json={"source": "test"}, ) session.add(task) session.commit() # READ — verify all fields round-trip loaded = ( session.query(TaskMetadata) .filter_by(task_id="orm-test-001") .first() ) assert loaded is not None assert loaded.progress_current == 5 assert loaded.progress_total == 100 assert loaded.progress_message == "Searching..." assert loaded.metadata_json == {"source": "test"} # UPDATE loaded.progress_current = 50 loaded.progress_message = "Halfway" session.commit() reloaded = ( session.query(TaskMetadata) .filter_by(task_id="orm-test-001") .first() ) assert reloaded.progress_current == 50 assert reloaded.progress_message == "Halfway" # CREATE — Setting model (value is JSON type, key must be unique) setting = Setting( key="test.migration.orm", value="works", type=SettingType.APP, name="Test Setting", category="test", ) session.add(setting) session.commit() loaded_setting = ( session.query(Setting) .filter_by(key="test.migration.orm") .first() ) assert loaded_setting.value == "works" # DELETE session.delete(reloaded) session.commit() assert ( session.query(TaskMetadata) .filter_by(task_id="orm-test-001") .first() is None ) finally: engine.dispose() def test_downgrade_upgrade_roundtrip_verifies_column_properties( self, tmp_path ): """Roundtrip downgrade/upgrade preserves column properties and ORM works. Starts at 0008 (last fully-reversible) rather than head; 0010 is intentionally non-reversible (SQLite ALTER TABLE limitations). Re-upgrade to head still verifies forward path through 0010. """ from alembic import command from local_deep_research.database.models import TaskMetadata migrated_to_head = create_engine( f"sqlite:///{tmp_path / 'roundtrip_head.db'}" ) try: # Upgrade to 0008 (last reversible) config = get_alembic_config(migrated_to_head) with migrated_to_head.begin() as conn: config.attributes["connection"] = conn command.upgrade(config, "0008") # Insert data at 0008 with migrated_to_head.begin() as conn: conn.execute( text( "INSERT INTO task_metadata (task_id, status, task_type, progress_current) " "VALUES ('roundtrip', 'queued', 'research', 42)" ) ) # DOWNGRADE to 0001 with migrated_to_head.begin() as conn: config.attributes["connection"] = conn command.downgrade(config, "0001") # After downgrade: progress columns gone, core data kept inspector = inspect(migrated_to_head) cols = {c["name"] for c in inspector.get_columns("task_metadata")} assert "progress_current" not in cols assert "task_id" in cols with migrated_to_head.connect() as conn: row = conn.execute( text( "SELECT task_id FROM task_metadata WHERE task_id='roundtrip'" ) ).fetchone() assert row is not None # RE-UPGRADE to head (passes through 0010 in forward direction) run_migrations(migrated_to_head, target="head") # After re-upgrade: verify column PROPERTIES (not just names) inspector = inspect(migrated_to_head) col_map = { c["name"]: c for c in inspector.get_columns("task_metadata") } assert col_map["progress_message"]["nullable"] is True assert col_map["metadata_json"]["nullable"] is True # Verify defaults still work after roundtrip with migrated_to_head.begin() as conn: conn.execute( text( "INSERT INTO task_metadata (task_id, status, task_type) " "VALUES ('after-roundtrip', 'queued', 'research')" ) ) row = conn.execute( text( "SELECT progress_current, progress_total FROM task_metadata " "WHERE task_id='after-roundtrip'" ) ).fetchone() assert row[0] == 0 assert row[1] == 0 # ORM CRUD works after roundtrip Session = sessionmaker(bind=migrated_to_head) with Session() as session: task = TaskMetadata( task_id="post-roundtrip", status="queued", task_type="research", progress_current=10, progress_total=50, ) session.add(task) session.commit() loaded = ( session.query(TaskMetadata) .filter_by(task_id="post-roundtrip") .first() ) assert loaded.progress_current == 10 finally: migrated_to_head.dispose() # --------------------------------------------------------------------------- # Helper for parametrized safety-guard tests # --------------------------------------------------------------------------- def _get_revision_chain(): """Return [(revision_id, down_revision), ...] ordered base → head. Resolved at import time so ``@pytest.mark.parametrize`` can consume it during test collection. """ from alembic.config import Config from alembic.script import ScriptDirectory from local_deep_research.database.alembic_runner import get_migrations_dir cfg = Config() cfg.set_main_option("script_location", str(get_migrations_dir())) script = ScriptDirectory.from_config(cfg) chain = [] for rev in script.walk_revisions(): down = rev.down_revision if rev.down_revision else "base" chain.append((rev.revision, down)) chain.reverse() # walk_revisions yields head-first return chain _REVISION_CHAIN = _get_revision_chain() def _get_migration_stems(): """Return migration file stems for parametrized import tests. Resolved at import time so ``@pytest.mark.parametrize`` can consume it. """ from local_deep_research.database.alembic_runner import get_migrations_dir versions_dir = get_migrations_dir() / "versions" return [p.stem for p in sorted(versions_dir.glob("[0-9]*.py"))] _MIGRATION_STEMS = _get_migration_stems() class TestNoCircularFkSawarning: """Regression: the documents ↔ research_resources circular FK used to emit a SAWarning from Base.metadata.sorted_tables on every cold start. Fixed by adding `use_alter=True` to ResearchResource.document_id so SQLAlchemy emits that one FK as a post-CREATE ALTER TABLE (breaking the dependency cycle for sorting purposes) while still creating the constraint at the database level. """ def test_create_all_emits_no_circular_fk_warning(self, tmp_path): import warnings as _warnings engine = create_engine(f"sqlite:///{tmp_path}/test.db") try: with _warnings.catch_warnings(record=True) as caught: _warnings.simplefilter("always") Base.metadata.create_all(engine) offending = [ w for w in caught if "unresolvable cycles" in str(w.message) and "documents" in str(w.message) and "research_resources" in str(w.message) ] assert not offending, ( f"create_all() emitted the circular-FK SAWarning: " f"{[str(w.message) for w in offending]}" ) finally: engine.dispose() class TestMigrationSafetyGuards: """Guards that catch common migration pitfalls. These are *structural* checks — they don't test application logic, they prevent forgotten migrations, broken downgrades, branch conflicts, and orphaned models. """ # -- fixtures ----------------------------------------------------------- @pytest.fixture def fresh_engine(self, tmp_path): """Disposable in-memory style SQLite engine (file-backed for inspect).""" db_path = tmp_path / "guard_test.db" engine = create_engine(f"sqlite:///{db_path}") yield engine engine.dispose() @pytest.fixture def alembic_cfg(self, fresh_engine): """Alembic Config wired to *fresh_engine*.""" return get_alembic_config(fresh_engine) # -- 1. schema drift ---------------------------------------------------- def test_migrations_produce_schema_matching_models( self, fresh_engine, alembic_cfg ): """After upgrading to head the DB schema must match ORM metadata. Diffs that are *expected* and filtered out: * ``add_table('users')`` — the ``users`` table lives in Base.metadata but 0001 deliberately skips it (auth-only DB). * ``remove_index`` for the 9 performance indexes added by 0003 — those indexes are migration-only and intentionally absent from ORM model declarations. """ from alembic import command from alembic.autogenerate import compare_metadata from alembic.runtime.migration import MigrationContext # Run all migrations on a clean DB with fresh_engine.begin() as conn: alembic_cfg.attributes["connection"] = conn command.upgrade(alembic_cfg, "head") # Diff the resulting schema against ORM metadata with fresh_engine.connect() as conn: mc = MigrationContext.configure(conn) raw_diffs = compare_metadata(mc, Base.metadata) # Filter expected diffs unexpected = [] for diff in raw_diffs: op_type = diff[0] # users table is intentionally excluded from migrations if op_type == "add_table" and diff[1].name == "users": continue # Index on users table (comes with the table definition) if op_type == "add_index" and diff[1].table.name == "users": continue # 0003 migration-only indexes show as "remove_index" if op_type == "remove_index": continue unexpected.append(diff) assert unexpected == [], ( f"Schema drift detected — {len(unexpected)} diff(s) between " f"migrations and models:\n" + "\n".join(str(d) for d in unexpected) ) # -- 2. single head ----------------------------------------------------- def test_single_head_revision(self): """There must be exactly one head revision (no branch conflicts).""" from alembic.config import Config as AlembicConfig from alembic.script import ScriptDirectory from local_deep_research.database.alembic_runner import ( get_migrations_dir, ) cfg = AlembicConfig() cfg.set_main_option("script_location", str(get_migrations_dir())) script = ScriptDirectory.from_config(cfg) heads = script.get_heads() assert len(heads) == 1, ( f"Expected exactly 1 head revision, found {len(heads)}: {heads}. " "This means two migrations share the same parent — " "resolve with `alembic merge`." ) # -- 3. stairway (up-down-up per revision) ------------------------------ @pytest.mark.parametrize( "revision,down_rev", _REVISION_CHAIN, ids=[r[0] for r in _REVISION_CHAIN], ) def test_stairway_up_down_up_per_revision( self, fresh_engine, alembic_cfg, revision, down_rev ): """Each revision must survive: parent → up → down → up.""" from alembic import command if revision in self.NON_REVERSIBLE_REVISIONS: pytest.skip( f"Revision {revision} is intentionally non-reversible " "(SQLite ALTER TABLE limitations); see NON_REVERSIBLE_REVISIONS." ) target_down = down_rev if down_rev != "base" else "base" expected_after_down = None if down_rev == "base" else down_rev # Upgrade to parent first (unless this is the base migration) if down_rev != "base": with fresh_engine.begin() as conn: alembic_cfg.attributes["connection"] = conn command.upgrade(alembic_cfg, down_rev) # Up with fresh_engine.begin() as conn: alembic_cfg.attributes["connection"] = conn command.upgrade(alembic_cfg, revision) assert get_current_revision(fresh_engine) == revision # Down with fresh_engine.begin() as conn: alembic_cfg.attributes["connection"] = conn command.downgrade(alembic_cfg, target_down) assert get_current_revision(fresh_engine) == expected_after_down # Up again with fresh_engine.begin() as conn: alembic_cfg.attributes["connection"] = conn command.upgrade(alembic_cfg, revision) assert get_current_revision(fresh_engine) == revision # -- 4. substantive downgrades ------------------------------------------ # Migrations whose downgrade is intentionally a no-op (e.g. one-time # data migrations that delete stale keys with no consumers). INTENTIONAL_NOOP_DOWNGRADES = { "0004_migrate_legacy_app_settings.py", # 0013: restoring 'auto'/'parallel' references would recreate # broken state — those engines no longer exist in the codebase. "0013_remove_meta_search_engines.py", # 0016: the dropped cache tables were orphaned dead code holding no # data, and their models are removed — nothing to recreate. "0016_drop_orphaned_cache_tables.py", # 0018: restoring 'mcp'/'agentic' strategy references would recreate # broken state — that strategy no longer exists in the codebase. "0018_remove_mcp_strategy.py", # 0019: the 'both' -> 'adaptive' coercion is lossy — after upgrade a # row is indistinguishable from an originally-'adaptive' one, so a # downgrade cannot know which rows to restore. Rewriting 'adaptive' # back to 'both' would clobber the legitimate default. No-op by design. "0019_retire_both_egress_scope.py", } # Migrations whose downgrade is intentionally NotImplementedError # (e.g. SQLite ALTER TABLE limitations against legacy unnamed # constraints + FK-target columns). These revisions are exempt from # parametrized stairway/residual tests and tested separately. NON_REVERSIBLE_REVISIONS = { "0010", # 0010_add_chat_tables.py — chat schema, dev-stage rollback path: recreate DB } def test_all_downgrades_are_substantive(self): """Every migration's ``downgrade()`` must contain real operations. A ``pass``-only or empty downgrade silently blocks rollback. Migrations listed in INTENTIONAL_NOOP_DOWNGRADES are exempt. """ import ast from local_deep_research.database.alembic_runner import ( get_migrations_dir, ) versions_dir = get_migrations_dir() / "versions" for py_file in sorted(versions_dir.glob("[0-9]*.py")): if py_file.name in self.INTENTIONAL_NOOP_DOWNGRADES: continue tree = ast.parse(py_file.read_text()) for node in ast.walk(tree): if not isinstance(node, ast.FunctionDef): continue if node.name != "downgrade": continue # Strip docstrings — a single Expr(Constant(str)) at position 0 body = list(node.body) if ( body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant) and isinstance(body[0].value.value, str) ): body = body[1:] # Must have at least one non-pass statement non_pass = [s for s in body if not isinstance(s, ast.Pass)] assert non_pass, ( f"{py_file.name}: downgrade() is empty or pass-only — " "this migration cannot be rolled back." ) # -- 5. all models registered on metadata ------------------------------- def test_all_models_registered_on_metadata(self): """Every model submodule must be imported so Alembic can see its tables.""" import importlib import pkgutil import local_deep_research.database.models as models_pkg # Force-import every submodule under the models package for _importer, modname, _ispkg in pkgutil.walk_packages( models_pkg.__path__, prefix=models_pkg.__name__ + ".", ): importlib.import_module(modname) # Critical tables that must exist in metadata critical_tables = { "settings", "research_tasks", "research_history", "task_metadata", "queue_status", "benchmark_runs", "benchmark_results", "benchmark_configs", "benchmark_progress", "token_usage", "search_calls", "reports", "report_sections", "app_logs", "provider_models", "documents", "collections", } registered = set(Base.metadata.tables.keys()) missing = critical_tables - registered assert not missing, ( f"Tables missing from Base.metadata (model not imported?): " f"{sorted(missing)}" ) # -- 6. downgrade leaves no residual tables ---------------------------- @pytest.mark.parametrize( "revision,down_rev", _REVISION_CHAIN, ids=[r[0] for r in _REVISION_CHAIN], ) def test_downgrade_leaves_no_residual_tables( self, fresh_engine, alembic_cfg, revision, down_rev ): """Downgrading a revision must not leave behind tables it created.""" from alembic import command if revision in self.NON_REVERSIBLE_REVISIONS: pytest.skip( f"Revision {revision} is intentionally non-reversible " "(SQLite ALTER TABLE limitations); see NON_REVERSIBLE_REVISIONS." ) with fresh_engine.begin() as conn: alembic_cfg.attributes["connection"] = conn # Upgrade to parent first if down_rev != "base": command.upgrade(alembic_cfg, down_rev) tables_before = set(inspect(fresh_engine).get_table_names()) with fresh_engine.begin() as conn: alembic_cfg.attributes["connection"] = conn # Upgrade to this revision command.upgrade(alembic_cfg, revision) with fresh_engine.begin() as conn: alembic_cfg.attributes["connection"] = conn # Downgrade back target_down = down_rev if down_rev != "base" else "base" command.downgrade(alembic_cfg, target_down) tables_after = set(inspect(fresh_engine).get_table_names()) # alembic_version may appear/disappear — that's fine residual = (tables_after - tables_before) - {"alembic_version"} assert not residual, ( f"Revision {revision}: downgrade left residual tables: " f"{sorted(residual)}" ) # -- 7. deterministic schema -------------------------------------------- def test_deterministic_schema(self, tmp_path): """Two independent fresh databases must produce identical schemas. Catches migrations that use timestamps, random values, or environment-dependent conditional logic. """ from alembic import command schemas = [] for i in range(2): db_path = tmp_path / f"deterministic_{i}.db" engine = create_engine(f"sqlite:///{db_path}") try: cfg = get_alembic_config(engine) with engine.begin() as conn: cfg.attributes["connection"] = conn command.upgrade(cfg, "head") insp = inspect(engine) schema = {} for table_name in sorted(insp.get_table_names()): cols = [ (c["name"], str(c["type"])) for c in insp.get_columns(table_name) ] idxs = sorted( [ (idx["name"], tuple(idx["column_names"])) for idx in insp.get_indexes(table_name) ] ) schema[table_name] = {"columns": cols, "indexes": idxs} schemas.append(schema) finally: engine.dispose() assert schemas[0] == schemas[1], ( "Two fresh databases produced different schemas — " "a migration is non-deterministic." ) # -- 8. each migration file is importable ------------------------------- @pytest.mark.parametrize("stem", _MIGRATION_STEMS) def test_each_migration_revision_is_importable(self, stem): """Every migration file must import without SyntaxError or ImportError.""" import importlib mod = importlib.import_module( f"local_deep_research.database.migrations.versions.{stem}" ) # Sanity: the module must expose upgrade/downgrade callables assert callable(getattr(mod, "upgrade", None)), ( f"{stem}: missing upgrade()" ) assert callable(getattr(mod, "downgrade", None)), ( f"{stem}: missing downgrade()" ) # -- 9. downgrade data loss is explicit (0002) -------------------------- def test_downgrade_data_loss_is_explicit_0002( self, fresh_engine, alembic_cfg ): """0002's downgrade drops progress columns; re-upgrade gets defaults, not old data.""" from alembic import command # Migrate to 0002 with fresh_engine.begin() as conn: alembic_cfg.attributes["connection"] = conn command.upgrade(alembic_cfg, "0002") # Insert data into the progress columns with fresh_engine.begin() as conn: conn.execute( text( "INSERT INTO task_metadata " "(task_id, task_type, status, progress_current, progress_total, progress_message) " "VALUES (:tid, :tt, :st, :pc, :pt, :pm)" ), { "tid": "test-task-42", "tt": "research", "st": "running", "pc": 7, "pt": 10, "pm": "Step 7 of 10", }, ) # Downgrade to 0001 with fresh_engine.begin() as conn: alembic_cfg.attributes["connection"] = conn command.downgrade(alembic_cfg, "0001") # Verify progress columns are gone cols = { c["name"] for c in inspect(fresh_engine).get_columns("task_metadata") } assert "progress_current" not in cols assert "progress_total" not in cols assert "progress_message" not in cols # Re-upgrade to 0002 with fresh_engine.begin() as conn: alembic_cfg.attributes["connection"] = conn command.upgrade(alembic_cfg, "0002") # Columns are back with defaults, old data is gone with fresh_engine.connect() as conn: row = conn.execute( text( "SELECT progress_current, progress_total, progress_message " "FROM task_metadata WHERE task_id = :tid" ), {"tid": "test-task-42"}, ).fetchone() assert row is not None, ( "Core row should survive the downgrade/upgrade cycle" ) # Defaults: 0, 0, NULL — the old values (7, 10, "Step 7 of 10") are lost assert row[0] == 0, ( f"progress_current should be default 0, got {row[0]}" ) assert row[1] == 0, f"progress_total should be default 0, got {row[1]}" assert row[2] is None, ( f"progress_message should be default NULL, got {row[2]}" ) # -- 10. env.py offline mode raises ------------------------------------- def test_env_offline_mode_raises(self): """env.py's run_migrations_offline() must raise NotImplementedError. Tested via AST to avoid import side-effects (env.py runs run_migrations_online() at module level). """ import ast from local_deep_research.database.alembic_runner import ( get_migrations_dir, ) env_path = get_migrations_dir() / "env.py" tree = ast.parse(env_path.read_text(), filename=str(env_path)) func = None for node in ast.walk(tree): if ( isinstance(node, ast.FunctionDef) and node.name == "run_migrations_offline" ): func = node break assert func is not None, "run_migrations_offline not found in env.py" # Strip docstring body = list(func.body) if ( body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant) and isinstance(body[0].value.value, str) ): body = body[1:] assert len(body) == 1, ( f"run_migrations_offline should contain only a Raise, " f"got {len(body)} statement(s)" ) stmt = body[0] assert isinstance(stmt, ast.Raise), ( f"Expected Raise statement, got {type(stmt).__name__}" ) # Verify it raises NotImplementedError specifically assert isinstance(stmt.exc, ast.Call), ( "Expected Raise with a Call (e.g. NotImplementedError(...))" ) assert isinstance(stmt.exc.func, ast.Name), ( "Expected exception to be a named type" ) assert stmt.exc.func.id == "NotImplementedError", ( f"Expected NotImplementedError, got {stmt.exc.func.id}" ) # -- 10b. env.py online mode requires connection ------------------------- def test_env_online_mode_requires_connection(self): """env.py's run_migrations_online() must raise RuntimeError when config.attributes['connection'] is None. Tested via AST to avoid import side-effects (env.py runs run_migrations_online() at module level). """ import ast from local_deep_research.database.alembic_runner import ( get_migrations_dir, ) env_path = get_migrations_dir() / "env.py" tree = ast.parse(env_path.read_text(), filename=str(env_path)) func = None for node in ast.walk(tree): if ( isinstance(node, ast.FunctionDef) and node.name == "run_migrations_online" ): func = node break assert func is not None, "run_migrations_online not found in env.py" # Find the RuntimeError raise in the function body has_runtime_error = False for node in ast.walk(func): if isinstance(node, ast.Raise) and isinstance(node.exc, ast.Call): if ( isinstance(node.exc.func, ast.Name) and node.exc.func.id == "RuntimeError" ): has_runtime_error = True break assert has_runtime_error, ( "run_migrations_online must raise RuntimeError when no " "connection is provided" ) # -- 10c. migration 0004 guards on missing settings table --------------- def test_migration_0004_skips_without_settings_table(self, tmp_path): """Migration 0004 should succeed silently on a database without a settings table (e.g., if only partial tables exist). Target is 0008 (last revision that doesn't require a fully-formed research_history schema). 0010 ADDs a column to research_history which the partial-DB shape doesn't have; tested separately. """ from local_deep_research.database.alembic_runner import ( get_alembic_config, ) # Create engine with only alembic_version stamped at 0003 db_path = tmp_path / "no_settings.db" engine = create_engine(f"sqlite:///{db_path}") # Create alembic_version and stamp at 0003 with engine.begin() as conn: conn.execute( text( "CREATE TABLE alembic_version " "(version_num VARCHAR(32) NOT NULL)" ) ) conn.execute( text( "INSERT INTO alembic_version (version_num) VALUES ('0003')" ) ) # Running migrations should not fail — 0004 should skip gracefully config = get_alembic_config(engine) from alembic import command with engine.begin() as conn: config.attributes["connection"] = conn command.upgrade(config, "0008") # Verify we're at 0008 despite no settings table with engine.connect() as conn: row = conn.execute( text("SELECT version_num FROM alembic_version") ).fetchone() assert row[0] == "0008" engine.dispose() # -- 11. revision IDs match filenames ----------------------------------- def test_migration_revision_ids_match_filenames(self): """Each migration file's ``revision`` variable must match its filename prefix. Catches copy-paste errors where a migration file is duplicated but the ``revision = "..."`` inside is not updated. """ import ast from local_deep_research.database.alembic_runner import ( get_migrations_dir, ) versions_dir = get_migrations_dir() / "versions" for py_file in sorted(versions_dir.glob("[0-9]*.py")): filename_prefix = py_file.stem.split("_")[0] # e.g. "0003" tree = ast.parse(py_file.read_text(), filename=str(py_file)) revision_value = None for node in ast.walk(tree): if isinstance(node, ast.Assign): for target in node.targets: if ( isinstance(target, ast.Name) and target.id == "revision" ): if isinstance(node.value, ast.Constant): revision_value = node.value.value break assert revision_value is not None, ( f"{py_file.name}: no 'revision = \"...\"' assignment found" ) assert revision_value == filename_prefix, ( f"{py_file.name}: revision='{revision_value}' does not match " f"filename prefix '{filename_prefix}'" )