WonderQ-Project/WonderQ-Admin/tests/test_bootstrap_migration_compatibility.py

from datetime import datetime
from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path

import pytest
import sqlalchemy as sa


VERSIONS_DIR = Path(__file__).parents[1] / "alembic" / "versions"


def load_migration(filename: str):
    path = VERSIONS_DIR / filename
    spec = spec_from_file_location(f"bootstrap_compatibility_{path.stem}", path)
    assert spec and spec.loader
    migration = module_from_spec(spec)
    spec.loader.exec_module(migration)
    return migration


class CurrentSchemaInspector:
    tables = {
        "Customer",
        "DetailRecord",
        "HomeTeamBuilding",
        "HomeWildArchive",
        "Lead",
        "VehicleServiceConfig",
    }
    columns = {
        "DetailRecord": {"conciergeAdvisorId"},
        "HomeTeamBuilding": {"detailParagraphs", "detailSubtitle"},
        "HomeWildArchive": {"images"},
        "Lead": {"contactName", "customerId", "leadType", "vehicleDemand"},
        "VehicleServiceConfig": {
            "advantages",
            "intro",
            "introTitle",
            "isActive",
            "processSteps",
            "serviceSections",
        },
    }

    def get_table_names(self) -> list[str]:
        return sorted(self.tables)

    def get_columns(self, table_name: str) -> list[dict[str, object]]:
        return [
            {"name": name, "type": sa.String()}
            for name in sorted(self.columns.get(table_name, set()))
        ]

    def get_indexes(self, table_name: str) -> list[dict[str, object]]:
        indexes = {
            "Lead": {"ix_Lead_customerId", "ix_Lead_leadType"},
            "VehicleServiceConfig": {"ix_VehicleServiceConfig_isActive"},
        }
        return [{"name": name} for name in sorted(indexes.get(table_name, set()))]

    def get_foreign_keys(self, table_name: str) -> list[dict[str, object]]:
        if table_name != "Lead":
            return []
        return [
            {
                "name": "Lead_customerId_fkey",
                "constrained_columns": ["customerId"],
                "referred_table": "Customer",
                "referred_columns": ["id"],
            }
        ]


class ExistingRowsResult:
    def first(self):
        return ("existing",)

    def mappings(self):
        return self

    def all(self) -> list[object]:
        return []


class ExistingRowsBind:
    def execute(self, _statement) -> ExistingRowsResult:
        return ExistingRowsResult()


def record_operation(monkeypatch, migration, operation: str, events: list[tuple[str, ...]]) -> None:
    monkeypatch.setattr(
        migration.op,
        operation,
        lambda *args, **_kwargs: events.append((operation, *(str(arg) for arg in args))),
    )


def test_home_content_seed_rows_cover_current_non_null_detail_columns():
    migration = load_migration("0017_home_content.py")
    timestamp = datetime(2026, 1, 1)

    team_rows = migration._prepare_seed_rows(
        migration.SEED_TEAM_BUILDINGS,
        {"id", "description", "detailSubtitle", "detailParagraphs", "createdAt", "updatedAt"},
        timestamp,
    )
    archive_rows = migration._prepare_seed_rows(
        migration.SEED_WILD_ARCHIVES,
        {"id", "image", "images", "createdAt", "updatedAt"},
        timestamp,
    )

    assert team_rows[0]["detailSubtitle"] == team_rows[0]["description"]
    assert team_rows[0]["detailParagraphs"] == [team_rows[0]["description"]]
    assert archive_rows[0]["images"] == [archive_rows[0]["image"]]


def test_home_content_online_seed_reflects_current_table_columns(monkeypatch):
    migration = load_migration("0017_home_content.py")
    inspector = CurrentSchemaInspector()
    inspector.tables = {"HomeTeamBuilding"}
    inspector.columns = {
        "HomeTeamBuilding": {
            "createdAt",
            "demandKeyword",
            "description",
            "detailParagraphs",
            "detailSubtitle",
            "id",
            "image",
            "isActive",
            "sortOrder",
            "tag",
            "title",
            "updatedAt",
        }
    }
    inserted: dict[str, object] = {}

    monkeypatch.setattr(migration.context, "is_offline_mode", lambda: False)
    monkeypatch.setattr(migration, "inspect", lambda _bind: inspector, raising=False)
    monkeypatch.setattr(migration.op, "get_bind", lambda: object())
    monkeypatch.setattr(
        migration.op,
        "bulk_insert",
        lambda table, rows: inserted.update(table=table, rows=rows),
    )

    historical_columns = [
        sa.Column("id", sa.String()),
        sa.Column("description", sa.Text()),
    ]
    migration._seed_table(
        "HomeTeamBuilding",
        historical_columns,
        migration.SEED_TEAM_BUILDINGS,
    )

    inserted_table = inserted["table"]
    inserted_rows = inserted["rows"]
    assert set(inserted_table.c.keys()) == inspector.columns["HomeTeamBuilding"]
    assert inserted_rows[0]["detailSubtitle"] == inserted_rows[0]["description"]
    assert inserted_rows[0]["detailParagraphs"] == [inserted_rows[0]["description"]]


@pytest.mark.parametrize(
    ("filename", "table_name", "column_names"),
    [
        ("0019_home_wild_archive_images.py", "HomeWildArchive", {"images"}),
        (
            "0020_home_team_building_details.py",
            "HomeTeamBuilding",
            {"detailParagraphs", "detailSubtitle"},
        ),
        ("0023_detail_concierge_advisor.py", "DetailRecord", {"conciergeAdvisorId"}),
    ],
)
def test_column_migrations_do_not_add_columns_already_in_current_schema(
    monkeypatch,
    filename: str,
    table_name: str,
    column_names: set[str],
):
    migration = load_migration(filename)
    inspector = CurrentSchemaInspector()
    inspector.tables = {table_name}
    inspector.columns = {table_name: column_names}
    added_columns: list[tuple[str, ...]] = []

    monkeypatch.setattr(migration, "inspect", lambda _bind: inspector, raising=False)
    monkeypatch.setattr(migration.op, "get_bind", lambda: object())
    record_operation(monkeypatch, migration, "add_column", added_columns)
    monkeypatch.setattr(migration.op, "execute", lambda *_args, **_kwargs: None)
    monkeypatch.setattr(migration.op, "alter_column", lambda *_args, **_kwargs: None)

    migration.upgrade()

    assert added_columns == []


def test_detail_record_migration_reuses_populated_current_table(monkeypatch):
    migration = load_migration("0021_detail_records.py")
    inspector = CurrentSchemaInspector()
    events: list[tuple[str, ...]] = []

    monkeypatch.setattr(migration, "inspect", lambda _bind: inspector, raising=False)
    monkeypatch.setattr(migration.op, "get_bind", lambda: ExistingRowsBind())
    for operation in ("bulk_insert", "create_index", "create_table"):
        record_operation(monkeypatch, migration, operation, events)

    migration.upgrade()

    assert events == []


def test_vehicle_demand_migration_reuses_current_schema_objects(monkeypatch):
    migration = load_migration("0024_vehicle_demand.py")
    inspector = CurrentSchemaInspector()
    events: list[tuple[str, ...]] = []

    monkeypatch.setattr(migration, "inspect", lambda _bind: inspector, raising=False)
    monkeypatch.setattr(migration.op, "get_bind", lambda: object())
    for operation in ("add_column", "create_foreign_key", "create_index", "create_table"):
        record_operation(monkeypatch, migration, operation, events)
    monkeypatch.setattr(migration.op, "alter_column", lambda *_args, **_kwargs: None)

    migration.upgrade()

    assert events == []