Neo-ZQYY/tools/db/verify_ddl_migration.py

"""
DDL 迁移验证脚本 — admin-web-enhancement spec (Task 17.2)

验证 NS4.1 + P16 迁移在 test_zqyy_app 上的正确性：
1. biz.connectors / biz.tenants / biz.sites / biz.site_code_history 四张表存在且字段正确
2. scheduled_tasks 新增 3 个字段
3. auth._archived_site_code_mapping 存在
"""

import os
import sys

from dotenv import load_dotenv

load_dotenv()

dsn = os.environ.get("APP_DB_DSN", "")
if not dsn:
    print("ERROR: APP_DB_DSN 未设置")
    sys.exit(1)
if "test_" not in dsn:
    print(f"ERROR: APP_DB_DSN 不包含 'test_'，拒绝连接非测试库: {dsn}")
    sys.exit(1)

import psycopg2  # noqa: E402

conn = psycopg2.connect(dsn)
cur = conn.cursor()

results: list[tuple[str, bool, str]] = []


def check(name: str, sql: str, expected_fn):
    """执行 SQL 并用 expected_fn 判断结果，记录 pass/fail。"""
    try:
        cur.execute(sql)
        rows = cur.fetchall()
        ok, detail = expected_fn(rows)
        results.append((name, ok, detail))
    except Exception as e:
        results.append((name, False, f"异常: {e}"))


# ── 1. biz.connectors 表字段 ──
check(
    "biz.connectors 表存在且字段正确",
    """
    SELECT column_name FROM information_schema.columns
    WHERE table_schema = 'biz' AND table_name = 'connectors'
    ORDER BY ordinal_position
    """,
    lambda rows: (
        (cols := [r[0] for r in rows])
        and set(cols) == {"id", "connector_key", "display_name", "is_active", "created_at"},
        f"字段: {cols}",
    ),
)

# ── 2. biz.tenants 表字段 ──
check(
    "biz.tenants 表存在且字段正确",
    """
    SELECT column_name FROM information_schema.columns
    WHERE table_schema = 'biz' AND table_name = 'tenants'
    ORDER BY ordinal_position
    """,
    lambda rows: (
        (cols := [r[0] for r in rows])
        and set(cols)
        == {"id", "connector_id", "tenant_id", "tenant_name", "is_active", "created_at", "updated_at"},
        f"字段: {cols}",
    ),
)

# ── 3. biz.sites 表字段 ──
check(
    "biz.sites 表存在且字段正确",
    """
    SELECT column_name FROM information_schema.columns
    WHERE table_schema = 'biz' AND table_name = 'sites'
    ORDER BY ordinal_position
    """,
    lambda rows: (
        (cols := [r[0] for r in rows])
        and set(cols)
        == {
            "id", "tenant_id", "site_id", "site_name", "site_code",
            "site_label", "is_active", "created_at", "updated_at",
        },
        f"字段: {cols}",
    ),
)

# ── 4. biz.site_code_history 表字段 ──
check(
    "biz.site_code_history 表存在且字段正确",
    """
    SELECT column_name FROM information_schema.columns
    WHERE table_schema = 'biz' AND table_name = 'site_code_history'
    ORDER BY ordinal_position
    """,
    lambda rows: (
        (cols := [r[0] for r in rows])
        and set(cols) == {"id", "site_id", "site_code", "is_current", "created_at", "retired_at"},
        f"字段: {cols}",
    ),
)

# ── 5. scheduled_tasks 新增 3 个字段 ──
check(
    "scheduled_tasks 新增字段存在",
    """
    SELECT column_name, data_type, column_default
    FROM information_schema.columns
    WHERE table_schema = 'public' AND table_name = 'scheduled_tasks'
      AND column_name IN ('min_run_interval_value', 'min_run_interval_unit', 'last_success_at')
    ORDER BY ordinal_position
    """,
    lambda rows: (
        len(rows) == 3,
        f"找到 {len(rows)} 个字段: {[r[0] for r in rows]}",
    ),
)

# ── 6. auth._archived_site_code_mapping 存在 ──
check(
    "auth._archived_site_code_mapping 存在",
    """
    SELECT table_name FROM information_schema.tables
    WHERE table_schema = 'auth' AND table_name = '_archived_site_code_mapping'
    """,
    lambda rows: (
        len(rows) == 1,
        f"找到 {len(rows)} 张表",
    ),
)

cur.close()
conn.close()

# ── 打印结果 ──
print("\n" + "=" * 60)
print("DDL 迁移验证结果 — admin-web-enhancement")
print("=" * 60)

passed = 0
failed = 0
for name, ok, detail in results:
    status = "✅ PASS" if ok else "❌ FAIL"
    print(f"  {status}  {name}")
    print(f"         {detail}")
    if ok:
        passed += 1
    else:
        failed += 1

print(f"\n总计: {passed} 通过, {failed} 失败")
if failed > 0:
    sys.exit(1)
print("全部通过 ✅")