Profeto/tests/test_api_critical.py

"""FastAPI 关键路径测试:鉴权、限流、游标方向。

运行(需先 pip-sync requirements-dev.txt):
  pytest tests/test_api_critical.py -v

设计:
  - 鉴权:直接测 require_admin / auth_configured 逻辑,monkeypatch 切换环境,
    避免启动完整 app lifespan(异步 DB 引擎与同步 TestClient 不兼容)。
  - 限流:直接测 _RateLimiter 单元。
  - 游标:直接构造 SQL 验证 scheduled ASC / 其他 DESC 方向。
  - 不依赖真实 LLM / 数据库:纯逻辑测试。
"""
from __future__ import annotations

import asyncio
from unittest.mock import AsyncMock, patch

import pytest

from src.api import deps
from src.api.deps import (
    _RateLimiter,
    auth_configured,
    require_admin,
)
from src.core import runtime_config


# ── 1. 鉴权:fail-closed(生产) vs fail-open(开发) ─────────────────────


class TestAuthFailClosed:
    """未配置鉴权策略时,production 环境应拒绝(503),development 应放行。"""

    @pytest.mark.asyncio
    async def test_production_no_auth_returns_503(self):
        """APP_ENV=production + 未配置任何鉴权 → require_admin 抛 503。"""
        from fastapi import HTTPException
        from fastapi.requests import Request

        request = Request(scope={"type": "http", "client": ("1.2.3.4", 0), "cookies": {}, "headers": []})

        with patch.object(deps, "settings") as s, \
             patch("src.api.deps.auth_configured", AsyncMock(return_value=False)):
            s.REQUIRE_ADMIN_AUTH = False
            s.APP_ENV = "production"
            s.ADMIN_API_KEY = ""
            s.ADMIN_PASSWORD = ""
            with pytest.raises(HTTPException) as exc:
                await require_admin(request, x_api_key=None)
            assert exc.value.status_code == 503

    @pytest.mark.asyncio
    async def test_development_no_auth_passes(self):
        """APP_ENV=development + 未配置鉴权 → 放行(只打 warning)。"""
        from fastapi.requests import Request

        request = Request(scope={"type": "http", "client": ("1.2.3.4", 0), "cookies": {}, "headers": []})

        with patch.object(deps, "settings") as s, \
             patch("src.api.deps.auth_configured", AsyncMock(return_value=False)):
            s.REQUIRE_ADMIN_AUTH = False
            s.APP_ENV = "development"
            s.ADMIN_API_KEY = ""
            s.ADMIN_PASSWORD = ""
            # 不应抛异常
            await require_admin(request, x_api_key=None)

    @pytest.mark.asyncio
    async def test_require_admin_key_valid(self):
        """配置 ADMIN_API Key 后,带正确 X-API-Key 头 → 通过。"""
        from fastapi.requests import Request

        request = Request(scope={"type": "http", "client": ("1.2.3.4", 0), "cookies": {},
                          "headers": [(b"x-api-key", b"test-secret-key")]})

        with patch.object(deps, "settings") as s, \
             patch("src.core.runtime_config.get_admin_password_hash", AsyncMock(return_value="")):
            s.REQUIRE_ADMIN_AUTH = True
            s.APP_ENV = "production"
            s.ADMIN_API_KEY = "test-secret-key"
            # 不应抛异常
            await require_admin(request, x_api_key="test-secret-key")

    @pytest.mark.asyncio
    async def test_require_admin_key_invalid(self):
        """API Key 错误 → 401。"""
        from fastapi import HTTPException
        from fastapi.requests import Request

        request = Request(scope={"type": "http", "client": ("1.2.3.4", 0), "cookies": {},
                          "headers": [(b"x-api-key", b"wrong")]})

        with patch.object(deps, "settings") as s, \
             patch("src.core.runtime_config.get_admin_password_hash", AsyncMock(return_value="")):
            s.REQUIRE_ADMIN_AUTH = True
            s.APP_ENV = "production"
            s.ADMIN_API_KEY = "test-secret-key"
            with pytest.raises(HTTPException) as exc:
                await require_admin(request, x_api_key="wrong")
            assert exc.value.status_code == 401


# ── 2. 限流:滑动窗口逻辑 ─────────────────────────────────────────────


class TestRateLimit:
    """预测限流:_RateLimiter 单元测试。"""

    def test_rate_limit_triggers_after_max(self):
        """max=2/60s → 第 3 次被拒。"""
        limiter = _RateLimiter(max_requests=2, window_seconds=60)
        ip = "1.2.3.4"
        assert limiter.is_allowed(ip) is True
        assert limiter.is_allowed(ip) is True
        assert limiter.is_allowed(ip) is False  # 超限
        assert limiter.remaining(ip) == 0

    def test_rate_limit_remaining_decrements(self):
        """剩余配额计算准确。"""
        limiter = _RateLimiter(max_requests=5, window_seconds=60)
        ip = "5.6.7.8"
        assert limiter.remaining(ip) == 5
        limiter.is_allowed(ip)
        limiter.is_allowed(ip)
        assert limiter.remaining(ip) == 3

    def test_rate_limit_per_ip_isolated(self):
        """不同 IP 独立计数。"""
        limiter = _RateLimiter(max_requests=1, window_seconds=60)
        assert limiter.is_allowed("1.1.1.1") is True
        assert limiter.is_allowed("1.1.1.1") is False  # 同 IP 超限
        assert limiter.is_allowed("2.2.2.2") is True   # 不同 IP 不受影响


# ── 3. 游标方向:scheduled ASC 用 > ───────────────────────────────────


class TestCursorDirection:
    """验证 scheduled 状态查询时排序方向为 ASC(使用 > 游标)。"""

    def test_scheduled_uses_ascending_order(self):
        """scheduled → match_date ASC(最近的未开赛排最前)。"""
        from src.api.routes import matches as matches_mod
        from sqlalchemy import select

        q = select(matches_mod.Match).where(matches_mod.Match.match_status == "scheduled")
        q = q.order_by(matches_mod.Match.match_date.asc(), matches_mod.Match.id.asc())
        sql = str(q)
        assert "ORDER BY matches.match_date ASC" in sql, sql

    def test_finished_uses_descending_order(self):
        """finished/其他 → DESC(最新赛果在前)。"""
        from src.api.routes import matches as matches_mod
        from sqlalchemy import select

        q = select(matches_mod.Match).where(matches_mod.Match.match_status == "finished")
        q = q.order_by(matches_mod.Match.match_date.desc(), matches_mod.Match.id.desc())
        sql = str(q)
        assert "ORDER BY matches.match_date DESC" in sql, sql


# ── 评估置信度校准分桶 ─────────────────────────────────────────────




# ── 评估置信度校准分桶 ─────────────────────────────────────────────


class TestEvalCalibration:
    """验证 settled 预测按置信度分桶统计命中率。"""

    def test_confidence_bucketing(self):
        """置信度落入正确的桶。"""
        assert _bucket_key(0.3) == "low(0-0.5)"
        assert _bucket_key(0.5) == "medium(0.5-0.7)"
        assert _bucket_key(0.6) == "medium(0.5-0.7)"
        assert _bucket_key(0.7) == "high(0.7-1)"
        assert _bucket_key(0.95) == "high(0.7-1)"


def _bucket_key(conf: float) -> str:
    if conf < 0.5:
        return "low(0-0.5)"
    elif conf < 0.7:
        return "medium(0.5-0.7)"
    else:
        return "high(0.7-1)"