数据迁移bug修复v2

WonderQ-Admin/tests/test_bootstrap_migration_compatibility.py

@@ -0,0 +1,229 @@
+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 == []