test(P1-F): BIGINT 主键表 INSERT 不带 id 自动生成验证

真 PG 测试(独立 engine,避免事件循环污染):
- data_lineage / raw_event 未指定 id → DB 自动生成递增唯一 id
- 连续 INSERT id 递增且唯一

3/3 通过。

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 应更大(递增)"