diff --git a/app/database.py b/app/database.py index bff3ac08..146b14c2 100644 --- a/app/database.py +++ b/app/database.py @@ -210,8 +210,10 @@ def _run_schema_migrations(engine: Any) -> None: if unique_filehash_indexes: logger.info("Migrating files: dropping unique index on 'filehash'") with engine.begin() as conn: + preparer = conn.dialect.identifier_preparer for index in unique_filehash_indexes: - conn.execute(text(f"DROP INDEX IF EXISTS {index['name']}")) + quoted_idx = preparer.quote(index["name"]) + conn.execute(text(f"DROP INDEX IF EXISTS {quoted_idx}")) logger.info("Migration complete: unique index on 'filehash' removed") except Exception as exc: logger.warning(f"Skipping filehash unique index drop: {exc}") @@ -263,12 +265,16 @@ def _ensure_indexes(engine: Any, inspector: Any) -> None: table_names = inspector.get_table_names() columns_by_table: dict[str, set[str]] = {} with engine.begin() as conn: + preparer = conn.dialect.identifier_preparer for idx_name, table, column in _PERF_INDEXES: if table in table_names: if table not in columns_by_table: columns_by_table[table] = {col["name"] for col in inspector.get_columns(table)} if column in columns_by_table[table]: - conn.execute(text(f"CREATE INDEX IF NOT EXISTS {idx_name} ON {table} ({column})")) + quoted_idx = preparer.quote(idx_name) + quoted_table = preparer.quote(table) + quoted_col = preparer.quote(column) + conn.execute(text(f"CREATE INDEX IF NOT EXISTS {quoted_idx} ON {quoted_table} ({quoted_col})")) logger.info("Performance indexes ensured") diff --git a/tests/test_database.py b/tests/test_database.py index 47a56a40..64e0e70e 100644 --- a/tests/test_database.py +++ b/tests/test_database.py @@ -671,6 +671,75 @@ class TestMultiVersionMigrations: engine.dispose() + def test_migration_drops_unique_index_with_quoted_name(self, tmp_path): + """Test that a unique filehash index whose name requires quoting is dropped correctly. + + Index names containing special characters such as dashes must be quoted by + the dialect's identifier_preparer before being interpolated into raw SQL. + This test verifies that _run_schema_migrations() handles such names without + SQL errors and leaves the underlying table and its data intact. + """ + from sqlalchemy import create_engine, inspect, text + + from app.database import _run_schema_migrations + + db_path = str(tmp_path / "quoted_index.db") + engine = create_engine(f"sqlite:///{db_path}") + + # Build a files table that contains all columns the migration expects, + # then add a unique index on filehash whose name contains a dash — a + # character that requires quoting by the dialect's identifier_preparer. + with engine.begin() as conn: + conn.execute( + text( + "CREATE TABLE files (" + "id INTEGER PRIMARY KEY, " + "filename VARCHAR, " + "filehash VARCHAR, " + "original_file_path VARCHAR, " + "processed_file_path VARCHAR, " + "is_duplicate BOOLEAN DEFAULT FALSE NOT NULL, " + "duplicate_of_id INTEGER)" + ) + ) + # Index name deliberately contains a dash to exercise the quoting path. + conn.execute(text('CREATE UNIQUE INDEX "ix-filehash-unique" ON files (filehash)')) + conn.execute(text("INSERT INTO files (filename, filehash) VALUES ('doc.pdf', 'abc123')")) + + # Confirm the index exists before migration. + inspector = inspect(engine) + pre_indexes = [idx["name"] for idx in inspector.get_indexes("files")] + assert "ix-filehash-unique" in pre_indexes + + # Run migration — must not raise despite the special character in the index name. + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + _run_schema_migrations(engine) + + # The unique index on filehash must have been dropped. + inspector = inspect(engine) + post_indexes = inspector.get_indexes("files") + remaining_unique_filehash = [ + idx for idx in post_indexes if idx.get("unique") and "filehash" in idx.get("column_names", []) + ] + assert remaining_unique_filehash == [], ( + f"Expected unique filehash index to be dropped, but found: {remaining_unique_filehash}" + ) + + # The table and its data must still be intact. + files_columns = {col["name"] for col in inspector.get_columns("files")} + assert "id" in files_columns + assert "filename" in files_columns + assert "filehash" in files_columns + + with engine.connect() as conn: + row = conn.execute(text("SELECT filename, filehash FROM files WHERE filehash = 'abc123'")).fetchone() + assert row is not None + assert row[0] == "doc.pdf" + assert row[1] == "abc123" + + engine.dispose() + def test_migration_exception_handling(self, tmp_path): """Test that migration handles exceptions gracefully for index operations.""" from sqlalchemy import create_engine, text