From 2b52478b8fb2b070f26add33a54bd1cb8fdabe40 Mon Sep 17 00:00:00 2001
From: shangfangjian <shangfangjian@bilidili.cn>
Date: Tue, 22 Sep 2026 08:57:20 +0800
Subject: [PATCH] =?UTF-8?q?test(P1-F):=20BIGINT=20=E4=B8=BB=E9=94=AE?=
 =?UTF-8?q?=E8=A1=A8=20INSERT=20=E4=B8=8D=E5=B8=A6=20id=20=E8=87=AA?=
 =?UTF-8?q?=E5=8A=A8=E7=94=9F=E6=88=90=E9=AA=8C=E8=AF=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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

3/3 通过。
---
 tests/test_p1_f_bigint_identity.py | 97 ++++++++++++++++++++++++++++++
 1 file changed, 97 insertions(+)
 create mode 100644 tests/test_p1_f_bigint_identity.py

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