shangfangjian
eae88f4cd9
P1-G runtime_config:DB 故障回落 env;解密失败在生产环境抛出(ValueError), 非生产回落;DB 异常与解密异常不再共用 except Exception。 P1-H Redis 预测缓存:去掉 pickle 改用 JSON + dataclasses.asdict, datetime 字段 ISO 序列化。 P1-I eval get_eval_summary:默认 limit=None(不冒充全集);新增 run_type(默认 live) 与 season 过滤;_build_filters 同步扩展。 P1-K Team/League get_or_create:INSERT 改为 PG UPSERT(ON CONFLICT DO NOTHING) 防并发重复插入。 测试 test_p1_g_runtime_config(4/4) + test_p1_k_upsert(4/4);全量绿。
101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
"""P1-G 回归测试: runtime_config DB 回落 env + 解密失败生产环境抛出。
|
|
|
|
运行: pytest tests/test_p1_g_runtime_config.py -v
|
|
(纯函数测试,mock session,无真实 PG 依赖。)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from src.core.config import settings
|
|
import src.core.runtime_config as rc
|
|
|
|
|
|
# ── 辅助:构造模拟 session ─────────────────────────────────────────
|
|
class _FakeRow:
|
|
def __init__(self, value): self.value = value
|
|
|
|
|
|
class _FakeSession:
|
|
def __init__(self, row=None, raise_on_get=None):
|
|
self._row = row
|
|
self._raise = raise_on_get
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
return None
|
|
|
|
async def get(self, cls, key):
|
|
if self._raise:
|
|
raise self._raise
|
|
return self._row
|
|
|
|
|
|
class _FakeSessionLocal:
|
|
def __init__(self, row=None, raise_on_get=None):
|
|
self._row = row
|
|
self._raise = raise_on_get
|
|
|
|
def __call__(self):
|
|
return _FakeSession(self._row, self._raise)
|
|
|
|
|
|
# ── 测试 ──────────────────────────────────────────────────────────
|
|
class TestDBFallback:
|
|
"""P1-G:DB 故障时回落环境变量。"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_db_error_falls_back_to_env(self):
|
|
"""DB 抛异常 → 回落 settings 同名属性。"""
|
|
# 模拟 DB 连接失败
|
|
fake = _FakeSessionLocal(raise_on_get=RuntimeError("DB down"))
|
|
with patch.object(rc, "AsyncSessionLocal", fake):
|
|
val = await rc.get_runtime_value("LLM_MODEL")
|
|
# 应回落 settings.LLM_MODEL(有默认值 gpt-4o)
|
|
assert val == settings.LLM_MODEL
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_db_returns_none_falls_back_to_env(self):
|
|
"""DB 行不存在 → 回落 env。"""
|
|
fake = _FakeSessionLocal(row=None)
|
|
with patch.object(rc, "AsyncSessionLocal", fake):
|
|
val = await rc.get_runtime_value("LLM_MODEL")
|
|
assert val == settings.LLM_MODEL
|
|
|
|
|
|
class TestDecryptFailure:
|
|
"""P1-G:解密失败在生产环境必须抛出,不得被 except Exception 吞掉。"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decrypt_failure_raises_in_production(self):
|
|
"""P1-G:production + 解密失败 → 必须 raise ValueError(不得被吞)。"""
|
|
from src.core import crypto
|
|
|
|
# 模拟 DB 返回一个加密值,但解密会失败
|
|
fake = _FakeSessionLocal(row=_FakeRow("enc:v1:corrupted_token"))
|
|
|
|
with patch.object(rc, "AsyncSessionLocal", fake), \
|
|
patch.object(crypto, "decrypt_value", side_effect=ValueError("解密失败")), \
|
|
patch.object(settings, "APP_ENV", "production"):
|
|
# P1-G:生产环境解密失败必须抛出,不得静默回落
|
|
with pytest.raises(ValueError, match="解密失败"):
|
|
await rc.get_setting_origin("LLM_API_KEY")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_decrypt_failure_falls_back_in_non_production(self):
|
|
"""非生产环境 + 解密失败 → 回落 env,不抛错。"""
|
|
from src.core import crypto
|
|
|
|
fake = _FakeSessionLocal(row=_FakeRow("enc:v1:corrupted_token"))
|
|
|
|
with patch.object(rc, "AsyncSessionLocal", fake), \
|
|
patch.object(crypto, "decrypt_value", side_effect=ValueError("解密失败")), \
|
|
patch.object(settings, "APP_ENV", "development"):
|
|
origin, val = await rc.get_setting_origin("LLM_API_KEY")
|
|
# 非生产 → 回落 env(origin=env 或 none)
|
|
assert origin in ("env", "none")
|