Files
live-hub-py/tests/test_migrations.py
T
yml2213 6e729e7a80 feat(account): S5 业务事实直发事件 + 删掉收集器 + 任务日志保留策略
时间线从此覆盖**全部**业务事实,`douyu_tasks` 也有了可控的增长边界。

## 一、删掉收集器(它必然漏数据)

`jobs/account_events_collect.py` 连同 `_task_to_event` / `COLLECTED_TASK_TYPES` /
水位线一起删除,迁移 `20261004_0043` 清掉它的作业行。删除理由是实测出来的:

- **靠 allowlist 认业务类型,新增类型就静默丢**:原有名单只有 5 个,
  而 `exchange_esports_goods`(电竞兑换)、`exchange_xpd_goods`(和平小店兑换)、
  `pay_locked_order`(锁单支付)、`donate_esports_*`(电竞礼物积分)**从来没进过时间线**;
- **「任务 success」≠「业务真的发生了」**(见下);
- **字段路径靠抽样猜**:二维码充值的 `result` 里根本没有金额,收集器恒写
  `amount=None`,`recharge_total` 系统性少计;
- 作业默认 `enabled=False`,生产上一直没跑。

历史**不回填**(既有结论不变):需要历史走斗鱼侧权威接口
(`query_exchange_records` / `query_xpd_purchase_records` / `query_points` 都是现成任务类型)。

## 二、业务事实直发:命名构造函数

新增 `services/business_events.py`:`record_recharge` / `record_exchange` /
`record_points_credit` / `record_bind`。每类事实的字段、标题、金额符号只定义一次
(与 S4 的 `record_account_action`、`on_ck_verdict` 同一套路)。

**9 处成功点接线**:`exchange_goods` 真兑换分支、`exchange_esports_goods`、
`exchange_xpd_goods`、`pay_locked_order`、`create_gold_qr`(二维码 + 供应商轮询 +
**供应商回调**三条路径)、`donate_elite_gift`、`donate_esports_gift`、
`confirm_bind`、`confirm_esports_bind`。

**只记真实事实**(按你的决定):`exchange_goods` 里「预约到货 / 已预兑 / 预兑确认」
三条 success 分支**不发事件** —— 它们不消耗积分也没换到东西,旧收集器只按 status
收集,于是每预约一次就生成一条金额为空的 `exchange` 把 `exchange_count` 灌大。
`confirm_bind` 里「检测到账号本来就已绑定」那条也跳过。

## 三、修掉两个真 bug(都是钱账)

1. **到账判据 `!=` → `>`**(`douyu_runner_gold.py`):余额**下降**也算「鱼翅已到账」。
   批内并发(最高 10)时同账号送礼会消耗鱼翅,余额一变就被充值任务读成到账,
   **往台账记一笔根本没发生的充值**。
2. **供应商回调晚于 runner 超时**:原来守卫是「任务状态没被改过才更新」,
   于是 runner 先判 failed、回调随后说成功的那笔真金白银**永远不进时间线**。
   改成按**回调自己的终态**判断,并补发事件。

## 四、发射顺序:一个会静默丢事件的坑

`_mark_task` 内部 `db.commit()`,而 `emit` 用 `db.begin_nested()` **只 RELEASE
SAVEPOINT、不提交外层事务**;runner 的 `finally` 只 `close()`(未提交即回滚)。
所以**在 `_mark_task` 之后发事件会被静默丢弃,且 `emit` 仍返回 True**。
所有接线一律紧贴 `_mark_task(success)` **之前**,并加了源码扫描用例钉住
(已反向验证:故意挪到之后,用例立刻报出 file:line)。

## 五、`douyu_tasks` 保留策略(新增作业,默认关闭)

`jobs/douyu_tasks_retention.py`,两档:

| 档 | 保留 |
| --- | --- |
| 噪声查询(只读看一眼就走,且前端不拿它派生账号状态) | **7 天** |
| 其余一切(业务事实 + 状态派生依赖的查询 + **以后新增的类型**) | **90 天** |

长保留档用 `task_type NOT IN (噪声名单)` 表达 → **新增类型默认长保留**,不会因为
「忘了加名单」被误删(正是收集器翻车的反面)。在途任务(planned/pending/running)
**绝不删**;分批删避免长锁;两个窗口都能在作业 `config` 里改,不必发版。

⚠️ 「前端从任务列表反推账号当前状态」那几类查询(`query_xpd_bind_info` /
`query_esports_game_name` 等)**故意留在长保留档** —— 删早了工作台账号行会变
「未查」,那是把**当前显示**搞坏。根治要靠 §3.6 的状态列拆解(后续阶段)。
作业默认 `enabled=False`,由运维决定何时打开。

## 验证

- `TZ=Asia/Shanghai`:**450 passed / 6 skipped**(S4 后 426 → 净增 24)
- `TZ=UTC`(相关文件)101 passed;ruff check/format 全绿
- 前端 build 通过、lint 0 errors
- 真实开发库:迁移已应用(head `20261004_0043`,孤儿作业行已清);
  保留作业实跑确认「作业行默认关闭、SQL 在 MySQL 上正确、近期行不误删」
- **两处反向验证**:把 `record_bind` 挪到 `_mark_task` 之后 → 顺序用例报红;
  修掉 `registered_jobs()` 那个**假绿用例**(它返回 dict,原来的集合推导式
  遍历空字典恒成立),并在断言里加了反向确认

未做(后续):S6 账号页重构 + `bind_status` 三列拆解(做完才能把状态派生类查询
也纳入短保留)、S7 账号检测改账号维度、S8 丢 `login_tasks` 旧凭据列。
2026-10-04 16:00:42 +08:00

131 lines
5.6 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_0043"
# 迁移新增、但不声明在模型里的复合查询索引。
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()
for name, table in model_tables.items():
db_columns = {column["name"] for column in inspector.get_columns(name)}
model_columns = set(table.columns.keys())
missing_columns = model_columns - db_columns
assert not missing_columns
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
# 迁移新增的复合查询索引(不入模型,但应存在)。
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}`;"))