fix:[FL-193][用户管理 - email 字段 NOT NULL 约束冲突]

修复 users.email 字段 NOT NULL 约束与前端可选表单不一致的问题。 ## 改动内容 1. 添加数据库兼容性检查函数 `_ensure_user_email_nullable()` - 检测 users.email 字段是否有 NOT NULL 约束 - 如果存在约束则自动移除，使字段变为可选 - 仅对 PostgreSQL 数据库生效 2. 在 `init_db()` 中调用该函数 - 确保应用启动时自动应用迁移 - 与现有兼容性检查函数保持一致的模式 3. 添加单元测试 `test_user_email_optional.py` - 验证可以创建不带 email 的用户 - 验证可以创建带 email 的用户 - 验证直接使用 User 模型创建用户时 email 可为 None ## 修复方案采用 Issue 中推荐的方案 1（数据库层面修复）： - 将 email 字段改为可选，与前端表单语义保持一致 - 用户可以选择不填写邮箱 - email 字段保持 UNIQUE 约束，但允许 NULL 值 ## 相关文件 - api/app/core/database.py:523-546 - 新增兼容性检查函数 - api/app/core/database.py:586 - 在 init_db() 中调用 - api/tests/test_user_email_optional.py - 新增单元测试 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com> Co-authored-by: multica-agent <github@multica.ai>
2026-06-18 11:34:22 +08:00
parent fa4834a9e2
commit f3e5640290
2 changed files with 131 additions and 0 deletions
@@ -0,0 +1,104 @@
+"""
+Test that user.email field is optional (nullable).
+
+This test verifies the fix for FL-193 where email field had a NOT NULL
+constraint that conflicted with the optional email field in the frontend.
+"""
+from __future__ import annotations
+
+import os
+import unittest
+
+os.environ.setdefault("DATABASE_URL", "sqlite+pysqlite:///:memory:")
+os.environ.setdefault("MINIO_ENABLED", "false")
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+
+from api.app.core.database import Base
+from api.app.core.security import hash_password
+from api.app.models.user import User
+from api.app.schemas.user import UserCreateRequest
+from api.app.services import user_service
+
+
+class UserEmailOptionalTest(unittest.TestCase):
+    """Test that users can be created without an email address."""
+
+    def setUp(self) -> None:
+        self.engine = create_engine(
+            "sqlite+pysqlite://",
+            connect_args={"check_same_thread": False},
+            poolclass=StaticPool,
+        )
+        self.SessionLocal = sessionmaker(
+            bind=self.engine,
+            autocommit=False,
+            autoflush=False,
+            expire_on_commit=False,
+        )
+        Base.metadata.create_all(bind=self.engine)
+        self.session = self.SessionLocal()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(bind=self.engine)
+        self.engine.dispose()
+
+    def test_create_user_without_email(self) -> None:
+        """Test creating a user with email=None should succeed."""
+        user = user_service.create_user(
+            self.session,
+            UserCreateRequest(
+                user_id="test_user_1",
+                email=None,
+                username="TestUser1",
+                password="password123",
+            ),
+            actor_user_id="system",
+        )
+
+        self.assertIsNotNone(user)
+        self.assertEqual(user.id, "test_user_1")
+        self.assertEqual(user.username, "TestUser1")
+        self.assertIsNone(user.email)
+
+    def test_create_user_with_email(self) -> None:
+        """Test creating a user with email should still work."""
+        user = user_service.create_user(
+            self.session,
+            UserCreateRequest(
+                user_id="test_user_2",
+                email="test2@example.com",
+                username="TestUser2",
+                password="password123",
+            ),
+            actor_user_id="system",
+        )
+
+        self.assertIsNotNone(user)
+        self.assertEqual(user.id, "test_user_2")
+        self.assertEqual(user.username, "TestUser2")
+        self.assertEqual(user.email, "test2@example.com")
+
+    def test_create_user_directly_without_email(self) -> None:
+        """Test creating a User model directly without email."""
+        user = User(
+            id="direct_user",
+            email=None,
+            username="DirectUser",
+            password_hash=hash_password("password123"),
+            status="ENABLED",
+        )
+        self.session.add(user)
+        self.session.commit()
+        self.session.refresh(user)
+
+        self.assertEqual(user.id, "direct_user")
+        self.assertEqual(user.username, "DirectUser")
+        self.assertIsNone(user.email)
+
+
+if __name__ == "__main__":
+    unittest.main()