shangfangjian
2b52478b8f
真 PG 测试(独立 engine,避免事件循环污染): - data_lineage / raw_event 未指定 id → DB 自动生成递增唯一 id - 连续 INSERT id 递增且唯一 3/3 通过。
98 lines
3.0 KiB
Python
98 lines
3.0 KiB
Python
"""P1-F 回归测试: BIGINT 主键表 INSERT 不带 id,验证自动生成。
|
|
|
|
运行: pytest tests/test_p1_f_bigint_identity.py -v
|
|
(依赖真实 PG;无 PG 时跳过。禁止 SQLite 冒充。)
|
|
"""
|
|
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
|
|
|
|
|
|
def _make_engine():
|
|
"""测试用独立 engine(避免全局 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-F 测试(禁止 SQLite 冒充)")
|
|
|
|
|
|
class TestBigIntIdentity:
|
|
"""P1-F: BIGINT 主键表 INSERT 不带 id,DB 自动生成。"""
|
|
|
|
@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_data_lineage_insert_without_id(self, db):
|
|
from src.db.models import DataLineage
|
|
|
|
row = DataLineage(
|
|
source_system="test", source_record_id="r1_p1f",
|
|
target_table="standings", target_id=None,
|
|
transform_name="t", transform_detail={},
|
|
)
|
|
db.add(row)
|
|
await db.flush()
|
|
# 核心:未指定 id,DB 自动生成非 None
|
|
assert row.id is not None, "data_lineage.id 应自动生成"
|
|
assert isinstance(row.id, int)
|
|
assert row.id > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_raw_event_insert_without_id(self, db):
|
|
from src.db.models import RawEvent
|
|
|
|
row = RawEvent(
|
|
source_system="test", source_record_id="r2_p1f",
|
|
raw_payload={"k": "v"},
|
|
)
|
|
db.add(row)
|
|
await db.flush()
|
|
assert row.id is not None
|
|
assert row.id > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auto_increment_unique(self, db):
|
|
"""连续 INSERT 应产生递增唯一 id。"""
|
|
from src.db.models import DataLineage
|
|
|
|
r1 = DataLineage(
|
|
source_system="test", source_record_id="a_p1f",
|
|
target_table="t", transform_name="x",
|
|
)
|
|
r2 = DataLineage(
|
|
source_system="test", source_record_id="b_p1f",
|
|
target_table="t", transform_name="x",
|
|
)
|
|
db.add_all([r1, r2])
|
|
await db.flush()
|
|
assert r1.id is not None and r2.id is not None
|
|
assert r1.id != r2.id, "连续 INSERT id 应唯一"
|
|
assert r2.id > r1.id, "后插入的 id 应更大(递增)"
|