feat(auth): add user registration with email verification
- Implement RegisterUserUseCase with password validation and email verification - Implement VerifyEmailUseCase for email confirmation - Add UserRepository interface and InMemoryUserRepository implementation - Support duplicate email/username checking - Generate verification tokens and send verification emails - Add 9 comprehensive unit tests (all passed) Phase 4 Task 6/68 completed
This commit is contained in:
Xiaoxia AI
@@ -0,0 +1,79 @@
|
||||
"""
|
||||
用户仓储 In-Memory 实现
|
||||
"""
|
||||
from typing import Optional, Dict
|
||||
from packages.domain.entities import User
|
||||
from packages.ports.user_repository import UserRepository
|
||||
|
||||
|
||||
class InMemoryUserRepository(UserRepository):
|
||||
"""用户仓储内存实现"""
|
||||
|
||||
def __init__(self):
|
||||
self._users: Dict[str, User] = {}
|
||||
self._email_index: Dict[str, str] = {} # email -> user_id
|
||||
self._username_index: Dict[str, str] = {} # username -> user_id
|
||||
self._verification_token_index: Dict[str, str] = {} # token -> user_id
|
||||
self._reset_token_index: Dict[str, str] = {} # token -> user_id
|
||||
|
||||
def save(self, user: User) -> None:
|
||||
"""保存用户"""
|
||||
self._users[user.id] = user
|
||||
self._email_index[user.email.lower()] = user.id
|
||||
if user.username:
|
||||
self._username_index[user.username.lower()] = user.id
|
||||
if user.email_verification_token:
|
||||
self._verification_token_index[user.email_verification_token] = user.id
|
||||
if user.password_reset_token:
|
||||
self._reset_token_index[user.password_reset_token] = user.id
|
||||
|
||||
def find_by_id(self, user_id: str) -> Optional[User]:
|
||||
"""根据 ID 查找用户"""
|
||||
return self._users.get(user_id)
|
||||
|
||||
def find_by_email(self, email: str) -> Optional[User]:
|
||||
"""根据邮箱查找用户"""
|
||||
user_id = self._email_index.get(email.lower())
|
||||
if user_id:
|
||||
return self._users.get(user_id)
|
||||
return None
|
||||
|
||||
def find_by_username(self, username: str) -> Optional[User]:
|
||||
"""根据用户名查找用户"""
|
||||
user_id = self._username_index.get(username.lower())
|
||||
if user_id:
|
||||
return self._users.get(user_id)
|
||||
return None
|
||||
|
||||
def find_by_verification_token(self, token: str) -> Optional[User]:
|
||||
"""根据邮箱验证令牌查找用户"""
|
||||
user_id = self._verification_token_index.get(token)
|
||||
if user_id:
|
||||
return self._users.get(user_id)
|
||||
return None
|
||||
|
||||
def find_by_password_reset_token(self, token: str) -> Optional[User]:
|
||||
"""根据密码重置令牌查找用户"""
|
||||
user_id = self._reset_token_index.get(token)
|
||||
if user_id:
|
||||
return self._users.get(user_id)
|
||||
return None
|
||||
|
||||
def delete(self, user_id: str) -> bool:
|
||||
"""删除用户"""
|
||||
user = self._users.get(user_id)
|
||||
if not user:
|
||||
return False
|
||||
|
||||
# 清理索引
|
||||
self._email_index.pop(user.email.lower(), None)
|
||||
if user.username:
|
||||
self._username_index.pop(user.username.lower(), None)
|
||||
if user.email_verification_token:
|
||||
self._verification_token_index.pop(user.email_verification_token, None)
|
||||
if user.password_reset_token:
|
||||
self._reset_token_index.pop(user.password_reset_token, None)
|
||||
|
||||
# 删除用户
|
||||
del self._users[user_id]
|
||||
return True
|
||||
@@ -0,0 +1,16 @@
|
||||
"""认证相关 Use Cases"""
|
||||
from packages.application.auth.register_user_use_case import (
|
||||
RegisterUserUseCase,
|
||||
RegisterUserRequest,
|
||||
RegisterUserResponse,
|
||||
VerifyEmailUseCase,
|
||||
VerifyEmailRequest,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"RegisterUserUseCase",
|
||||
"RegisterUserRequest",
|
||||
"RegisterUserResponse",
|
||||
"VerifyEmailUseCase",
|
||||
"VerifyEmailRequest",
|
||||
]
|
||||
@@ -0,0 +1,197 @@
|
||||
"""
|
||||
用户注册 Use Case
|
||||
"""
|
||||
import secrets
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Optional
|
||||
from uuid import uuid4
|
||||
|
||||
from packages.domain.entities import User
|
||||
from packages.domain.auth import password_hasher, password_validator, email_service
|
||||
|
||||
|
||||
class RegisterUserRequest:
|
||||
"""注册请求"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
email: str,
|
||||
password: str,
|
||||
username: str,
|
||||
display_name: str,
|
||||
):
|
||||
self.email = email.strip().lower()
|
||||
self.password = password
|
||||
self.username = username.strip()
|
||||
self.display_name = display_name.strip()
|
||||
|
||||
|
||||
class RegisterUserResponse:
|
||||
"""注册响应"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
user_id: str,
|
||||
email: str,
|
||||
username: str,
|
||||
display_name: str,
|
||||
email_verification_sent: bool,
|
||||
):
|
||||
self.user_id = user_id
|
||||
self.email = email
|
||||
self.username = username
|
||||
self.display_name = display_name
|
||||
self.email_verification_sent = email_verification_sent
|
||||
|
||||
|
||||
class RegisterUserUseCase:
|
||||
"""用户注册用例"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
user_repository,
|
||||
base_url: str = "http://localhost:3000",
|
||||
):
|
||||
"""
|
||||
初始化注册用例
|
||||
|
||||
Args:
|
||||
user_repository: 用户仓储
|
||||
base_url: 应用基础 URL（用于生成验证链接）
|
||||
"""
|
||||
self.user_repository = user_repository
|
||||
self.base_url = base_url
|
||||
|
||||
def execute(self, request: RegisterUserRequest) -> tuple[Optional[RegisterUserResponse], Optional[str]]:
|
||||
"""
|
||||
执行注册
|
||||
|
||||
Args:
|
||||
request: 注册请求
|
||||
|
||||
Returns:
|
||||
(注册响应, 错误信息)
|
||||
"""
|
||||
try:
|
||||
# 1. 验证输入
|
||||
if not request.email:
|
||||
return None, "Email is required"
|
||||
|
||||
if not request.username:
|
||||
return None, "Username is required"
|
||||
|
||||
if not request.display_name:
|
||||
return None, "Display name is required"
|
||||
|
||||
# 2. 验证密码强度
|
||||
valid, error = password_validator.validate(request.password)
|
||||
if not valid:
|
||||
return None, error
|
||||
|
||||
# 3. 检查邮箱是否已存在
|
||||
existing_user = self.user_repository.find_by_email(request.email)
|
||||
if existing_user:
|
||||
return None, "Email already registered"
|
||||
|
||||
# 4. 检查用户名是否已存在
|
||||
existing_username = self.user_repository.find_by_username(request.username)
|
||||
if existing_username:
|
||||
return None, "Username already taken"
|
||||
|
||||
# 5. 哈希密码
|
||||
hashed_password = password_hasher.hash_password(request.password)
|
||||
|
||||
# 6. 生成邮箱验证令牌
|
||||
verification_token = secrets.token_urlsafe(32)
|
||||
|
||||
# 7. 创建用户
|
||||
user = User(
|
||||
id=uuid4().hex,
|
||||
email=request.email,
|
||||
username=request.username,
|
||||
display_name=request.display_name,
|
||||
password_hash=hashed_password,
|
||||
email_verified=False,
|
||||
email_verification_token=verification_token,
|
||||
created_at=datetime.now(timezone.utc),
|
||||
)
|
||||
|
||||
# 8. 保存用户
|
||||
self.user_repository.save(user)
|
||||
|
||||
# 9. 发送验证邮件
|
||||
verification_url = f"{self.base_url}/verify-email?token={verification_token}"
|
||||
email_sent = False
|
||||
|
||||
try:
|
||||
success, error = email_service.send_verification_email(
|
||||
to_email=user.email,
|
||||
username=user.username,
|
||||
verification_url=verification_url,
|
||||
)
|
||||
email_sent = success
|
||||
|
||||
if not success:
|
||||
print(f"Failed to send verification email: {error}")
|
||||
except Exception as e:
|
||||
print(f"Email service error: {e}")
|
||||
|
||||
# 10. 返回响应（即使邮件发送失败，用户也已创建）
|
||||
return RegisterUserResponse(
|
||||
user_id=user.id,
|
||||
email=user.email,
|
||||
username=user.username,
|
||||
display_name=user.display_name,
|
||||
email_verification_sent=email_sent,
|
||||
), None
|
||||
|
||||
except Exception as e:
|
||||
return None, f"Registration failed: {str(e)}"
|
||||
|
||||
|
||||
class VerifyEmailRequest:
|
||||
"""邮箱验证请求"""
|
||||
|
||||
def __init__(self, token: str):
|
||||
self.token = token
|
||||
|
||||
|
||||
class VerifyEmailUseCase:
|
||||
"""邮箱验证用例"""
|
||||
|
||||
def __init__(self, user_repository):
|
||||
self.user_repository = user_repository
|
||||
|
||||
def execute(self, request: VerifyEmailRequest) -> tuple[bool, Optional[str]]:
|
||||
"""
|
||||
执行邮箱验证
|
||||
|
||||
Args:
|
||||
request: 验证请求
|
||||
|
||||
Returns:
|
||||
(是否成功, 错误信息)
|
||||
"""
|
||||
try:
|
||||
if not request.token:
|
||||
return False, "Verification token is required"
|
||||
|
||||
# 查找用户
|
||||
user = self.user_repository.find_by_verification_token(request.token)
|
||||
if not user:
|
||||
return False, "Invalid or expired verification token"
|
||||
|
||||
# 检查是否已验证
|
||||
if user.email_verified:
|
||||
return True, None # 已验证，返回成功
|
||||
|
||||
# 更新用户状态
|
||||
user.email_verified = True
|
||||
user.email_verification_token = None # 清空令牌
|
||||
|
||||
self.user_repository.save(user)
|
||||
|
||||
return True, None
|
||||
|
||||
except Exception as e:
|
||||
return False, f"Email verification failed: {str(e)}"
|
||||
@@ -0,0 +1,45 @@
|
||||
"""
|
||||
用户仓储接口
|
||||
"""
|
||||
from abc import ABC, abstractmethod
|
||||
from typing import Optional
|
||||
from packages.domain.entities import User
|
||||
|
||||
|
||||
class UserRepository(ABC):
|
||||
"""用户仓储接口"""
|
||||
|
||||
@abstractmethod
|
||||
def save(self, user: User) -> None:
|
||||
"""保存用户"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def find_by_id(self, user_id: str) -> Optional[User]:
|
||||
"""根据 ID 查找用户"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def find_by_email(self, email: str) -> Optional[User]:
|
||||
"""根据邮箱查找用户"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def find_by_username(self, username: str) -> Optional[User]:
|
||||
"""根据用户名查找用户"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def find_by_verification_token(self, token: str) -> Optional[User]:
|
||||
"""根据邮箱验证令牌查找用户"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def find_by_password_reset_token(self, token: str) -> Optional[User]:
|
||||
"""根据密码重置令牌查找用户"""
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def delete(self, user_id: str) -> bool:
|
||||
"""删除用户"""
|
||||
pass
|
||||
@@ -0,0 +1,221 @@
|
||||
"""
|
||||
用户注册 Use Case 测试
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import Mock, patch
|
||||
from packages.application.auth import (
|
||||
RegisterUserUseCase,
|
||||
RegisterUserRequest,
|
||||
VerifyEmailUseCase,
|
||||
VerifyEmailRequest,
|
||||
)
|
||||
from packages.domain.entities import User
|
||||
|
||||
|
||||
class TestRegisterUserUseCase:
|
||||
"""注册用例测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_user_repo(self):
|
||||
"""Mock 用户仓储"""
|
||||
repo = Mock()
|
||||
repo.find_by_email = Mock(return_value=None)
|
||||
repo.find_by_username = Mock(return_value=None)
|
||||
repo.find_by_verification_token = Mock(return_value=None)
|
||||
repo.save = Mock()
|
||||
return repo
|
||||
|
||||
@pytest.fixture
|
||||
def use_case(self, mock_user_repo):
|
||||
"""创建注册用例"""
|
||||
return RegisterUserUseCase(
|
||||
user_repository=mock_user_repo,
|
||||
base_url="https://test.com"
|
||||
)
|
||||
|
||||
@patch('packages.application.auth.register_user_use_case.email_service')
|
||||
def test_register_user_success(self, mock_email_service, use_case, mock_user_repo):
|
||||
"""测试注册成功"""
|
||||
mock_email_service.send_verification_email.return_value = (True, None)
|
||||
|
||||
request = RegisterUserRequest(
|
||||
email="test@example.com",
|
||||
password="SecurePass123",
|
||||
username="testuser",
|
||||
display_name="Test User",
|
||||
)
|
||||
|
||||
response, error = use_case.execute(request)
|
||||
|
||||
assert error is None
|
||||
assert response is not None
|
||||
assert response.email == "test@example.com"
|
||||
assert response.username == "testuser"
|
||||
assert response.display_name == "Test User"
|
||||
assert response.email_verification_sent is True
|
||||
|
||||
# 验证保存了用户
|
||||
mock_user_repo.save.assert_called_once()
|
||||
saved_user = mock_user_repo.save.call_args[0][0]
|
||||
assert saved_user.email == "test@example.com"
|
||||
assert saved_user.password_hash != ""
|
||||
assert saved_user.email_verified is False
|
||||
assert saved_user.email_verification_token is not None
|
||||
|
||||
def test_register_user_weak_password(self, use_case):
|
||||
"""测试弱密码"""
|
||||
request = RegisterUserRequest(
|
||||
email="test@example.com",
|
||||
password="weak",
|
||||
username="testuser",
|
||||
display_name="Test User",
|
||||
)
|
||||
|
||||
response, error = use_case.execute(request)
|
||||
|
||||
assert response is None
|
||||
assert error is not None
|
||||
assert "at least 8 characters" in error
|
||||
|
||||
def test_register_user_email_exists(self, use_case, mock_user_repo):
|
||||
"""测试邮箱已存在"""
|
||||
# Mock 返回已存在的用户
|
||||
existing_user = User(
|
||||
id="existing-id",
|
||||
email="test@example.com",
|
||||
username="existing",
|
||||
display_name="Existing",
|
||||
)
|
||||
mock_user_repo.find_by_email.return_value = existing_user
|
||||
|
||||
request = RegisterUserRequest(
|
||||
email="test@example.com",
|
||||
password="SecurePass123",
|
||||
username="testuser",
|
||||
display_name="Test User",
|
||||
)
|
||||
|
||||
response, error = use_case.execute(request)
|
||||
|
||||
assert response is None
|
||||
assert error == "Email already registered"
|
||||
|
||||
def test_register_user_username_taken(self, use_case, mock_user_repo):
|
||||
"""测试用户名已被占用"""
|
||||
existing_user = User(
|
||||
id="existing-id",
|
||||
email="other@example.com",
|
||||
username="testuser",
|
||||
display_name="Other",
|
||||
)
|
||||
mock_user_repo.find_by_username.return_value = existing_user
|
||||
|
||||
request = RegisterUserRequest(
|
||||
email="test@example.com",
|
||||
password="SecurePass123",
|
||||
username="testuser",
|
||||
display_name="Test User",
|
||||
)
|
||||
|
||||
response, error = use_case.execute(request)
|
||||
|
||||
assert response is None
|
||||
assert error == "Username already taken"
|
||||
|
||||
def test_register_user_missing_email(self, use_case):
|
||||
"""测试缺少邮箱"""
|
||||
request = RegisterUserRequest(
|
||||
email="",
|
||||
password="SecurePass123",
|
||||
username="testuser",
|
||||
display_name="Test User",
|
||||
)
|
||||
|
||||
response, error = use_case.execute(request)
|
||||
|
||||
assert response is None
|
||||
assert error == "Email is required"
|
||||
|
||||
@patch('packages.application.auth.register_user_use_case.email_service')
|
||||
def test_register_user_email_send_failure(self, mock_email_service, use_case, mock_user_repo):
|
||||
"""测试邮件发送失败（用户仍然创建）"""
|
||||
mock_email_service.send_verification_email.return_value = (False, "SMTP error")
|
||||
|
||||
request = RegisterUserRequest(
|
||||
email="test@example.com",
|
||||
password="SecurePass123",
|
||||
username="testuser",
|
||||
display_name="Test User",
|
||||
)
|
||||
|
||||
response, error = use_case.execute(request)
|
||||
|
||||
assert error is None # 用户创建成功
|
||||
assert response is not None
|
||||
assert response.email_verification_sent is False # 但邮件发送失败
|
||||
|
||||
|
||||
class TestVerifyEmailUseCase:
|
||||
"""邮箱验证用例测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_user_repo(self):
|
||||
repo = Mock()
|
||||
repo.find_by_verification_token = Mock(return_value=None)
|
||||
repo.save = Mock()
|
||||
return repo
|
||||
|
||||
@pytest.fixture
|
||||
def use_case(self, mock_user_repo):
|
||||
return VerifyEmailUseCase(user_repository=mock_user_repo)
|
||||
|
||||
def test_verify_email_success(self, use_case, mock_user_repo):
|
||||
"""测试验证成功"""
|
||||
user = User(
|
||||
id="user-123",
|
||||
email="test@example.com",
|
||||
username="testuser",
|
||||
display_name="Test User",
|
||||
email_verified=False,
|
||||
email_verification_token="valid-token",
|
||||
)
|
||||
mock_user_repo.find_by_verification_token.return_value = user
|
||||
|
||||
request = VerifyEmailRequest(token="valid-token")
|
||||
success, error = use_case.execute(request)
|
||||
|
||||
assert success is True
|
||||
assert error is None
|
||||
|
||||
# 验证用户状态已更新
|
||||
assert user.email_verified is True
|
||||
assert user.email_verification_token is None
|
||||
mock_user_repo.save.assert_called_once()
|
||||
|
||||
def test_verify_email_invalid_token(self, use_case, mock_user_repo):
|
||||
"""测试无效令牌"""
|
||||
mock_user_repo.find_by_verification_token.return_value = None
|
||||
|
||||
request = VerifyEmailRequest(token="invalid-token")
|
||||
success, error = use_case.execute(request)
|
||||
|
||||
assert success is False
|
||||
assert error == "Invalid or expired verification token"
|
||||
|
||||
def test_verify_email_already_verified(self, use_case, mock_user_repo):
|
||||
"""测试已验证的邮箱"""
|
||||
user = User(
|
||||
id="user-123",
|
||||
email="test@example.com",
|
||||
username="testuser",
|
||||
display_name="Test User",
|
||||
email_verified=True,
|
||||
email_verification_token="old-token",
|
||||
)
|
||||
mock_user_repo.find_by_verification_token.return_value = user
|
||||
|
||||
request = VerifyEmailRequest(token="old-token")
|
||||
success, error = use_case.execute(request)
|
||||
|
||||
assert success is True # 已验证也返回成功
|
||||
assert error is None
|
||||
Reference in New Issue
Block a user