life-echo/api/tests/test_alembic_migration_policy.py

"""Alembic 迁移链与项目策略的静态校验（不依赖线上 Postgres）。"""

from __future__ import annotations

import re
from pathlib import Path

import pytest
from alembic.config import Config
from alembic.script import ScriptDirectory

_API_DIR = Path(__file__).resolve().parent.parent
_VERSIONS_DIR = _API_DIR / "alembic" / "versions"

# 0019 必须显式覆盖的老库缺列（与 ORM / 历史事故相关）
_REQUIRED_LEGACY_COLUMNS = frozenset(
    {
        ("segments", "audio_duration_seconds"),
        ("conversations", "deleted_at"),
        ("segments", "tts_audio_urls"),
        ("conversation_messages", "tts_audio_urls"),
    }
)

_FORBIDDEN_WITHDRAWN_REVISIONS = frozenset(
    {
        "0020_add_tts_audio_urls_column",
        "0020_backfill_missing_schema",
        "0020_backfill_all_missing_columns",
        "0019_backfill_missing_columns",
    }
)


def _script_dir() -> ScriptDirectory:
    cfg = Config(str(_API_DIR / "alembic.ini"))
    return ScriptDirectory.from_config(cfg)


def test_single_alembic_head() -> None:
    heads = _script_dir().get_heads()
    assert heads == ["0021_memory_source_segment_id"], f"unexpected heads: {heads}"


def test_no_withdrawn_revision_ids_in_tree() -> None:
    for rev in _script_dir().walk_revisions():
        assert rev.revision not in _FORBIDDEN_WITHDRAWN_REVISIONS, (
            f"withdrawn revision still in tree: {rev.revision}"
        )


def test_no_withdrawn_migration_files() -> None:
    names = {p.name for p in _VERSIONS_DIR.glob("*.py")}
    assert "0020_add_tts_audio_urls_column.py" not in names
    assert "0019_backfill_missing_columns.py" not in names


def test_0019_align_legacy_schema_covers_required_columns() -> None:
    path = _VERSIONS_DIR / "0019_align_legacy_schema.py"
    src = path.read_text(encoding="utf-8")
    assert 'revision: str = "0019_align_legacy_schema"' in src
    assert "Base.metadata" not in src, "0019 must not introspect full ORM metadata"
    assert "sorted_tables" not in src

    found: set[tuple[str, str]] = set()
    for table, column in _REQUIRED_LEGACY_COLUMNS:
        if f'"{table}"' in src and f'"{column}"' in src:
            found.add((table, column))

    missing = _REQUIRED_LEGACY_COLUMNS - found
    assert not missing, f"0019 missing explicit legacy columns: {missing}"


def test_all_revisions_have_unique_ids() -> None:
    ids: list[str] = []
    for rev in _script_dir().walk_revisions():
        ids.append(rev.revision)
    assert len(ids) == len(set(ids)), "duplicate revision ids"


def test_revision_chain_reaches_0021_from_0020() -> None:
    script = _script_dir()
    rev = script.get_revision("0021_memory_source_segment_id")
    assert rev is not None
    assert rev.down_revision == "0020_refresh_rt_lineage"


def test_no_autogenerate_introspection_backfill_pattern() -> None:
    """禁止再次引入「遍历 ORM 全表补列」类迁移。"""
    pattern = re.compile(r"for table in Base\.metadata\.sorted_tables")
    for path in _VERSIONS_DIR.glob("*.py"):
        text = path.read_text(encoding="utf-8")
        assert not pattern.search(text), (
            f"{path.name} uses full-ORM introspection backfill; use explicit column list"
        )