"""Tests for the idempotent property-slug migration script. Builds a temporary SQLite DB with the OLD `properties` schema (no slug / list_on_landing columns) via raw DDL, points the migration module at it, and verifies the migration adds the columns/index and deterministically backfills RO-transliterated, collision-safe slugs -- twice, with no error and no drift the second time. """ import sqlite3 import pytest from sqlalchemy import create_engine, inspect, text from sqlalchemy.orm import sessionmaker import migrate_add_property_slug as migrate_mod def _create_old_schema_db(path: str) -> None: """Create a properties table matching the pre-slug schema and seed rows.""" conn = sqlite3.connect(path) conn.execute( """ CREATE TABLE properties ( id INTEGER PRIMARY KEY, name VARCHAR NOT NULL, description VARCHAR, address VARCHAR, is_public BOOLEAN NOT NULL DEFAULT 1, is_active BOOLEAN NOT NULL DEFAULT 1, created_at DATETIME NOT NULL ) """ ) rows = [ ("Clădirea Centrală", "2024-01-01T00:00:00"), # Same (accented) name again -> must collide and get a -2 suffix. ("Clădirea Centrală", "2024-01-02T00:00:00"), ("Sediul Ș.ț Test", "2024-01-03T00:00:00"), ] for name, created_at in rows: conn.execute( "INSERT INTO properties (name, description, address, is_public, is_active, created_at) " "VALUES (?, NULL, NULL, 1, 1, ?)", (name, created_at), ) conn.commit() conn.close() @pytest.fixture def temp_db_path(tmp_path) -> str: path = tmp_path / "migrate_test.db" _create_old_schema_db(str(path)) return str(path) @pytest.fixture def patched_engine(monkeypatch: pytest.MonkeyPatch, temp_db_path: str): """Point the migration module's module-level engine/SessionLocal at the temp DB.""" engine = create_engine(f"sqlite:///{temp_db_path}", connect_args={"check_same_thread": False}) session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine) monkeypatch.setattr(migrate_mod, "engine", engine) monkeypatch.setattr(migrate_mod, "SessionLocal", session_local) yield engine engine.dispose() def _fetch_rows(engine) -> list[tuple]: with engine.connect() as conn: return conn.execute(text("SELECT id, name, slug FROM properties ORDER BY id")).fetchall() def test_migration_adds_columns_and_unique_index(patched_engine) -> None: migrate_mod.migrate() inspector = inspect(patched_engine) columns = {c["name"] for c in inspector.get_columns("properties")} assert "slug" in columns assert "list_on_landing" in columns indexes = inspector.get_indexes("properties") slug_indexes = [ix for ix in indexes if "slug" in ix["column_names"]] assert slug_indexes, "expected an index on slug" assert any(ix["unique"] for ix in slug_indexes), "slug index must be unique" def test_migration_backfills_deterministic_ro_slugs_with_collision_suffix(patched_engine) -> None: migrate_mod.migrate() rows = _fetch_rows(patched_engine) assert len(rows) == 3 slugs = [r[2] for r in rows] # RO diacritic transliteration: ă->a, â->a, î->i, ș/ş->s, ț/ţ->t. assert slugs[0] == "cladirea-centrala" # Second property has the identical (transliterated) name -> collision -> -2 suffix. assert slugs[1] == "cladirea-centrala-2" assert slugs[2] == "sediul-s-t-test" # All slugs non-null and unique. assert all(s for s in slugs) assert len(set(slugs)) == len(slugs) def test_migration_is_idempotent_and_slugs_stable_on_rerun(patched_engine) -> None: migrate_mod.migrate() first = _fetch_rows(patched_engine) # Running a second time must not raise and must not change existing slugs # or column/index state. migrate_mod.migrate() second = _fetch_rows(patched_engine) assert first == second inspector = inspect(patched_engine) columns = {c["name"] for c in inspector.get_columns("properties")} assert "slug" in columns assert "list_on_landing" in columns