diff --git a/packages/adapters/in_memory/user_repository.py b/packages/adapters/in_memory/user_repository.py new file mode 100644 index 000000000..f93eb0299 --- /dev/null +++ b/packages/adapters/in_memory/user_repository.py @@ -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 diff --git a/packages/application/auth/__init__.py b/packages/application/auth/__init__.py new file mode 100644 index 000000000..e73afd1bc --- /dev/null +++ b/packages/application/auth/__init__.py @@ -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", +] diff --git a/packages/application/auth/register_user_use_case.py b/packages/application/auth/register_user_use_case.py new file mode 100644 index 000000000..74a41c640 --- /dev/null +++ b/packages/application/auth/register_user_use_case.py @@ -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)}" diff --git a/packages/ports/user_repository.py b/packages/ports/user_repository.py new file mode 100644 index 000000000..68e9f4e6d --- /dev/null +++ b/packages/ports/user_repository.py @@ -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 diff --git a/tests/unit/test_register_user_use_case.py b/tests/unit/test_register_user_use_case.py new file mode 100644 index 000000000..049f0ff8c --- /dev/null +++ b/tests/unit/test_register_user_use_case.py @@ -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