Files
disknext/tests/fixtures/users.py
于小丘 f93cb3eedb Add unit tests for models and services
- Implemented unit tests for Object model including folder and file creation, properties, and path retrieval.
- Added unit tests for Setting model covering creation, unique constraints, and type enumeration.
- Created unit tests for User model focusing on user creation, uniqueness, and group relationships.
- Developed unit tests for Login service to validate login functionality, including 2FA and token generation.
- Added utility tests for JWT creation and verification, ensuring token integrity and expiration handling.
- Implemented password utility tests for password generation, hashing, and TOTP verification.
2025-12-19 19:48:05 +08:00

180 lines
4.7 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
用户测试数据工厂
提供创建测试用户的便捷方法。
"""
from uuid import UUID
from sqlmodel.ext.asyncio.session import AsyncSession
from models.user import User
from utils.password.pwd import Password
class UserFactory:
"""用户工厂类,用于创建各种类型的测试用户"""
@staticmethod
async def create(
session: AsyncSession,
group_id: UUID,
username: str | None = None,
password: str | None = None,
**kwargs
) -> User:
"""
创建普通用户
参数:
session: 数据库会话
group_id: 用户组UUID
username: 用户名(默认: test_user_{随机}
password: 明文密码(默认: password123
**kwargs: 其他用户字段
返回:
User: 创建的用户实例
"""
import uuid
if username is None:
username = f"test_user_{uuid.uuid4().hex[:8]}"
if password is None:
password = "password123"
user = User(
username=username,
nickname=kwargs.get("nickname", username),
password=Password.hash(password),
status=kwargs.get("status", True),
storage=kwargs.get("storage", 0),
score=kwargs.get("score", 100),
group_id=group_id,
two_factor=kwargs.get("two_factor"),
avatar=kwargs.get("avatar", "default"),
group_expires=kwargs.get("group_expires"),
theme=kwargs.get("theme", "system"),
language=kwargs.get("language", "zh-CN"),
timezone=kwargs.get("timezone", 8),
previous_group_id=kwargs.get("previous_group_id"),
)
user = await user.save(session)
return user
@staticmethod
async def create_admin(
session: AsyncSession,
admin_group_id: UUID,
username: str | None = None,
password: str | None = None
) -> User:
"""
创建管理员用户
参数:
session: 数据库会话
admin_group_id: 管理员组UUID
username: 用户名(默认: admin_{随机}
password: 明文密码(默认: admin_password
返回:
User: 创建的管理员用户实例
"""
import uuid
if username is None:
username = f"admin_{uuid.uuid4().hex[:8]}"
if password is None:
password = "admin_password"
admin = User(
username=username,
nickname=f"管理员 {username}",
password=Password.hash(password),
status=True,
storage=0,
score=9999,
group_id=admin_group_id,
avatar="default",
)
admin = await admin.save(session)
return admin
@staticmethod
async def create_banned(
session: AsyncSession,
group_id: UUID,
username: str | None = None
) -> User:
"""
创建被封禁用户
参数:
session: 数据库会话
group_id: 用户组UUID
username: 用户名(默认: banned_user_{随机}
返回:
User: 创建的被封禁用户实例
"""
import uuid
if username is None:
username = f"banned_user_{uuid.uuid4().hex[:8]}"
banned_user = User(
username=username,
nickname=f"封禁用户 {username}",
password=Password.hash("banned_password"),
status=False, # 封禁状态
storage=0,
score=0,
group_id=group_id,
avatar="default",
)
banned_user = await banned_user.save(session)
return banned_user
@staticmethod
async def create_with_storage(
session: AsyncSession,
group_id: UUID,
storage_bytes: int,
username: str | None = None
) -> User:
"""
创建已使用指定存储空间的用户
参数:
session: 数据库会话
group_id: 用户组UUID
storage_bytes: 已使用的存储空间(字节)
username: 用户名(默认: storage_user_{随机}
返回:
User: 创建的用户实例
"""
import uuid
if username is None:
username = f"storage_user_{uuid.uuid4().hex[:8]}"
user = User(
username=username,
nickname=username,
password=Password.hash("password123"),
status=True,
storage=storage_bytes,
score=100,
group_id=group_id,
avatar="default",
)
user = await user.save(session)
return user