fix(quality): 质量检查引用已删除的 MatchStats.id 必然 500 + 检查无日志

生产实测: POST /admin/data-quality/run 100% 失败,且日志无任何记录。

根因链:
1. P0-02 把 MatchStats 主键改为 match_id(无 id 属性),但质量检查
   查询仍写 MatchStats.id.is_(None) → AttributeError → 500
2. handler 无 logger 调用;未捕获异常走 uvicorn.error(默认
   propagate=False),不经过 root handler → 内存日志页/文件日志
   都看不到 traceback,排障无据可查

修复:
- MatchStats.id → MatchStats.match_id(与 P0-02 模型对齐)
- 成功路径记 info 日志(检查名=通过/未通过(计数))
- setup_logging 打开 uvicorn/uvicorn.error 的 propagate,
  未捕获异常 traceback 进入内存缓冲与滚动文件
- 回归测试 3 例(mock session):修复前红(3 failed),修复后绿;
  全量 329 passed 8 skipped 零回归
This commit is contained in:
WorkBuddy
2026-09-22 18:03:20 +08:00
parent f00bf71e8f
commit 0742eef52e
3 changed files with 83 additions and 1 deletions
+66
View File
@@ -0,0 +1,66 @@
"""数据质量检查回归测试。
背景(P0-02 遗留): MatchStats 主键改为 match_id 后,质量检查查询仍引用
MatchStats.id → AttributeError → POST /admin/data-quality/run 必然 500,
前端显示「运行失败」;且 handler 无日志,未捕获异常走 uvicorn.error
(propagate=False)不进内存缓冲/文件日志,排障时无据可查。
"""
from __future__ import annotations
from unittest.mock import AsyncMock, MagicMock
from src.api.routes import admin_quality
def _fake_db(scalars: list[int]) -> MagicMock:
"""按顺序返回 scalar() 计数的假 AsyncSession。"""
db = MagicMock()
results = []
for v in scalars:
r = MagicMock()
r.scalar.return_value = v
results.append(r)
db.execute = AsyncMock(side_effect=results)
db.add = MagicMock()
db.commit = AsyncMock()
return db
async def test_run_data_quality_check_no_attribute_error():
"""检查查询不得引用 MatchStats.id(P0-02 后该属性不存在)。
修复前: run_data_quality_check 抛 AttributeError → 500。
"""
db = _fake_db([3, 1])
out = await admin_quality.run_data_quality_check(db)
assert out["ok"] is True
assert out["checks"] == [
{"name": "finished_without_stats", "passed": False},
{"name": "league_without_standings", "passed": False},
]
# 检查结果落库(2 条 DataQualityCheck)
assert db.add.call_count == 2
db.commit.assert_awaited_once()
async def test_run_data_quality_check_all_passed():
db = _fake_db([0, 0])
out = await admin_quality.run_data_quality_check(db)
assert out["ok"] is True
assert all(c["passed"] for c in out["checks"])
async def test_run_data_quality_check_logs_summary(caplog):
"""成功路径必须留日志:否则线上无从得知检查何时跑过、结果如何。"""
import logging
db = _fake_db([0, 0])
with caplog.at_level(logging.INFO, logger="src.api.routes.admin_quality"):
await admin_quality.run_data_quality_check(db)
assert any("数据质量检查" in r.message for r in caplog.records)