"""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)"