012-kaopeilian/backend/tests/test_user_service.py

"""
用户服务测试
"""

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import ConflictError, NotFoundError
from app.core.security import verify_password
from app.models.user import User
from app.schemas.user import UserCreate, UserFilter, UserUpdate
from app.services.user_service import UserService


@pytest.mark.asyncio
class TestUserService:
    """用户服务测试类"""

    async def test_create_user(self, db_session: AsyncSession):
        """测试创建用户"""
        # 准备数据
        user_in = UserCreate(
            username="newuser",
            email="newuser@example.com",
            password="password123",
            full_name="New User",
            role="trainee",
        )

        # 创建用户
        service = UserService(db_session)
        user = await service.create_user(obj_in=user_in)

        # 验证结果
        assert user.username == "newuser"
        assert user.email == "newuser@example.com"
        assert user.full_name == "New User"
        assert user.role == "trainee"
        assert verify_password("password123", user.hashed_password)

    async def test_create_user_duplicate_username(
        self,
        db_session: AsyncSession,
        test_user: User,
    ):
        """测试创建重复用户名的用户"""
        user_in = UserCreate(
            username=test_user.username,  # 使用已存在的用户名
            email="another@example.com",
            password="password123",
        )

        service = UserService(db_session)
        with pytest.raises(ConflictError) as exc_info:
            await service.create_user(obj_in=user_in)

        assert f"用户名 {test_user.username} 已存在" in str(exc_info.value)

    async def test_create_user_duplicate_email(
        self,
        db_session: AsyncSession,
        test_user: User,
    ):
        """测试创建重复邮箱的用户"""
        user_in = UserCreate(
            username="anotheruser",
            email=test_user.email,  # 使用已存在的邮箱
            password="password123",
        )

        service = UserService(db_session)
        with pytest.raises(ConflictError) as exc_info:
            await service.create_user(obj_in=user_in)

        assert f"邮箱 {test_user.email} 已存在" in str(exc_info.value)

    async def test_get_by_username(
        self,
        db_session: AsyncSession,
        test_user: User,
    ):
        """测试根据用户名获取用户"""
        service = UserService(db_session)
        user = await service.get_by_username(test_user.username)

        assert user is not None
        assert user.id == test_user.id
        assert user.username == test_user.username

    async def test_get_by_email(
        self,
        db_session: AsyncSession,
        test_user: User,
    ):
        """测试根据邮箱获取用户"""
        service = UserService(db_session)
        user = await service.get_by_email(test_user.email)

        assert user is not None
        assert user.id == test_user.id
        assert user.email == test_user.email

    async def test_update_user(
        self,
        db_session: AsyncSession,
        test_user: User,
    ):
        """测试更新用户"""
        user_update = UserUpdate(
            full_name="Updated Name",
            bio="Updated bio",
        )

        service = UserService(db_session)
        user = await service.update_user(
            user_id=test_user.id,
            obj_in=user_update,
        )

        assert user.full_name == "Updated Name"
        assert user.bio == "Updated bio"

    async def test_update_user_not_found(self, db_session: AsyncSession):
        """测试更新不存在的用户"""
        user_update = UserUpdate(full_name="Updated Name")

        service = UserService(db_session)
        with pytest.raises(NotFoundError) as exc_info:
            await service.update_user(
                user_id=999,
                obj_in=user_update,
            )

        assert "用户不存在" in str(exc_info.value)

    async def test_update_password(
        self,
        db_session: AsyncSession,
        test_user: User,
    ):
        """测试更新密码"""
        service = UserService(db_session)

        # 更新密码
        user = await service.update_password(
            user_id=test_user.id,
            old_password="testpass123",
            new_password="newpass123",
        )

        # 验证新密码
        assert verify_password("newpass123", user.hashed_password)
        assert user.password_changed_at is not None

    async def test_update_password_wrong_old(
        self,
        db_session: AsyncSession,
        test_user: User,
    ):
        """测试使用错误的旧密码更新"""
        service = UserService(db_session)

        with pytest.raises(ConflictError) as exc_info:
            await service.update_password(
                user_id=test_user.id,
                old_password="wrongpass",
                new_password="newpass123",
            )

        assert "旧密码错误" in str(exc_info.value)

    async def test_get_users_with_filter(
        self,
        db_session: AsyncSession,
        test_user: User,
        admin_user: User,
        manager_user: User,
    ):
        """测试根据筛选条件获取用户"""
        service = UserService(db_session)

        # 测试角色筛选
        filter_params = UserFilter(role="admin")
        users, total = await service.get_users_with_filter(
            skip=0,
            limit=10,
            filter_params=filter_params,
        )
        assert total == 1
        assert users[0].id == admin_user.id

        # 测试关键词搜索
        filter_params = UserFilter(keyword="manager")
        users, total = await service.get_users_with_filter(
            skip=0,
            limit=10,
            filter_params=filter_params,
        )
        assert total == 1
        assert users[0].id == manager_user.id

    async def test_authenticate_username(
        self,
        db_session: AsyncSession,
        test_user: User,
    ):
        """测试使用用户名认证"""
        service = UserService(db_session)

        # 正确的密码
        user = await service.authenticate(
            username=test_user.username,
            password="testpass123",
        )
        assert user is not None
        assert user.id == test_user.id

        # 错误的密码
        user = await service.authenticate(
            username=test_user.username,
            password="wrongpass",
        )
        assert user is None

    async def test_authenticate_email(
        self,
        db_session: AsyncSession,
        test_user: User,
    ):
        """测试使用邮箱认证"""
        service = UserService(db_session)

        user = await service.authenticate(
            username=test_user.email,
            password="testpass123",
        )
        assert user is not None
        assert user.id == test_user.id

    async def test_soft_delete(
        self,
        db_session: AsyncSession,
        test_user: User,
    ):
        """测试软删除用户"""
        service = UserService(db_session)

        # 软删除
        user = await service.soft_delete(db_obj=test_user)
        assert user.is_deleted is True
        assert user.deleted_at is not None

        # 验证无法通过常规方法获取
        user = await service.get_by_id(test_user.id)
        assert user is None