470 lines
20 KiB
Python
470 lines
20 KiB
Python
"""Alembic 迁移链冒烟测试（在真实 MySQL 临时库上执行）。
|
||
|
||
覆盖三件事：
|
||
1. 链完整性：只有一个头版本、头与最新迁移一致、每个修订的
|
||
down_revision 都能回溯到根（无悬挂引用、无分叉链）。
|
||
2. 全部迁移脚本可编译：防止迁移文件被改坏后部署流程才暴露。
|
||
3. 空库整链执行：迁移脚本在真实 MySQL 临时库上执行 `upgrade head`，
|
||
随后校验表/列/索引与 Base.metadata 模型元数据对齐（模型要求的不允许缺失），
|
||
并确认迁移新增的复合查询索引确实落库。
|
||
"""
|
||
|
||
import json
|
||
import secrets
|
||
from datetime import UTC, datetime
|
||
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
|
||
from sqlalchemy.orm import Session
|
||
|
||
import web.backend.database as database_module
|
||
from tests.support.db import get_mysql_credentials, get_test_engine
|
||
from web.backend.database import Base
|
||
from web.backend.models import (
|
||
AuditLog,
|
||
HuyaExternalOrder,
|
||
User,
|
||
WeiboAccount,
|
||
WeiboExternalOrder,
|
||
)
|
||
|
||
TESTS_DIR = Path(__file__).resolve().parent
|
||
PROJECT_ROOT = TESTS_DIR.parent
|
||
VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions"
|
||
|
||
HEAD_REVISION = "20261112_0079"
|
||
|
||
# 迁移新增、但不声明在模型里的复合查询索引。
|
||
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}`;"))
|
||
|
||
def test_0077_converts_super_admin_membership_to_overlay(self):
|
||
creds = get_mysql_credentials()
|
||
mig_db = f"douyu_login_mig_{secrets.token_hex(4)}"
|
||
mig_url = (
|
||
f"mysql+pymysql://{quote_plus(creds['user'])}:"
|
||
f"{quote_plus(creds['password'])}@{creds['host']}:{creds['port']}/"
|
||
f"{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
|
||
mig_engine = create_engine(mig_url)
|
||
try:
|
||
database_module.DATABASE_URL = mig_url
|
||
cfg = _migration_config()
|
||
cfg.set_main_option("sqlalchemy.url", mig_url)
|
||
command.upgrade(cfg, "20261111_0076")
|
||
|
||
with Session(mig_engine) as session:
|
||
admin = User(
|
||
username="migration-admin",
|
||
password_hash="hash",
|
||
role="super_admin",
|
||
is_active=True,
|
||
)
|
||
support = User(
|
||
username="migration-support",
|
||
password_hash="hash",
|
||
role="support",
|
||
is_active=True,
|
||
)
|
||
admin_account = WeiboAccount(
|
||
uid="4014000771",
|
||
nickname="管理员旧映射",
|
||
cookie="SCF=a; SUB=admin",
|
||
)
|
||
support_account = WeiboAccount(
|
||
uid="4014000772",
|
||
nickname="客服旧映射",
|
||
cookie="SCF=b; SUB=support",
|
||
)
|
||
session.add_all([admin, support, admin_account, support_account])
|
||
session.commit()
|
||
admin_id = admin.id
|
||
support_id = support.id
|
||
admin_account_id = admin_account.id
|
||
support_account_id = support_account.id
|
||
|
||
with mig_engine.begin() as conn:
|
||
for user_id, account_id in (
|
||
(admin_id, admin_account_id),
|
||
(support_id, support_account_id),
|
||
):
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO weibo_workbench_accounts "
|
||
"(user_id, account_id, created_at) "
|
||
"VALUES (:user_id, :account_id, UTC_TIMESTAMP())"
|
||
),
|
||
{"user_id": user_id, "account_id": account_id},
|
||
)
|
||
|
||
command.upgrade(cfg, "20261112_0077")
|
||
|
||
with mig_engine.begin() as conn:
|
||
rows = conn.execute(
|
||
text(
|
||
"SELECT user_id, account_id, exclusive_slot "
|
||
"FROM weibo_workbench_accounts ORDER BY user_id"
|
||
)
|
||
).all()
|
||
assert {
|
||
(user_id, account_id, exclusive_slot)
|
||
for user_id, account_id, exclusive_slot in rows
|
||
} == {
|
||
(admin_id, admin_account_id, None),
|
||
(support_id, support_account_id, 1),
|
||
}
|
||
|
||
# 管理员可以给同一个账号再挂一条并行映射，普通唯一槽位保持不变。
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO weibo_workbench_accounts "
|
||
"(user_id, account_id, exclusive_slot, created_at) "
|
||
"VALUES (:user_id, :account_id, NULL, UTC_TIMESTAMP())"
|
||
),
|
||
{"user_id": admin_id, "account_id": support_account_id},
|
||
)
|
||
assert (
|
||
conn.execute(
|
||
text(
|
||
"SELECT COUNT(*) FROM weibo_workbench_accounts "
|
||
"WHERE account_id = :account_id"
|
||
),
|
||
{"account_id": support_account_id},
|
||
).scalar_one()
|
||
== 2
|
||
)
|
||
finally:
|
||
database_module.DATABASE_URL = original_url
|
||
mig_engine.dispose()
|
||
with admin_engine.begin() as conn:
|
||
conn.execute(text(f"DROP DATABASE IF EXISTS `{mig_db}`;"))
|
||
|
||
def test_0078_cleans_only_pre_void_points_snapshots(self):
|
||
creds = get_mysql_credentials()
|
||
mig_db = f"douyu_login_mig_{secrets.token_hex(4)}"
|
||
mig_url = (
|
||
f"mysql+pymysql://{quote_plus(creds['user'])}:"
|
||
f"{quote_plus(creds['password'])}@{creds['host']}:{creds['port']}/"
|
||
f"{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
|
||
mig_engine = create_engine(mig_url)
|
||
try:
|
||
database_module.DATABASE_URL = mig_url
|
||
cfg = _migration_config()
|
||
cfg.set_main_option("sqlalchemy.url", mig_url)
|
||
command.upgrade(cfg, "20261112_0077")
|
||
|
||
with Session(mig_engine) as session:
|
||
user = User(
|
||
username="migration-collab",
|
||
password_hash="hash",
|
||
role="super_admin",
|
||
is_active=True,
|
||
)
|
||
session.add(user)
|
||
session.commit()
|
||
|
||
def make_order(
|
||
order_no: str,
|
||
*,
|
||
status: str,
|
||
checked_at: datetime,
|
||
required_total: int,
|
||
surplus: int,
|
||
) -> HuyaExternalOrder:
|
||
return HuyaExternalOrder(
|
||
order_no=order_no,
|
||
shop="向日葵",
|
||
uid="1199665363355",
|
||
cookie="C" * 32,
|
||
raw_cookie="raw",
|
||
ck_status="valid",
|
||
bind_game_account="泡泡",
|
||
bind_game_role="厶婕",
|
||
gift_name="套装-芝士鼠鼠",
|
||
required_points=988,
|
||
home_images=["collab/2026/11/12/home.png"],
|
||
status=status,
|
||
points_checked_at=checked_at,
|
||
points_balance=988,
|
||
points_required_total=required_total,
|
||
points_surplus=surplus,
|
||
points_threshold=100,
|
||
points_check_reason="shortage" if surplus < 0 else "ok",
|
||
points_check_message="旧核账",
|
||
points_check_detail={"orders": []},
|
||
created_by=user.id,
|
||
updated_by=user.id,
|
||
)
|
||
|
||
stale = make_order(
|
||
"MIG-0078-STALE",
|
||
status="points_error",
|
||
checked_at=datetime(2026, 11, 12, 1, 46, tzinfo=UTC),
|
||
required_total=1976,
|
||
surplus=-988,
|
||
)
|
||
voided = make_order(
|
||
"MIG-0078-VOID",
|
||
status="void",
|
||
checked_at=datetime(2026, 11, 12, 1, 46, tzinfo=UTC),
|
||
required_total=1976,
|
||
surplus=-988,
|
||
)
|
||
current = make_order(
|
||
"MIG-0078-CURRENT",
|
||
status="recharged",
|
||
checked_at=datetime(2026, 11, 12, 7, 51, 50, tzinfo=UTC),
|
||
required_total=988,
|
||
surplus=0,
|
||
)
|
||
session.add_all([stale, voided, current])
|
||
session.commit()
|
||
session.add(
|
||
AuditLog(
|
||
user_id=user.id,
|
||
username=user.username,
|
||
action="huya:collab:status",
|
||
target=f"huya_external_order:{voided.id}",
|
||
detail=json.dumps({"to": "void"}),
|
||
success=True,
|
||
created_at=datetime(2026, 11, 12, 1, 49, 53, tzinfo=UTC),
|
||
)
|
||
)
|
||
session.commit()
|
||
|
||
command.upgrade(cfg, "20261112_0078")
|
||
|
||
with mig_engine.connect() as conn:
|
||
rows = {
|
||
row.order_no: row
|
||
for row in conn.execute(
|
||
text(
|
||
"SELECT order_no, status, points_checked_at, "
|
||
"points_required_total, points_surplus, "
|
||
"points_check_reason, points_check_detail "
|
||
"FROM huya_external_orders"
|
||
)
|
||
)
|
||
}
|
||
assert rows["MIG-0078-VOID"].status == "void"
|
||
assert rows["MIG-0078-VOID"].points_checked_at is None
|
||
assert rows["MIG-0078-VOID"].points_required_total is None
|
||
assert rows["MIG-0078-VOID"].points_check_reason == "stale_group"
|
||
|
||
assert rows["MIG-0078-STALE"].status == "to_recharge"
|
||
assert rows["MIG-0078-STALE"].points_checked_at is None
|
||
assert rows["MIG-0078-STALE"].points_required_total is None
|
||
assert rows["MIG-0078-STALE"].points_check_reason == "stale_group"
|
||
|
||
assert rows["MIG-0078-CURRENT"].status == "recharged"
|
||
assert rows["MIG-0078-CURRENT"].points_required_total == 988
|
||
assert rows["MIG-0078-CURRENT"].points_surplus == 0
|
||
assert rows["MIG-0078-CURRENT"].points_check_reason == "ok"
|
||
finally:
|
||
database_module.DATABASE_URL = original_url
|
||
mig_engine.dispose()
|
||
with admin_engine.begin() as conn:
|
||
conn.execute(text(f"DROP DATABASE IF EXISTS `{mig_db}`;"))
|
||
|
||
def test_0079_moves_exchange_pending_to_recharged(self):
|
||
creds = get_mysql_credentials()
|
||
mig_db = f"douyu_login_mig_{secrets.token_hex(4)}"
|
||
mig_url = (
|
||
f"mysql+pymysql://{quote_plus(creds['user'])}:"
|
||
f"{quote_plus(creds['password'])}@{creds['host']}:{creds['port']}/"
|
||
f"{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
|
||
mig_engine = create_engine(mig_url)
|
||
try:
|
||
database_module.DATABASE_URL = mig_url
|
||
cfg = _migration_config()
|
||
cfg.set_main_option("sqlalchemy.url", mig_url)
|
||
command.upgrade(cfg, "20261112_0078")
|
||
|
||
with Session(mig_engine) as session:
|
||
order = WeiboExternalOrder(
|
||
order_no="MIG-0079-PENDING",
|
||
status="exchange_pending",
|
||
exchange_started_at=datetime(2026, 11, 12, 8, 30, tzinfo=UTC),
|
||
)
|
||
session.add(order)
|
||
session.commit()
|
||
|
||
command.upgrade(cfg, "20261112_0079")
|
||
|
||
with mig_engine.connect() as conn:
|
||
row = conn.execute(
|
||
text(
|
||
"SELECT status, recharged_at "
|
||
"FROM weibo_external_orders "
|
||
"WHERE order_no = 'MIG-0079-PENDING'"
|
||
)
|
||
).one()
|
||
assert row.status == "recharged"
|
||
assert row.recharged_at is not None
|
||
finally:
|
||
database_module.DATABASE_URL = original_url
|
||
mig_engine.dispose()
|
||
with admin_engine.begin() as conn:
|
||
conn.execute(text(f"DROP DATABASE IF EXISTS `{mig_db}`;"))
|