space-booking/backend/tests/test_migrate_slug.py

"""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