xiaoxia-saas/tests/unit/test_password_reset_use_case.py

"""密码重置 Use Case 单元测试."""

from datetime import datetime, timedelta, timezone
from unittest.mock import MagicMock, patch

import pytest

from packages.application.auth.password_reset_use_case import (
    RequestPasswordResetRequest,
    RequestPasswordResetUseCase,
    ResetPasswordRequest,
    ResetPasswordUseCase,
)
from packages.domain.entities import User

# ── Test Fixtures ────────────────────────────────────────────────────────────


def _make_user(
    user_id="user-1",
    email="user@example.com",
    username="testuser",
    display_name="Test User",
    password_hash="hashed_password_123",
):
    """创建一个测试用户."""
    return User(
        id=user_id,
        email=email,
        display_name=display_name,
        username=username,
        password_hash=password_hash,
    )


@pytest.fixture
def mock_user_repo():
    """mock 用户仓储."""
    repo = MagicMock()
    repo.find_by_email.return_value = None
    repo.find_by_password_reset_token.return_value = None
    repo.save.return_value = None
    return repo


@pytest.fixture
def mock_email_service():
    """mock 邮件服务."""
    svc = MagicMock()
    svc.send_password_reset_email.return_value = (True, None)
    return svc


# ── RequestPasswordResetUseCase 测试 ────────────────────────────────────────


class TestRequestPasswordReset:
    """请求密码重置用例测试"""

    def test_request_success_sends_email(self, mock_user_repo, mock_email_service):
        """成功请求时发送重置邮件."""
        user = _make_user()
        mock_user_repo.find_by_email.return_value = user

        uc = RequestPasswordResetUseCase(
            user_repository=mock_user_repo,
            base_url="https://app.example.com",
            email_service=mock_email_service,
        )
        req = RequestPasswordResetRequest(email="user@example.com")
        ok, error = uc.execute(req)

        assert ok is True
        assert error is None
        # 用户被更新了 reset_token
        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 len(saved_user.password_reset_token) > 0
        assert saved_user.password_reset_expires_at is not None
        # 邮件发送了
        mock_email_service.send_password_reset_email.assert_called_once()
        call_kwargs = mock_email_service.send_password_reset_email.call_args.kwargs
        assert call_kwargs["to_email"] == "user@example.com"
        assert "reset-password?token=" in call_kwargs["reset_url"]
        assert "https://app.example.com" in call_kwargs["reset_url"]

    def test_request_nonexistent_user_returns_success(self, mock_user_repo, mock_email_service):
        """用户不存在时也返回成功（不暴露用户存在性）."""
        mock_user_repo.find_by_email.return_value = None

        uc = RequestPasswordResetUseCase(
            user_repository=mock_user_repo,
            base_url="https://app.example.com",
            email_service=mock_email_service,
        )
        req = RequestPasswordResetRequest(email="nonexistent@example.com")
        ok, error = uc.execute(req)

        assert ok is True
        assert error is None
        # 不保存任何东西
        mock_user_repo.save.assert_not_called()
        # 不发邮件
        mock_email_service.send_password_reset_email.assert_not_called()

    def test_request_empty_email(self, mock_user_repo, mock_email_service):
        """空邮箱返回错误."""
        uc = RequestPasswordResetUseCase(
            user_repository=mock_user_repo,
            base_url="https://app.example.com",
            email_service=mock_email_service,
        )
        req = RequestPasswordResetRequest(email="")
        ok, error = uc.execute(req)

        assert ok is False
        assert "Email is required" in error

    def test_request_email_normalized(self, mock_user_repo, mock_email_service):
        """邮箱会被规范化（小写+去空格）."""
        user = _make_user()
        mock_user_repo.find_by_email.return_value = user

        uc = RequestPasswordResetUseCase(
            user_repository=mock_user_repo,
            base_url="https://app.example.com",
            email_service=mock_email_service,
        )
        req = RequestPasswordResetRequest(email="  USER@Example.COM  ")
        ok, _ = uc.execute(req)

        assert ok is True
        # find_by_email 收到的是小写的
        mock_user_repo.find_by_email.assert_called_with("user@example.com")

    def test_request_token_expiry_custom_hours(self, mock_user_repo, mock_email_service):
        """自定义令牌过期时间."""
        user = _make_user()
        mock_user_repo.find_by_email.return_value = user

        uc = RequestPasswordResetUseCase(
            user_repository=mock_user_repo,
            base_url="https://app.example.com",
            token_expire_hours=6,
            email_service=mock_email_service,
        )
        req = RequestPasswordResetRequest(email="user@example.com")
        before = datetime.now(timezone.utc)
        ok, _ = uc.execute(req)
        after = datetime.now(timezone.utc)

        assert ok is True
        saved_user = mock_user_repo.save.call_args[0][0]
        expires_at = saved_user.password_reset_expires_at
        # 过期时间应该在 ~6 小时后
        expected_min = before + timedelta(hours=6)
        expected_max = after + timedelta(hours=6)
        assert expected_min <= expires_at <= expected_max

    def test_request_email_failure_returns_success(self, mock_user_repo, mock_email_service):
        """邮件发送失败不影响返回结果（安全考虑）."""
        user = _make_user()
        mock_user_repo.find_by_email.return_value = user
        mock_email_service.send_password_reset_email.return_value = (False, "SMTP error")

        uc = RequestPasswordResetUseCase(
            user_repository=mock_user_repo,
            base_url="https://app.example.com",
            email_service=mock_email_service,
        )
        req = RequestPasswordResetRequest(email="user@example.com")
        ok, error = uc.execute(req)

        assert ok is True
        assert error is None
        # token 仍然保存了
        mock_user_repo.save.assert_called_once()

    def test_request_email_exception_does_not_propagate(self, mock_user_repo, mock_email_service):
        """邮件服务异常不向外传播."""
        user = _make_user()
        mock_user_repo.find_by_email.return_value = user
        mock_email_service.send_password_reset_email.side_effect = Exception("SMTP down")

        uc = RequestPasswordResetUseCase(
            user_repository=mock_user_repo,
            base_url="https://app.example.com",
            email_service=mock_email_service,
        )
        req = RequestPasswordResetRequest(email="user@example.com")
        ok, error = uc.execute(req)

        assert ok is True
        assert error is None

    def test_request_username_uses_display_name_fallback(self, mock_user_repo, mock_email_service):
        """用户名为空时用 display_name."""
        user = _make_user(username="", display_name="Display Name")
        mock_user_repo.find_by_email.return_value = user

        uc = RequestPasswordResetUseCase(
            user_repository=mock_user_repo,
            base_url="https://app.example.com",
            email_service=mock_email_service,
        )
        req = RequestPasswordResetRequest(email="user@example.com")
        uc.execute(req)

        call_kwargs = mock_email_service.send_password_reset_email.call_args.kwargs
        assert call_kwargs["username"] == "Display Name"

    def test_request_uses_username_when_available(self, mock_user_repo, mock_email_service):
        """有用户名时用用户名."""
        user = _make_user(username="myusername", display_name="Display Name")
        mock_user_repo.find_by_email.return_value = user

        uc = RequestPasswordResetUseCase(
            user_repository=mock_user_repo,
            base_url="https://app.example.com",
            email_service=mock_email_service,
        )
        req = RequestPasswordResetRequest(email="user@example.com")
        uc.execute(req)

        call_kwargs = mock_email_service.send_password_reset_email.call_args.kwargs
        assert call_kwargs["username"] == "myusername"

    def test_request_generates_unique_tokens(self, mock_user_repo, mock_email_service):
        """每次请求生成不同的令牌."""
        user = _make_user()
        mock_user_repo.find_by_email.return_value = user

        uc = RequestPasswordResetUseCase(
            user_repository=mock_user_repo,
            base_url="https://app.example.com",
            email_service=mock_email_service,
        )

        tokens = []
        for _ in range(3):
            req = RequestPasswordResetRequest(email="user@example.com")
            uc.execute(req)
            saved_user = mock_user_repo.save.call_args[0][0]
            tokens.append(saved_user.password_reset_token)

        assert len(set(tokens)) == 3  # 三个不同的令牌

    def test_request_general_exception_returns_error(self, mock_user_repo, mock_email_service):
        """其他异常返回错误信息."""
        mock_user_repo.find_by_email.side_effect = Exception("DB connection error")

        uc = RequestPasswordResetUseCase(
            user_repository=mock_user_repo,
            base_url="https://app.example.com",
            email_service=mock_email_service,
        )
        req = RequestPasswordResetRequest(email="user@example.com")
        ok, error = uc.execute(req)

        assert ok is False
        assert "failed" in error.lower()


# ── ResetPasswordUseCase 测试 ───────────────────────────────────────────────


class TestResetPassword:
    """重置密码用例测试"""

    def test_reset_success(self, mock_user_repo):
        """成功重置密码."""
        user = _make_user()
        user.password_reset_token = "valid-token-123"
        user.password_reset_expires_at = datetime.now(timezone.utc) + timedelta(hours=1)
        mock_user_repo.find_by_password_reset_token.return_value = user

        uc = ResetPasswordUseCase(user_repository=mock_user_repo)

        # mock password_hasher 和 password_validator
        with (
            patch("packages.application.auth.password_reset_use_case.password_hasher") as mock_hasher,
            patch("packages.application.auth.password_reset_use_case.password_validator") as mock_validator,
        ):
            mock_validator.validate.return_value = (True, None)
            mock_hasher.hash_password.return_value = "new_hashed_password"

            req = ResetPasswordRequest(token="valid-token-123", new_password="NewPass123!")
            ok, error = uc.execute(req)

        assert ok 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_hash == "new_hashed_password"
        # 令牌被清除
        assert saved_user.password_reset_token is None
        assert saved_user.password_reset_expires_at is None

    def test_reset_empty_token(self, mock_user_repo):
        """空令牌返回错误."""
        uc = ResetPasswordUseCase(user_repository=mock_user_repo)
        req = ResetPasswordRequest(token="", new_password="NewPass123!")
        ok, error = uc.execute(req)

        assert ok is False
        assert "token is required" in error.lower()

    def test_reset_empty_password(self, mock_user_repo):
        """空密码返回错误."""
        uc = ResetPasswordUseCase(user_repository=mock_user_repo)
        req = ResetPasswordRequest(token="valid-token", new_password="")
        ok, error = uc.execute(req)

        assert ok is False
        assert "password is required" in error.lower()

    def test_reset_invalid_token(self, mock_user_repo):
        """无效令牌返回错误."""
        mock_user_repo.find_by_password_reset_token.return_value = None

        uc = ResetPasswordUseCase(user_repository=mock_user_repo)

        with patch("packages.application.auth.password_reset_use_case.password_validator") as mock_validator:
            mock_validator.validate.return_value = (True, None)
            req = ResetPasswordRequest(token="invalid-token", new_password="NewPass123!")
            ok, error = uc.execute(req)

        assert ok is False
        assert "Invalid or expired" in error

    def test_reset_expired_token(self, mock_user_repo):
        """过期令牌返回错误."""
        user = _make_user()
        user.password_reset_token = "expired-token"
        user.password_reset_expires_at = datetime.now(timezone.utc) - timedelta(hours=1)
        mock_user_repo.find_by_password_reset_token.return_value = user

        uc = ResetPasswordUseCase(user_repository=mock_user_repo)

        with patch("packages.application.auth.password_reset_use_case.password_validator") as mock_validator:
            mock_validator.validate.return_value = (True, None)
            req = ResetPasswordRequest(token="expired-token", new_password="NewPass123!")
            ok, error = uc.execute(req)

        assert ok is False
        assert "expired" in error.lower()

    def test_reset_naive_datetime_treated_as_utc(self, mock_user_repo):
        """不带时区的过期时间被当作 UTC 处理."""
        user = _make_user()
        user.password_reset_token = "token-123"
        # 用 naive datetime（无时区），应该被当作 UTC
        user.password_reset_expires_at = datetime.utcnow() - timedelta(hours=1)  # type: ignore
        mock_user_repo.find_by_password_reset_token.return_value = user

        uc = ResetPasswordUseCase(user_repository=mock_user_repo)

        with patch("packages.application.auth.password_reset_use_case.password_validator") as mock_validator:
            mock_validator.validate.return_value = (True, None)
            req = ResetPasswordRequest(token="token-123", new_password="NewPass123!")
            ok, error = uc.execute(req)

        assert ok is False
        assert "expired" in error.lower()

    def test_reset_weak_password_fails(self, mock_user_repo):
        """弱密码被拒绝."""
        user = _make_user()
        user.password_reset_token = "valid-token"
        user.password_reset_expires_at = datetime.now(timezone.utc) + timedelta(hours=1)
        mock_user_repo.find_by_password_reset_token.return_value = user

        uc = ResetPasswordUseCase(user_repository=mock_user_repo)

        with patch("packages.application.auth.password_reset_use_case.password_validator") as mock_validator:
            mock_validator.validate.return_value = (False, "Password too short")
            req = ResetPasswordRequest(token="valid-token", new_password="123")
            ok, error = uc.execute(req)

        assert ok is False
        assert "too short" in error.lower()
        # 密码没被更新
        mock_user_repo.save.assert_not_called()

    def test_reset_no_expires_at_still_works(self, mock_user_repo):
        """没有过期时间时视为不过期."""
        user = _make_user()
        user.password_reset_token = "valid-token"
        user.password_reset_expires_at = None
        mock_user_repo.find_by_password_reset_token.return_value = user

        uc = ResetPasswordUseCase(user_repository=mock_user_repo)

        with (
            patch("packages.application.auth.password_reset_use_case.password_hasher") as mock_hasher,
            patch("packages.application.auth.password_reset_use_case.password_validator") as mock_validator,
        ):
            mock_validator.validate.return_value = (True, None)
            mock_hasher.hash_password.return_value = "newhash"
            req = ResetPasswordRequest(token="valid-token", new_password="NewPass123!")
            ok, error = uc.execute(req)

        assert ok is True
        assert error is None

    def test_reset_exception_returns_error(self, mock_user_repo):
        """异常情况返回错误信息."""
        mock_user_repo.find_by_password_reset_token.side_effect = Exception("DB error")

        uc = ResetPasswordUseCase(user_repository=mock_user_repo)
        req = ResetPasswordRequest(token="token", new_password="NewPass123!")
        ok, error = uc.execute(req)

        assert ok is False
        assert "failed" in error.lower()

    def test_reset_clears_token_on_success(self, mock_user_repo):
        """成功重置后令牌被清除，防止重复使用."""
        user = _make_user()
        user.password_reset_token = "valid-token"
        user.password_reset_expires_at = datetime.now(timezone.utc) + timedelta(hours=1)
        mock_user_repo.find_by_password_reset_token.return_value = user

        uc = ResetPasswordUseCase(user_repository=mock_user_repo)

        with (
            patch("packages.application.auth.password_reset_use_case.password_hasher") as mock_hasher,
            patch("packages.application.auth.password_reset_use_case.password_validator") as mock_validator,
        ):
            mock_validator.validate.return_value = (True, None)
            mock_hasher.hash_password.return_value = "newhash"
            req = ResetPasswordRequest(token="valid-token", new_password="NewPass123!")
            ok, _ = uc.execute(req)

        assert ok is True
        saved_user = mock_user_repo.save.call_args[0][0]
        assert saved_user.password_reset_token is None
        assert saved_user.password_reset_expires_at is None

    def test_reset_hashes_new_password(self, mock_user_repo):
        """密码被哈希后保存."""
        user = _make_user()
        user.password_reset_token = "valid-token"
        user.password_reset_expires_at = datetime.now(timezone.utc) + timedelta(hours=1)
        mock_user_repo.find_by_password_reset_token.return_value = user

        uc = ResetPasswordUseCase(user_repository=mock_user_repo)

        with (
            patch("packages.application.auth.password_reset_use_case.password_hasher") as mock_hasher,
            patch("packages.application.auth.password_reset_use_case.password_validator") as mock_validator,
        ):
            mock_validator.validate.return_value = (True, None)
            mock_hasher.hash_password.return_value = "hashed_abcdef"
            req = ResetPasswordRequest(token="valid-token", new_password="MyNewPass123!")
            uc.execute(req)

        mock_hasher.hash_password.assert_called_once_with("MyNewPass123!")
        saved_user = mock_user_repo.save.call_args[0][0]
        assert saved_user.password_hash == "hashed_abcdef"


# ── RequestPasswordResetRequest 测试 ────────────────────────────────────────


class TestRequestPasswordResetRequest:
    """请求数据类测试"""

    def test_email_stripped_and_lowercased(self):
        req = RequestPasswordResetRequest(email="  USER@Example.COM  ")
        assert req.email == "user@example.com"

    def test_email_already_lowercase(self):
        req = RequestPasswordResetRequest(email="user@example.com")
        assert req.email == "user@example.com"


# ── ResetPasswordRequest 测试 ───────────────────────────────────────────────


class TestResetPasswordRequest:
    """重置密码请求数据类测试"""

    def test_stores_token_and_password(self):
        req = ResetPasswordRequest(token="token123", new_password="password123")
        assert req.token == "token123"
        assert req.new_password == "password123"