xiaoxia-saas/tests/unit/test_auth.py

import asyncio
import sys
from pathlib import Path

import jwt
import pytest
from fastapi import HTTPException
from fastapi.security import HTTPAuthorizationCredentials

ROOT = Path(__file__).resolve().parents[2]
API_ROOT = ROOT / "apps" / "api"
if str(API_ROOT) not in sys.path:
    sys.path.insert(0, str(API_ROOT))

from app.api.routes.auth import (
    PasswordResetRequestModel,
    RegisterRequest,
    ResetPasswordModel,
    VerifyEmailRequestModel,
    forgot_password,
    register,
    reset_password,
    verify_email_post,
)
from app.auth import get_current_user
from app.config import settings

from packages.adapters.smtp import NoopEmailService
from packages.application.auth.login_use_case import LoginRequest, LoginUseCase
from packages.application.auth.register_user_use_case import RegisterUserRequest, RegisterUserUseCase
from packages.domain.auth import password_hasher
from packages.domain.entities import User


class InMemoryUserRepository:
    def __init__(self):
        self.users = {}

    def save(self, user):
        self.users[user.id] = user

    def find_by_id(self, user_id):
        return self.users.get(user_id)

    def find_by_email(self, email):
        return next((user for user in self.users.values() if user.email == email), None)

    def find_by_username(self, username):
        return next((user for user in self.users.values() if user.username == username), None)

    def find_by_verification_token(self, token):
        return next((user for user in self.users.values() if user.email_verification_token == token), None)

    def find_by_password_reset_token(self, token):
        return next((user for user in self.users.values() if user.password_reset_token == token), None)

    def delete(self, user_id):
        return self.users.pop(user_id, None) is not None


class DummySessionStore:
    def __init__(self):
        self.saved = []

    def save_session(self, **kwargs):
        self.saved.append(kwargs)


class DummyEmailService:
    def send_verification_email(self, **kwargs):
        return False, "disabled"


def test_register_use_case_hashes_password_and_normalizes_email():
    repo = InMemoryUserRepository()
    use_case = RegisterUserUseCase(repo, email_service=DummyEmailService())

    response, error = use_case.execute(
        RegisterUserRequest(
            email="USER@EXAMPLE.COM",
            password="Password1",
            username="user",
            display_name="User",
        )
    )

    assert error is None
    assert response is not None
    user = repo.find_by_email("user@example.com")
    assert user is not None
    assert user.password_hash.startswith("$2")
    assert password_hasher.verify_password("Password1", user.password_hash)


def test_register_route_defaults_display_name_to_username():
    repo = InMemoryUserRepository()

    response = asyncio.run(
        register(
            RegisterRequest(email="route@example.com", password="Password1", username="routeuser"),
            repo,
            DummyEmailService(),
        )
    )

    assert response.display_name == "routeuser"
    assert repo.find_by_email("route@example.com").display_name == "routeuser"


def test_login_use_case_returns_verifiable_jwt():
    repo = InMemoryUserRepository()
    session_store = DummySessionStore()
    user = User(
        id="user-1",
        email="user@example.com",
        username="user",
        display_name="User",
        password_hash=password_hasher.hash_password("Password1"),
    )
    repo.save(user)

    response, error = LoginUseCase(repo, session_store=session_store, jwt_secret_key=settings.JWT_SECRET_KEY).execute(
        LoginRequest("user@example.com", "Password1")
    )

    assert error is None
    assert response is not None
    payload = jwt.decode(response.access_token, settings.JWT_SECRET_KEY, algorithms=["HS256"])
    assert response.expires_in == 1800
    assert payload["sub"] == "user-1"
    assert payload["type"] == "user_auth"
    assert session_store.saved


def test_login_use_case_upgrades_legacy_sha256_hash():
    repo = InMemoryUserRepository()
    user = User(
        id="user-1",
        email="user@example.com",
        username="user",
        display_name="User",
        password_hash="19513fdc9da4fb72a4a05eb66917548d3c90ff94d5419e1f2363eea89dfee1dd",
    )
    repo.save(user)

    response, error = LoginUseCase(
        repo, session_store=DummySessionStore(), jwt_secret_key=settings.JWT_SECRET_KEY
    ).execute(LoginRequest("user@example.com", "Password1"))

    assert error is None
    assert response is not None
    assert user.password_hash.startswith("$2")
    assert password_hasher.verify_password("Password1", user.password_hash)


def test_login_use_case_rejects_wrong_password():
    repo = InMemoryUserRepository()
    repo.save(
        User(
            id="user-1",
            email="user@example.com",
            username="user",
            display_name="User",
            password_hash=password_hasher.hash_password("Password1"),
        )
    )

    response, error = LoginUseCase(repo, session_store=DummySessionStore()).execute(
        LoginRequest("user@example.com", "WrongPassword1")
    )

    assert response is None
    assert error == "Invalid email or password"


def test_current_user_dependency_resolves_bearer_user_token():
    repo = InMemoryUserRepository()
    repo.save(
        User(
            id="user-1",
            email="user@example.com",
            username="user",
            display_name="User",
            password_hash=password_hasher.hash_password("Password1"),
        )
    )
    token = jwt.encode({"sub": "user-1", "sid": "session-1", "type": "user_auth"}, settings.JWT_SECRET_KEY, "HS256")

    authenticated_user = asyncio.run(
        get_current_user(HTTPAuthorizationCredentials(scheme="Bearer", credentials=token), repo)
    )

    assert authenticated_user.user.id == "user-1"
    assert authenticated_user.session_id == "session-1"


def test_current_user_dependency_rejects_missing_token():
    with pytest.raises(HTTPException) as error:
        asyncio.run(get_current_user(None, InMemoryUserRepository()))

    assert error.value.status_code == 401


def test_current_user_dependency_rejects_wrong_token_type():
    token = jwt.encode({"sub": "user-1", "type": "refresh"}, settings.JWT_SECRET_KEY, "HS256")

    with pytest.raises(HTTPException) as error:
        asyncio.run(
            get_current_user(HTTPAuthorizationCredentials(scheme="Bearer", credentials=token), InMemoryUserRepository())
        )

    assert error.value.status_code == 401


def test_verify_email_route_uses_use_case():
    repo = InMemoryUserRepository()
    user = User(
        id="user-1",
        email="user@example.com",
        username="user",
        display_name="User",
        password_hash=password_hasher.hash_password("Password1"),
        email_verification_token="verify-token",
    )
    repo.save(user)

    response = asyncio.run(verify_email_post(VerifyEmailRequestModel(token="verify-token"), repo))

    assert response.message == "邮箱验证成功"
    assert user.email_verified is True
    assert user.email_verification_token is None


def test_password_reset_routes_use_use_cases():
    repo = InMemoryUserRepository()
    user = User(
        id="user-1",
        email="user@example.com",
        username="user",
        display_name="User",
        password_hash=password_hasher.hash_password("Password1"),
    )
    repo.save(user)

    forgot_response = asyncio.run(
        forgot_password(PasswordResetRequestModel(email="user@example.com"), repo, NoopEmailService())
    )
    assert forgot_response.message == "如果账户存在，密码重置邮件已发送"
    assert user.password_reset_token

    reset_response = asyncio.run(
        reset_password(ResetPasswordModel(token=user.password_reset_token, new_password="NewPassword1"), repo)
    )

    assert reset_response.message == "密码重置成功"
    assert password_hasher.verify_password("NewPassword1", user.password_hash)
    assert user.password_reset_token is None