feat(auth): add password reset functionality
- Implement RequestPasswordResetUseCase to generate reset token - Send password reset email with 1-hour expiration - Implement ResetPasswordUseCase to verify token and update password - Security: return success even if user not exists (avoid enumeration) - Validate new password strength before reset - Clear reset token after successful password change - Add 9 comprehensive unit tests (all passed) Phase 4 Task 8/68 completed
This commit is contained in:
@@ -0,0 +1,197 @@
|
||||
"""
|
||||
密码重置 Use Case 测试
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import Mock, patch
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from packages.application.auth.password_reset_use_case import (
|
||||
RequestPasswordResetUseCase,
|
||||
RequestPasswordResetRequest,
|
||||
ResetPasswordUseCase,
|
||||
ResetPasswordRequest,
|
||||
)
|
||||
from packages.domain.entities import User
|
||||
|
||||
|
||||
class TestRequestPasswordResetUseCase:
|
||||
"""请求密码重置测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_user_repo(self):
|
||||
repo = Mock()
|
||||
repo.find_by_email = Mock(return_value=None)
|
||||
repo.save = Mock()
|
||||
return repo
|
||||
|
||||
@pytest.fixture
|
||||
def use_case(self, mock_user_repo):
|
||||
return RequestPasswordResetUseCase(
|
||||
user_repository=mock_user_repo,
|
||||
base_url="https://test.com",
|
||||
token_expire_hours=1,
|
||||
)
|
||||
|
||||
@pytest.fixture
|
||||
def test_user(self):
|
||||
return User(
|
||||
id="user-123",
|
||||
email="test@example.com",
|
||||
username="testuser",
|
||||
display_name="Test User",
|
||||
password_hash="hash",
|
||||
)
|
||||
|
||||
@patch('packages.application.auth.password_reset_use_case.email_service')
|
||||
def test_request_reset_success(self, mock_email_service, use_case, mock_user_repo, test_user):
|
||||
"""测试请求重置成功"""
|
||||
mock_user_repo.find_by_email.return_value = test_user
|
||||
mock_email_service.send_password_reset_email.return_value = (True, None)
|
||||
|
||||
request = RequestPasswordResetRequest(email="test@example.com")
|
||||
success, error = use_case.execute(request)
|
||||
|
||||
assert success is True
|
||||
assert error is None
|
||||
|
||||
# 验证保存了用户
|
||||
mock_user_repo.save.assert_called_once()
|
||||
saved_user = mock_user_repo.save.call_args[0][0]
|
||||
assert saved_user.password_reset_token is not None
|
||||
assert saved_user.password_reset_expires_at is not None
|
||||
|
||||
# 验证发送了邮件
|
||||
mock_email_service.send_password_reset_email.assert_called_once()
|
||||
|
||||
@patch('packages.application.auth.password_reset_use_case.email_service')
|
||||
def test_request_reset_user_not_exists(self, mock_email_service, use_case, mock_user_repo):
|
||||
"""测试用户不存在(仍返回成功,避免暴露)"""
|
||||
mock_user_repo.find_by_email.return_value = None
|
||||
|
||||
request = RequestPasswordResetRequest(email="nonexistent@example.com")
|
||||
success, error = use_case.execute(request)
|
||||
|
||||
assert success is True # 安全考虑,仍返回成功
|
||||
assert error is None
|
||||
|
||||
# 不发送邮件
|
||||
mock_email_service.send_password_reset_email.assert_not_called()
|
||||
|
||||
def test_request_reset_missing_email(self, use_case):
|
||||
"""测试缺少邮箱"""
|
||||
request = RequestPasswordResetRequest(email="")
|
||||
success, error = use_case.execute(request)
|
||||
|
||||
assert success is False
|
||||
assert error == "Email is required"
|
||||
|
||||
|
||||
class TestResetPasswordUseCase:
|
||||
"""重置密码测试"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_user_repo(self):
|
||||
repo = Mock()
|
||||
repo.find_by_password_reset_token = Mock(return_value=None)
|
||||
repo.save = Mock()
|
||||
return repo
|
||||
|
||||
@pytest.fixture
|
||||
def use_case(self, mock_user_repo):
|
||||
return ResetPasswordUseCase(user_repository=mock_user_repo)
|
||||
|
||||
@pytest.fixture
|
||||
def test_user(self):
|
||||
return User(
|
||||
id="user-123",
|
||||
email="test@example.com",
|
||||
username="testuser",
|
||||
display_name="Test User",
|
||||
password_hash="old-hash",
|
||||
password_reset_token="valid-token",
|
||||
password_reset_expires_at=datetime.now(timezone.utc) + timedelta(hours=1),
|
||||
)
|
||||
|
||||
def test_reset_password_success(self, use_case, mock_user_repo, test_user):
|
||||
"""测试重置密码成功"""
|
||||
mock_user_repo.find_by_password_reset_token.return_value = test_user
|
||||
|
||||
request = ResetPasswordRequest(
|
||||
token="valid-token",
|
||||
new_password="NewSecurePass123",
|
||||
)
|
||||
success, error = use_case.execute(request)
|
||||
|
||||
assert success is True
|
||||
assert error is None
|
||||
|
||||
# 验证密码已更新
|
||||
assert test_user.password_hash != "old-hash"
|
||||
assert test_user.password_reset_token is None
|
||||
assert test_user.password_reset_expires_at is None
|
||||
|
||||
# 验证保存了用户
|
||||
mock_user_repo.save.assert_called_once()
|
||||
|
||||
def test_reset_password_weak_password(self, use_case, mock_user_repo, test_user):
|
||||
"""测试弱密码"""
|
||||
mock_user_repo.find_by_password_reset_token.return_value = test_user
|
||||
|
||||
request = ResetPasswordRequest(
|
||||
token="valid-token",
|
||||
new_password="weak",
|
||||
)
|
||||
success, error = use_case.execute(request)
|
||||
|
||||
assert success is False
|
||||
assert "at least 8 characters" in error
|
||||
|
||||
def test_reset_password_invalid_token(self, use_case, mock_user_repo):
|
||||
"""测试无效令牌"""
|
||||
mock_user_repo.find_by_password_reset_token.return_value = None
|
||||
|
||||
request = ResetPasswordRequest(
|
||||
token="invalid-token",
|
||||
new_password="NewSecurePass123",
|
||||
)
|
||||
success, error = use_case.execute(request)
|
||||
|
||||
assert success is False
|
||||
assert error == "Invalid or expired reset token"
|
||||
|
||||
def test_reset_password_expired_token(self, use_case, mock_user_repo, test_user):
|
||||
"""测试过期令牌"""
|
||||
test_user.password_reset_expires_at = datetime.now(timezone.utc) - timedelta(hours=1)
|
||||
mock_user_repo.find_by_password_reset_token.return_value = test_user
|
||||
|
||||
request = ResetPasswordRequest(
|
||||
token="valid-token",
|
||||
new_password="NewSecurePass123",
|
||||
)
|
||||
success, error = use_case.execute(request)
|
||||
|
||||
assert success is False
|
||||
assert error == "Reset token has expired"
|
||||
|
||||
def test_reset_password_missing_token(self, use_case):
|
||||
"""测试缺少令牌"""
|
||||
request = ResetPasswordRequest(
|
||||
token="",
|
||||
new_password="NewSecurePass123",
|
||||
)
|
||||
success, error = use_case.execute(request)
|
||||
|
||||
assert success is False
|
||||
assert error == "Reset token is required"
|
||||
|
||||
def test_reset_password_missing_password(self, use_case, mock_user_repo, test_user):
|
||||
"""测试缺少新密码"""
|
||||
mock_user_repo.find_by_password_reset_token.return_value = test_user
|
||||
|
||||
request = ResetPasswordRequest(
|
||||
token="valid-token",
|
||||
new_password="",
|
||||
)
|
||||
success, error = use_case.execute(request)
|
||||
|
||||
assert success is False
|
||||
assert error == "New password is required"
|
||||
Reference in New Issue
Block a user