Profeto/tests/test_public_readonly_api.py

"""公开只读 API 回归(P1-2 / P1-3):context 与 leagues 匿名可访问。

背景:
  - 公开站 MatchDetailSection 展开详情会请求 /matches/{id}/context,
    此前该端点挂 require_admin,未登录 401 被 fetchMatchContext 的
    catch 吞掉 → 近况/交锋静默为空;
  - 公开站联赛筛选需要 /leagues,此前同样 require_admin,前端写死五大联赛。

守卫(双保险):
  1. 功能层:最小 FastAPI app + dependency_overrides 注入 fake session,
     匿名请求(无任何凭据)→ 200;不存在的 match → 404。
     (不启动完整 app lifespan,遵循 test_api_critical.py 的既定约束)
  2. 鉴权层:检查路由依赖声明,require_admin 不得出现在
     /leagues 与 /matches/{match_id}/context。dev 环境 require_admin
     未配置鉴权时 fail-open,功能层测不出「加回了 require_admin」的回归,
     必须靠本层声明检查;并用 ingest 路由证明检查器本身有判别力
     (变异保护:若有人给公开端点加回 require_admin,此处变红)。
"""
from __future__ import annotations

from datetime import datetime, timezone

from fastapi import FastAPI
from fastapi.testclient import TestClient

from src.api.deps import require_admin
from src.api.routes.ingest import router as ingest_router
from src.api.routes.matches import router as matches_router
from src.db.base import get_db_read
from src.db.models import League, Match


# ── fake DB(对齐 routes/matches.py 的实际查询面) ──────────────────

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:
    """match_context 的查询次序:① match 主查询(scalar_one_or_none)
    ② home_recent ③ away_recent ④ h2h(均 scalars().all())。
    League 查询按实体识别直接返回列表(对应 /leagues)。"""

    def __init__(self, match=None, match_lists=(), leagues=()):
        self._match = match
        self._match_lists = list(match_lists)
        self._leagues = list(leagues)
        self._calls = 0

    async def execute(self, stmt):
        entity = stmt.column_descriptions[0]["entity"]
        if entity is League:
            return _FakeResult(self._leagues)
        if self._calls == 0:
            self._calls += 1
            return _FakeResult([self._match] if self._match is not None else [])
        idx = self._calls - 1
        self._calls += 1
        return _FakeResult(self._match_lists[idx] if idx < len(self._match_lists) else [])


def _client(fake_db: _FakeDB) -> TestClient:
    app = FastAPI()
    app.include_router(matches_router)
    app.dependency_overrides[get_db_read] = lambda: fake_db
    # 不用 with:不触发 lifespan,无真实 DB 引擎连接
    return TestClient(app)


def _league(lid: int, code: str) -> League:
    return League(id=lid, code=code, name=f"League {code}", country=f"Country {code}")


def _match(mid: int) -> Match:
    return Match(
        id=mid,
        match_date=datetime(2026, 9, 20, 15, 0, tzinfo=timezone.utc),
        home_goals=2,
        away_goals=1,
    )


# ── 功能层:匿名可访问(P1-2 / P1-3) ──────────────────────────────

def test_leagues_anonymous_200_and_shape():
    """/leagues 匿名 200;仅暴露 id/code/name/country 四字段。"""
    db = _FakeDB(leagues=[_league(1, "E0"), _league(2, "SP1")])
    r = _client(db).get("/api/v1/leagues")
    assert r.status_code == 200
    body = r.json()
    assert body == [
        {"id": 1, "code": "E0", "name": "League E0", "country": "Country E0"},
        {"id": 2, "code": "SP1", "name": "League SP1", "country": "Country SP1"},
    ]
    # 不暴露敏感配置字段
    assert all(set(item.keys()) == {"id", "code", "name", "country"} for item in body)


def test_context_anonymous_200_empty_data():
    """/context 匿名 200;无数据时三个列表为空(前端空态),结构不变。"""
    db = _FakeDB(match=_match(42), match_lists=[[], [], []])
    r = _client(db).get("/api/v1/matches/42/context")
    assert r.status_code == 200
    assert r.json() == {"home_recent": [], "away_recent": [], "h2h": []}


def test_context_anonymous_200_row_shape():
    """/context 行结构与既有前端契约一致(5 字段)。"""
    db = _FakeDB(match=_match(42), match_lists=[[_match(1)], [], [_match(2), _match(3)]])
    r = _client(db).get("/api/v1/matches/42/context")
    assert r.status_code == 200
    body = r.json()
    assert len(body["home_recent"]) == 1
    assert len(body["h2h"]) == 2
    assert set(body["h2h"][0].keys()) == {
        "match_date", "home_team", "away_team", "home_goals", "away_goals",
    }


def test_context_not_found_404():
    """/context 匿名访问不存在的 match → 404(而非 401/503)。"""
    db = _FakeDB(match=None)
    r = _client(db).get("/api/v1/matches/999/context")
    assert r.status_code == 404


# ── 鉴权层:路由依赖声明检查(防 require_admin 回潜) ────────────────

def _admin_paths(router) -> set[str]:
    paths: set[str] = set()
    for route in router.routes:
        for dep in route.dependant.dependencies:
            if dep.call is require_admin:
                paths.add(route.path)
                break
    return paths


def test_public_routes_have_no_admin_dependency():
    """/leagues 与 /context 不得挂 require_admin;matches 路由全部公开只读。"""
    admin_paths = _admin_paths(matches_router)
    assert "/api/v1/leagues" not in admin_paths
    assert "/api/v1/matches/{match_id}/context" not in admin_paths
    assert admin_paths == set(), f"matches 路由应全部公开只读,仍有 {admin_paths}"


def test_guard_detector_has_discrimination_power():
    """变异保护:检查器必须能在 ingest 路由上发现 require_admin,
    否则上一条「无 admin 依赖」断言恒真、毫无判别力。"""
    assert "/api/v1/ingest/bzzoiro" in _admin_paths(ingest_router), (
        "ingest 路由应仍存在 require_admin 保护;若本断言失败,"
        "说明公开只读守卫的检查器已失效,请修复检查逻辑"
    )