modify: 测试

test/test_auth_race.py

@@ -0,0 +1,65 @@
+"""
+并发启动竞态测试：模拟 gunicorn 多 worker 同时启动时，
+多个进程并发调用 plugins.auth.create_workers 初始化默认管理员。
+
+预期：不抛 IntegrityError，users 表中只有一条 admin 记录。
+"""
+import asyncio
+import sys
+import tempfile
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent / "src"))
+
+import plugins.auth as auth_plugin
+
+
+def fake_create_workers(session, config, count=1, worker_manager=None):
+    """替换真实 worker 启动，避免依赖 zenoh 会话。与原始实现同为同步函数。"""
+    return []
+
+
+async def main():
+    # 替换 Plugin 基类的 worker 创建实现，避免依赖 zenoh 会话
+    auth_plugin.plugin._create_workers_impl = fake_create_workers
+
+    with tempfile.TemporaryDirectory() as tmp:
+        config = {"type": "auth", "data_dir": tmp}
+
+        # 模拟 4 个 gunicorn worker 进程同时启动
+        results = await asyncio.gather(
+            auth_plugin.create_workers(None, config, 3),
+            auth_plugin.create_workers(None, config, 3),
+            auth_plugin.create_workers(None, config, 3),
+            auth_plugin.create_workers(None, config, 3),
+            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_database import Database
+        from plugins.auth.models.auth import Users
+        from sqlalchemy import select
+
+        db = Database(Path(tmp) / "auth.db")
+        await db.initialize()
+        async with db.get_session() as sess:
+            res = await sess.execute(select(Users))
+            rows = res.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)} 条"
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
+    print("🎉 竞态测试通过")