shangfangjian
49d78136a1
去掉 uq_standings_league_season_team,改为 (league, season, team, available_at) 唯一; 每次采集 INSERT 新行(available_at=now),ON CONFLICT DO NOTHING,不覆盖旧行。 standings_slice(before):DISTINCT ON (team_id) WHERE available_at<=cutoff ORDER available_at DESC;before=None → cutoff=now()。 公开 list_standings 取每队最新可用快照(子查询 max available_at)。 迁移 0023 + 切片/路由/docs 同步;测试 test_p0_standings_cutoff(5/5)。 284 测试全绿。
139 lines
5.0 KiB
Python
139 lines
5.0 KiB
Python
"""P0-02 回归测试: 积分榜改为追加快照(append-only) + available_at cutoff。
|
|
|
|
运行: pytest tests/test_p0_standings_cutoff.py -v
|
|
(模型约束用 fake DB;cutoff 过滤语义用 fake session 验证参数传递。)
|
|
"""
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from src.db.models import League, Standing, Team
|
|
|
|
|
|
# ── fake DB(对齐现有测试约定) ────────────────────────────────────
|
|
class _FakeResult:
|
|
def __init__(self, items): self._items = list(items)
|
|
def scalars(self):
|
|
class _S:
|
|
def __init__(self, items): self._items = items
|
|
def all(self): return list(self._items)
|
|
return _S(self._items)
|
|
def scalar_one_or_none(self):
|
|
return self._items[0] if self._items else None
|
|
|
|
|
|
class _FakeDB:
|
|
captured: list = []
|
|
|
|
def __init__(self, league=None, standing_rows=None):
|
|
self._league = league
|
|
self._rows = standing_rows or []
|
|
_FakeDB.captured = []
|
|
|
|
def add(self, obj):
|
|
_FakeDB.captured.append(obj)
|
|
|
|
async def execute(self, stmt):
|
|
# 记录生成的 SQL(字符串化)供断言
|
|
_FakeDB.captured.append(str(stmt))
|
|
compiled = str(stmt)
|
|
if "league" in compiled.lower() and "standing" not in compiled.lower():
|
|
return _FakeResult([self._league] if self._league else [])
|
|
return _FakeResult(self._rows)
|
|
|
|
async def flush(self):
|
|
pass
|
|
|
|
|
|
class _FakeTeam:
|
|
def __init__(self, tid, name):
|
|
self.id = tid
|
|
self.name = name
|
|
self.name_zh = None
|
|
|
|
|
|
class _Header:
|
|
def __init__(self):
|
|
from src.llm.slices.common import MatchHeader
|
|
self._h = MatchHeader(
|
|
match_id=1, home_name="A", away_name="B", league_name="E0",
|
|
season="2026", match_date="2026-01-01", match_dt=None,
|
|
stage=None, home_team_id=10, away_team_id=20, league_id=1,
|
|
)
|
|
|
|
def __getattr__(self, name):
|
|
return getattr(self._h, name)
|
|
|
|
|
|
class TestStandingsModel:
|
|
"""standings 模型必须有 available_at + 新唯一约束。"""
|
|
|
|
def test_available_at_column(self):
|
|
cols = {c.name for c in Standing.__table__.columns}
|
|
assert "available_at" in cols
|
|
|
|
def test_unique_constraint_includes_available_at(self):
|
|
names = {c.name for c in Standing.__table__.constraints}
|
|
assert any("available" in n and n.startswith("uq_") for n in names), \
|
|
f"缺少含 available_at 的唯一约束,现有: {names}"
|
|
|
|
def test_old_unique_constraint_removed(self):
|
|
names = {c.name for c in Standing.__table__.constraints}
|
|
assert "uq_standings_league_season_team" not in names, \
|
|
"旧约束 uq_standings_league_season_team 应已被替换"
|
|
|
|
|
|
class TestStandingsSliceCutoff:
|
|
"""standings_slice 必须尊重 before(cutoff):before=None → now()。"""
|
|
|
|
def test_before_none_uses_now(self):
|
|
"""before=None 时应将 cutoff 视为 now()(取最新可用快照)。"""
|
|
from src.llm.slices import standings as st_mod
|
|
from datetime import datetime, timezone
|
|
|
|
calls = {}
|
|
real_execute = None
|
|
|
|
class _DB:
|
|
def __init__(self): self._league = League(id=1, code="E0", name="E0")
|
|
def add(self, obj): pass
|
|
async def execute(self, stmt):
|
|
# 捕获 WHERE available_at <= ? 的参数
|
|
sql = str(stmt)
|
|
if "available_at" in sql:
|
|
# 提取编译后的 params
|
|
try:
|
|
params = stmt.compile().params
|
|
calls["cutoff"] = params.get("available_at_1")
|
|
except Exception:
|
|
pass
|
|
if "league" in sql.lower() and "standing" not in sql.lower():
|
|
return _FakeResult([self._league])
|
|
return _FakeResult([])
|
|
async def flush(self): pass
|
|
|
|
async def run():
|
|
db = _DB()
|
|
before = None
|
|
await st_mod.standings_slice(_Header(), before=before, db=db)
|
|
|
|
import asyncio
|
|
asyncio.run(run())
|
|
# before=None 时应注入 now() 作为 cutoff
|
|
assert "cutoff" in calls, "未对 available_at 施加 cutoff 过滤"
|
|
assert calls["cutoff"] is not None
|
|
|
|
|
|
class TestStandingsAppendOnly:
|
|
"""采集应 INSERT 新行(带 available_at),不覆盖旧行。"""
|
|
|
|
def test_values_include_available_at(self):
|
|
"""采集构造的 Standing 必须含 available_at 字段。"""
|
|
from src.data import bzzoiro_standings as bzs
|
|
# 检查函数源码是否包含 available_at(编译期守卫)
|
|
import inspect
|
|
src = inspect.getsource(bzs.ingest_bzzoiro_standings)
|
|
assert "available_at" in src, "采集函数必须设置 available_at"
|
|
# 不应再出现按 (league, season, team) 的 upsert 查询
|
|
assert "scalar_one_or_none" not in src or "Standing.league_id == league.id" not in src.replace("available_at", ""), \
|
|
"不应再按 (league, season, team) 做 upsert 查询"
|