live-hub-py/tests/test_migrations.py

"""Alembic 迁移链冒烟测试（在真实 MySQL 临时库上执行）。

覆盖三件事：
1. 链完整性：只有一个头版本、头与最新迁移一致、每个修订的
   down_revision 都能回溯到根（无悬挂引用、无分叉链）。
2. 全部迁移脚本可编译：防止迁移文件被改坏后部署流程才暴露。
3. 空库整链执行：迁移脚本在真实 MySQL 临时库上执行 `upgrade head`，
   随后校验表/列/索引与 Base.metadata 模型元数据对齐（模型要求的不允许缺失），
   并确认迁移新增的复合查询索引确实落库。
"""

import secrets
from pathlib import Path
from urllib.parse import quote_plus

from alembic import command
from alembic.config import Config
from alembic.script import ScriptDirectory
from sqlalchemy import create_engine, inspect, text

import web.backend.database as database_module
from tests.support.db import get_mysql_credentials, get_test_engine
from web.backend.database import Base

TESTS_DIR = Path(__file__).resolve().parent
PROJECT_ROOT = TESTS_DIR.parent
VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions"

HEAD_REVISION = "20261007_0052"

# 迁移新增、但不声明在模型里的复合查询索引。
EXTRA_INDEXES = (
    "ix_douyu_tasks_handbook_scope_id",
    "ix_douyu_tasks_handbook_scope_task_type_id",
    "ix_douyu_tasks_status_finished_at_id",
    "ix_login_tasks_status_finished_at_id",
    # 账号时间线按「账号 + 时间倒序」查，模型里不声明这个复合索引。
    "ix_account_events_account_occurred",
    # CK 搬到账号后，「在途尝试」只能留在登录记录上，巡检/重登每轮按
    # (account_id, attempt_status) 过滤 —— 模型里不声明这个索引。
    "ix_login_tasks_account_attempt_status",
)


def _migration_config() -> Config:
    config = Config(str(PROJECT_ROOT / "alembic.ini"))
    config.set_main_option(
        "script_location", str(PROJECT_ROOT / "web" / "backend" / "migrations")
    )
    return config


class TestMigrationSmoke:
    def test_chain_is_linear_and_head_matches_latest_file(self):
        directory = ScriptDirectory.from_config(_migration_config())

        heads = directory.get_heads()
        assert len(heads) == 1, f"迁移链出现多个头版本（存在分叉）: {heads}"
        assert heads[0] == HEAD_REVISION, f"头版本 {heads[0]} 与最新迁移不一致"

        version_files = sorted(path.name for path in VERSIONS_DIR.glob("*.py"))
        revisions = list(directory.walk_revisions())
        revision_ids = {revision.revision for revision in revisions}
        assert len(revision_ids) == len(version_files), (
            "迁移文件数量与已加载修订数量不一致（存在无法加载的脚本）"
        )
        for revision in revisions:
            if revision.down_revision is not None:
                assert revision.down_revision in revision_ids, (
                    f"{revision.revision} 引用了不存在的父版本 {revision.down_revision}"
                )

    def test_all_migration_scripts_compile(self):
        for path in sorted(VERSIONS_DIR.glob("*.py")):
            source = path.read_text(encoding="utf-8")
            compile(source, str(path), "exec")  # 语法/缩进等编译错误会在此抛出

    def test_full_chain_applies_on_fresh_mysql_and_matches_models(self):
        """空库执行整条迁移链，校验表/列/索引与模型元数据对齐。"""
        creds = get_mysql_credentials()
        mig_db = f"douyu_login_mig_{secrets.token_hex(4)}"
        mig_url = f"mysql+pymysql://{quote_plus(creds['user'])}:{quote_plus(creds['password'])}@{creds['host']}:{creds['port']}/{mig_db}?charset=utf8mb4"

        admin_engine = get_test_engine()
        with admin_engine.begin() as conn:
            conn.execute(
                text(
                    f"CREATE DATABASE `{mig_db}` CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci;"
                )
            )

        original_url = database_module.DATABASE_URL
        try:
            database_module.DATABASE_URL = mig_url
            cfg = _migration_config()
            cfg.set_main_option("sqlalchemy.url", mig_url)
            command.upgrade(cfg, "head")

            inspector = inspect(create_engine(mig_url))
            db_tables = set(inspector.get_table_names())
            assert db_tables

            model_tables = Base.metadata.tables
            missing_tables = set(model_tables) - db_tables
            assert not missing_tables

            all_db_indexes = set()
            nullability_mismatches: list[str] = []
            for name, table in model_tables.items():
                db_cols = inspector.get_columns(name)
                db_columns = {column["name"] for column in db_cols}
                model_columns = set(table.columns.keys())
                missing_columns = model_columns - db_columns
                assert not missing_columns

                # 可空性只查**危险的那个方向**：模型说可空、库里却是 NOT NULL。
                # 那样带 NULL 的写入会**在运行时静默失败**（`emit()` 按设计吞异常），
                # 正是 `account_events.source_ref` 那次事故的形态 ——
                # 只比对「列在不在」是抓不到的。
                #
                # 反方向（模型 NOT NULL、库里允许 NULL）是**无害的宽松**：
                # SQLAlchemy 总会写值，库里宽松一点不会丢数据。历史遗留里有不少，
                # 一并报出来只会制造噪声、逼人去修没有收益的东西。
                db_nullable = {c["name"]: c["nullable"] for c in db_cols}
                for column in table.columns:
                    if column.name not in db_nullable:
                        continue
                    if bool(column.nullable) and not bool(db_nullable[column.name]):
                        nullability_mismatches.append(
                            f"{name}.{column.name}: 模型说可空，"
                            f"但迁移建成了 NOT NULL（带 NULL 的写入会静默丢失）"
                        )

                db_indexes = set()
                for index in inspector.get_indexes(name):
                    index_name = index["name"]
                    if index_name:
                        db_indexes.add(index_name)
                all_db_indexes |= db_indexes
                model_indexes = {index.name for index in table.indexes if index.name}
                missing_indexes = model_indexes - db_indexes
                assert not missing_indexes

            assert not nullability_mismatches, (
                "迁移建出的列比模型更严（会在运行期静默丢数据）：\n"
                + "\n".join(nullability_mismatches)
            )

            # 迁移新增的复合查询索引（不入模型，但应存在）。
            for index_name in EXTRA_INDEXES:
                assert index_name in all_db_indexes, f"迁移链未创建索引 {index_name}"
        finally:
            database_module.DATABASE_URL = original_url
            with admin_engine.begin() as conn:
                conn.execute(text(f"DROP DATABASE IF EXISTS `{mig_db}`;"))


class TestRollbackBackfillScript:
    """回滚前置脚本：把账号侧 CK 搬回 `login_tasks`。

    为什么必须有它：S2 之后 CK **只写 `accounts.cookie`、不双写** `login_tasks`，
    而降级会 drop 掉 `accounts.cookie` —— 不先搬回去，
    **部署期间新拿到的 CK 会永久丢失**（本项目唯一不能承受的损失）。

    演练验证过：跑了它再降级，部署期间新拿到的 CK 完整幸存。
    """

    def test_moves_account_cookie_back_to_login_tasks(self):
        import importlib.util
        from pathlib import Path

        import web.backend.database as db_module
        from tests.support.db import make_test_session
        from web.backend.models import Account, LoginTask, User

        path = Path(PROJECT_ROOT) / "scripts" / "backfill_login_tasks_from_accounts.py"
        spec = importlib.util.spec_from_file_location("rollback_backfill", path)
        assert spec and spec.loader
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

        engine = get_test_engine()
        db_module.SessionLocal.configure(bind=engine)
        db = make_test_session(engine)
        try:
            db.query(LoginTask).filter(LoginTask.batch_id == "rb-test").delete()
            db.commit()
            user = User(username="rb-user", password_hash="h", role="super_admin")
            db.add(user)
            db.commit()
            account = Account(
                username="rb-acc",
                password="p",
                email="e@e.com",
                email_password="ep",
            )
            db.add(account)
            db.commit()
            # 账号侧有新 CK；登录记录侧还是旧值（模拟「部署期间新拿到」）
            account.cookie = "PHPSESSID=NEW_DEPLOY_WINDOW_CK"
            db.add(
                LoginTask(
                    batch_id="rb-test",
                    account_id=account.id,
                    status="success",
                    message="旧行",
                    cookie="PHPSESSID=OLD",
                    created_by=user.id,
                )
            )
            db.commit()

            result = module.run(db, apply=True)
            assert result["updated"] >= 1

            db.expire_all()
            tasks = (
                db.query(LoginTask)
                .filter(
                    LoginTask.account_id == account.id,
                    LoginTask.cookie.isnot(None),
                    LoginTask.cookie != "",
                )
                .all()
            )
            assert any(t.cookie == "PHPSESSID=NEW_DEPLOY_WINDOW_CK" for t in tasks), (
                "部署期间的新 CK 没有搬回登录记录 —— 降级会丢"
            )
        finally:
            db.close()

    def test_creates_synthetic_row_when_no_history(self):
        """新代码跑期间新导入、没有任何登录记录的账号，也要能回填。

        旧代码靠 `login_tasks` 找 CK，没有行就等于「这账号没 CK」。
        """
        import importlib.util
        from pathlib import Path

        import web.backend.database as db_module
        from tests.support.db import make_test_session
        from web.backend.models import Account, LoginTask, User

        path = Path(PROJECT_ROOT) / "scripts" / "backfill_login_tasks_from_accounts.py"
        spec = importlib.util.spec_from_file_location("rollback_backfill2", path)
        assert spec and spec.loader
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

        engine = get_test_engine()
        db_module.SessionLocal.configure(bind=engine)
        db = make_test_session(engine)
        try:
            user = User(username="rb-user2", password_hash="h", role="super_admin")
            db.add(user)
            db.commit()
            account = Account(
                username="rb-orphan",
                password="p",
                email="e@e.com",
                email_password="ep",
                cookie="PHPSESSID=ORPHAN_CK",
            )
            db.add(account)
            db.commit()
            assert (
                db.query(LoginTask).filter(LoginTask.account_id == account.id).count()
                == 0
            )

            module.run(db, apply=True)
            db.expire_all()
            row = (
                db.query(LoginTask)
                .filter(LoginTask.account_id == account.id)
                .order_by(LoginTask.id.desc())
                .first()
            )
            assert row is not None, "没有历史行时必须补一条合成行"
            assert row.cookie == "PHPSESSID=ORPHAN_CK"
            assert row.batch_id == module.BACKFILL_BATCH
        finally:
            db.close()