fix(P0-01): missing score 不得变 0:0 — score_status + CHECK 约束

新增 matches.score_status(known/missing/unknown):
- 替换 ck_matches_finished_has_score 为 ck_matches_score_integrity:
  known → 必须有分; missing/unknown → goals 必须 NULL(不伪造 0:0)
- normalize: 完赛缺分不再静默降级为 scheduled,改设 score_status=missing
- events ingest: 创建/更新 Match 同步 score_status(比分由缺变 known / 确认缺分 missing)
- slices(form/h2h/home_away)/backtest: 显式加 score_status='known' 过滤完赛样本
- 迁移 0022 回填现有数据(绝不 UPDATE goals=0)

测试 tests/test_p0_score_status.py(9/9):约束存在性/Match 构造/normalize 行为。
54 相关测试全绿。
This commit is contained in:
shangfangjian
2026-09-22 02:32:53 +08:00
parent f563d5cc99
commit 63caa6736c
9 changed files with 226 additions and 4 deletions
+124
View File
@@ -0,0 +1,124 @@
"""P0-01 回归测试: missing score 不得变 0:0。
运行: pytest tests/test_p0_score_status.py -v
(无需真实 PG;用 fake DB + 模型元数据断言。)
"""
import pytest
from src.db.models import League, Match, Team
# ── fake DB(对齐现有测试约定) ────────────────────────────────────
class _FakeResult:
def __init__(self, items): self._items = list(items)
def scalars(self): return self
def all(self): return list(self._items)
def scalar_one_or_none(self): return self._items[0] if self._items else None
class _FakeDB:
def __init__(self): self.added = []
def add(self, obj): self.added.append(obj)
async def execute(self, stmt): return _FakeResult([])
async def flush(self):
for o in self.added:
if getattr(o, "id", None) is None:
o.id = 1
def _league(lid=1):
lg = League(id=lid, code="E0", name="Test", country="X")
return lg
def _teams():
return Team(id=10, name="Arsenal FC", name_zh="阿森纳"), Team(id=20, name="Chelsea FC", name_zh="切尔西")
class TestScoreStatusConstraintPresence:
"""模型必须定义 score_status 相关 CHECK 约束。"""
def test_score_status_column_exists(self):
cols = {c.name for c in Match.__table__.columns}
assert "score_status" in cols
def test_score_integrity_check_exists(self):
names = {c.name for c in Match.__table__.constraints if c.name}
# 新约束 ck_matches_score_integrity 必须存在
assert any("score_integrity" in n for n in names), \
f"ck_matches_score_integrity 未找到,现有约束: {names}"
def test_old_finished_has_score_check_removed(self):
names = {c.name for c in Match.__table__.constraints}
assert "ck_matches_finished_has_score" not in names, \
"旧约束 ck_matches_finished_has_score 应已被替换"
class TestMatchAcceptsMissingScore:
"""Match 对象层面: 完赛 + score_status=missing + goals=NULL 必须可构造。"""
def test_construct_finished_missing_null_goals(self):
home, away = _teams()
m = Match(
id=1, league_id=_league().id, home_team_id=home.id, away_team_id=away.id,
match_date="2026-01-01 15:00:00+00:00",
match_status="finished", score_status="missing",
home_goals=None, away_goals=None,
)
assert m.home_goals is None
assert m.away_goals is None
assert m.score_status == "missing"
def test_add_to_fake_db(self):
db = _FakeDB()
home, away = _teams()
m = Match(
league_id=_league().id, home_team_id=home.id, away_team_id=away.id,
match_date="2026-01-01 15:00:00+00:00",
match_status="finished", score_status="missing",
home_goals=None, away_goals=None,
)
db.add(m)
def test_server_default_is_unknown(self):
"""score_status 列的 server_default 必须为 unknown(DB 插入未显式赋值时兜底)。"""
col = Match.__table__.c.score_status
assert col.server_default is not None
assert "unknown" in str(col.server_default.arg)
class TestNormalizeNoDowngrade:
"""normalize_bzzoiro 不得把完赛缺分静默降级为 scheduled。"""
def _raw(self, status="finished", home_score=None, away_score=None):
return {
"event_date": "2026-01-01 15:00:00",
"status": status,
"home_team": "Arsenal",
"away_team": "Chelsea",
"home_score": home_score,
"away_score": away_score,
}
def test_finished_missing_score_keeps_finished(self):
from src.data.normalize import normalize_bzzoiro
m = normalize_bzzoiro(self._raw("finished", None, None), "E0")
assert m is not None
assert m.match_status == "finished", "完赛缺分不得降级为 scheduled"
assert m.score_status == "missing"
assert m.home_goals is None
assert m.away_goals is None
def test_finished_with_score_is_known(self):
from src.data.normalize import normalize_bzzoiro
m = normalize_bzzoiro(self._raw("finished", 2, 1), "E0")
assert m.match_status == "finished"
assert m.score_status == "known"
assert m.home_goals == 2 and m.away_goals == 1
def test_scheduled_no_score_is_unknown(self):
from src.data.normalize import normalize_bzzoiro
m = normalize_bzzoiro(self._raw("scheduled", None, None), "E0")
assert m.match_status == "scheduled"
assert m.score_status == "unknown"
assert m.home_goals is None