190 lines
7.4 KiB
Python
190 lines
7.4 KiB
Python
"""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)"
|