Files
live-hub-py/tests/test_migrations.py
T
yml2213 0b1fc1dee0 feat(events): 补齐资产【出项】—— 开通金额、鱼翅消耗、积分到期(C 最全)
审「哪些操作该进账号生命周期」时发现:**进项有账、出项没账** ——
充值/积分到账都有金额,而开通(花积分)、送礼(花鱼翅)几乎没记,
于是「一共充了多少」算得出来,「花哪儿了」算不出来。

按确认的范围一次做完(应用宝充值**不动** —— 它按游戏角色记账、且未启用)。

## A. 开通类事件补金额

`elite_opened` / `esports_opened` 以前只记「开通后的积分余额」,不记**花了多少**。
现在补 `amount = -(开通前后积分差)`,并用**实测余额差**而不是"开通要多少分"的常量
(活动价、折扣会让实际扣减不同)。任一侧拿不到就不写金额,不编数字。

## B. 鱼翅消耗事件(新 `gold_spent`)

送礼以前只发「积分到账」,**花掉的鱼翅没有任何记录**。现在:
- 送礼**前后各刷一次**鱼翅(原来只在送礼后刷),取**实测余额差**;
- 发 `gold_spent`,`amount` 用负数(与 `exchange` 一致);
- 余额没减少或拿不到 → **不发**:宁可少一条,也不要记一笔没发生的消耗。

## C. 积分到期(新 `points_expire`)

积分到期是资产真实减少,但**斗鱼不会通知** —— 只在积分查询里给 `willExpireTime`。
所以:
- 新增 `accounts.points_expire_at`(迁移 `0046`),送礼流程学到就写下来;
- 新增作业 `points_expire_check`:扫「已到期且积分未归零」的账号,补写事件并清标记
  (不重复记);积分已为 0 则只清标记、不发事件(避免 `-0` 噪声)。

⚠️ **金额的诚实边界**:到期那一刻**到底扣了多少,我们没亲眼看到**。
所以默认 `confirmed=False`、标题带「(预计)」,记的是「当时余额」这个**预计要归零**的数量,
**不假装它是确认过的扣减**。作业刻意不发网络请求(纯 DB 可跑、不需要 CK),
要变成已确认只能由「到期后重查、真看到余额下降」的调用方补。

## 顺带修掉一个会静默丢事件的坑

同一个任务可能产生**两条事件**(送礼:既「积分到账」又「消耗鱼翅」)。
`source_ref` 是唯一键 —— 共用 `task_ref(task_id)` 会让**后写的那条被静默丢弃**。
已让副事件带后缀(`douyu_task:{id}:gold`),主事件保持原形态(与旧收集器一致)。
有用例专门钉住「一个任务两条事件都不丢」。

## 验证

- `TZ=Asia/Shanghai`:**550 passed / 6 skipped**(+18 条)
- `TZ=UTC`(事件 + 迁移 + 绑定)95 passed;ruff 全绿;前端 build 通过、lint 0 errors
- 真实开发库:迁移 `0046` 已应用;作业烟测 → 扫到 1 个到期账号,
  写入「积分到期 -500(预计)」`confirmed=False`,标记被清、重跑不再重复

## 现在的资产流水(斗鱼侧)

| 方向 | 覆盖 |
| --- | --- |
| 进项 | 充值(二维码/供应商/回调三路径去重)✅、积分到账 ✅ |
| 出项 | 兑换 ✅、**开通 ✅(新补金额)**、**送礼鱼翅 ✅(新)**、**积分到期 ✅(新,标预计)** |
| 未动 | 应用宝充值(独立系统、按游戏角色记、当前未启用) |
2026-10-04 18:10:46 +08:00

156 lines
7.1 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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 = "20261004_0046"
# 迁移新增、但不声明在模型里的复合查询索引。
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}`;"))