harness_zbt/backend/tests/test_product_management_migration.py

from importlib.util import module_from_spec, spec_from_file_location
from pathlib import Path
from types import ModuleType
from typing import Any

from sqlalchemy.schema import Table


def load_migration() -> ModuleType:
    migration_path = (
        Path(__file__).resolve().parents[1]
        / "migrations"
        / "core"
        / "versions"
        / "core_0004_product_management.py"
    )
    spec = spec_from_file_location("core_0004_product_management", migration_path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"无法加载迁移文件：{migration_path}")
    module = module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


migration = load_migration()


class ExistingPlansInspector:
    def get_columns(self, table_name: str) -> list[dict[str, str]]:
        assert table_name == "plans"
        return [
            {"name": "premium_cents"},
            {"name": "coverage_amount_cents"},
            {"name": "min_age"},
            {"name": "max_age"},
        ]

    def has_table(self, table_name: str) -> bool:
        assert table_name == "product_change_logs"
        return False


def test_product_management_migration_accepts_fresh_schema_from_current_baseline(
    monkeypatch: Any,
) -> None:
    added_columns: list[str] = []
    created_tables: list[str] = []
    monkeypatch.setattr(migration.op, "get_bind", lambda: object())
    monkeypatch.setattr(
        migration.op,
        "add_column",
        lambda _table_name, column: added_columns.append(str(column.name)),
    )
    monkeypatch.setattr(
        Table,
        "create",
        lambda self, bind: created_tables.append(self.name),
    )
    monkeypatch.setattr(
        migration.sa,
        "inspect",
        lambda _bind: ExistingPlansInspector(),
    )

    migration.upgrade()

    assert added_columns == []
    assert created_tables == ["product_change_logs"]