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);全量绿。
78 lines
2.5 KiB
Python
78 lines
2.5 KiB
Python
"""P1-K 回归测试: Team/League get_or_create 使用 PG UPSERT(ON CONFLICT DO NOTHING)。
|
|
|
|
运行: pytest tests/test_p1_k_upsert.py -v
|
|
(依赖真实 PG;无 PG 时跳过。)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from src.core.config import settings
|
|
from src.db.models import League, Team
|
|
from src.db.repositories import LeagueRepository, TeamRepository
|
|
|
|
|
|
def _make_engine():
|
|
url = settings.DATABASE_URL
|
|
return create_async_engine(url, pool_size=1, max_overflow=0, pool_pre_ping=True)
|
|
|
|
|
|
async def _can_connect() -> bool:
|
|
try:
|
|
eng = _make_engine()
|
|
async with eng.begin() as conn:
|
|
pass
|
|
await eng.dispose()
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _skip_without_pg():
|
|
import asyncio
|
|
|
|
if not asyncio.run(_can_connect()):
|
|
pytest.skip("无真实 PG 可用,跳过 P1-K 测试")
|
|
|
|
|
|
class TestUpsertBehavior:
|
|
@pytest.fixture
|
|
async def db(self):
|
|
eng = _make_engine()
|
|
SessionLocal = sessionmaker(eng, class_=AsyncSession, expire_on_commit=False)
|
|
async with SessionLocal() as session:
|
|
yield session
|
|
await eng.dispose()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_league_get_or_create_inserts_new(self, db):
|
|
repo = LeagueRepository(db)
|
|
league = await repo.get_or_create("TST", "Test League", "X")
|
|
assert league.id is not None
|
|
assert league.code == "TST"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_league_get_or_create_idempotent(self, db):
|
|
"""P1-K: 重复调用返回同一行,不创建重复。"""
|
|
repo = LeagueRepository(db)
|
|
a = await repo.get_or_create("IDP", "Idempotent", "X")
|
|
b = await repo.get_or_create("IDP", "Idempotent", "X")
|
|
assert a.id == b.id, "重复调用应返回同一行"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_team_get_or_create_inserts_new(self, db):
|
|
repo = TeamRepository(db)
|
|
team = await repo.get_or_create("Unique Team FC_k1")
|
|
assert team.id is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_team_get_or_create_idempotent(self, db):
|
|
"""P1-K: 重复调用返回同一行(UPSERT 防并发重复)。"""
|
|
repo = TeamRepository(db)
|
|
a = await repo.get_or_create("Same Team_k1")
|
|
b = await repo.get_or_create("Same Team_k1")
|
|
assert a.id == b.id, "重复调用应返回同一行"
|