Profeto/tests/test_p0_standings_cutoff.py

"""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}
        # 匿名约束(PK/Check)的 name 是 None,必须先判真值再子串匹配,
        # 否则 any() 撞上 None 直接 TypeError(集合顺序不定 → flaky)
        assert any(n and "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 查询"