chongming-core/test/test_auth_race.py

"""
并发启动竞态测试：模拟 gunicorn 多 worker 同时启动时，
多个进程并发执行网关内置的 admin 种子（chongming_auth.seed_admin）。

auth 域拆分后，admin 种子从 plugins.auth 的 on_init 迁移到网关
lifespan（setup_auth_store → chongming_auth.seed_admin），本测试随之改指向。

预期：不抛 IntegrityError，users 表中只有一条 admin 记录。
"""
import asyncio
import tempfile

from chongming_auth import open_database, seed_admin


async def main():
    with tempfile.TemporaryDirectory() as tmp:
        db = await open_database({"data_dir": tmp})

        # 模拟 4 个 gunicorn worker 进程同时启动
        results = await asyncio.gather(
            *[seed_admin(db) for _ in range(4)],
            return_exceptions=True,
        )

        errors = [r for r in results if isinstance(r, Exception)]
        if errors:
            for e in errors:
                print("❌ 异常:", type(e).__name__, e)
            raise RuntimeError("并发启动存在未捕获异常")

        # 校验 users 表中应只有一条 admin
        from chongming_auth import Users
        from sqlalchemy import select

        async with db.get_session() as sess:
            rows = (await sess.execute(select(Users))).scalars().all()

        print("✅ 4 个并发启动全部成功")
        print(f"✅ users 表记录数: {len(rows)}（应为 1）")
        for u in rows:
            print(f"   - {u.username} <{u.email}> superuser={u.is_superuser}")

        assert len(rows) == 1, f"预期 1 条 admin，实际 {len(rows)} 条"
        assert sum(1 for r in results if r is True) == 1, "应恰好一个并发方完成新建"
        await db.dispose()


if __name__ == "__main__":
    asyncio.run(main())
    print("🎉 竞态测试通过")