fix alembic migration

api/tests/test_alembic_migration_policy.py

@@ -0,0 +1,95 @@
+"""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 == ["0019_align_legacy_schema"], 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_0019_from_0018() -> None:
+    script = _script_dir()
+    rev = script.get_revision("0019_align_legacy_schema")
+    assert rev is not None
+    assert rev.down_revision == "0018_users_language_preference"
+
+
+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"
+        )