diff --git a/tests/test_p1_f_bigint_identity.py b/tests/test_p1_f_bigint_identity.py new file mode 100644 index 0000000..c0c5c37 --- /dev/null +++ b/tests/test_p1_f_bigint_identity.py @@ -0,0 +1,97 @@ +"""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 应更大(递增)"