117 lines
3.7 KiB
Python
117 lines
3.7 KiB
Python
"""Tests for UserService"""
|
|
|
|
import pytest
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from litestar.exceptions import PermissionDeniedException
|
|
from sqlalchemy.exc import IntegrityError
|
|
|
|
from chitai.services import UserService
|
|
from chitai.database import models as m
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestUserServiceAuthentication:
|
|
"""Test authentication functionality."""
|
|
|
|
async def test_authenticate_success(
|
|
self, session: AsyncSession, user_service: UserService
|
|
) -> None:
|
|
"""Test successful user authentication."""
|
|
|
|
# Create a user with a known password
|
|
password = "password123"
|
|
user = m.User(email=f"test@example.com", password=password)
|
|
|
|
session.add(user)
|
|
await session.commit()
|
|
|
|
# Authenticate user
|
|
authenticated_user = await user_service.authenticate(
|
|
"test@example.com", password
|
|
)
|
|
|
|
assert authenticated_user is not None
|
|
assert authenticated_user.email == "test@example.com"
|
|
assert authenticated_user.id == user.id
|
|
|
|
async def test_authenticate_user_not_found(
|
|
self, session: AsyncSession, user_service: UserService
|
|
) -> None:
|
|
"""Test authentication with non-existent user."""
|
|
|
|
with pytest.raises(PermissionDeniedException) as exc_info:
|
|
await user_service.authenticate("nonexistent@example.com", "password")
|
|
|
|
assert "User not found or password invalid" in str(exc_info.value)
|
|
|
|
async def test_authenticate_wrong_password(
|
|
self, session: AsyncSession, user_service: UserService
|
|
) -> None:
|
|
"""Test authentication with wrong password."""
|
|
|
|
# Create user
|
|
password = "password123"
|
|
user = m.User(email=f"test@example.com", password=password)
|
|
|
|
session.add(user)
|
|
await session.commit()
|
|
|
|
# Attempt authentication
|
|
with pytest.raises(PermissionDeniedException) as exc_info:
|
|
await user_service.authenticate("test@example.com", "WrongPassword")
|
|
|
|
assert "User not found or password invalid" in str(exc_info.value)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestUserServiceCRUD:
|
|
"""Test basic CRUD operations."""
|
|
|
|
async def test_create_user_with_password_hashing(
|
|
self, session: AsyncSession, user_service: UserService
|
|
) -> None:
|
|
user_data = {"email": "newuser@example.com", "password": "password123"}
|
|
|
|
user = await user_service.create(data=user_data)
|
|
|
|
assert user.email == "newuser@example.com"
|
|
assert user.password is not None
|
|
assert user.password != "password123" # Password should be hashed
|
|
assert user.password.verify("password123")
|
|
|
|
async def test_get_user_by_email(
|
|
self, session: AsyncSession, user_service: UserService
|
|
) -> None:
|
|
"""Test getting user by email."""
|
|
|
|
user = m.User(email=f"test@example.com", password="password123")
|
|
|
|
session.add(user)
|
|
await session.commit()
|
|
|
|
found_user = await user_service.get_one_or_none(email="test@example.com")
|
|
|
|
assert found_user is not None
|
|
assert found_user.id == user.id
|
|
assert found_user.email == user.email
|
|
|
|
async def test_create_user_with_duplicate_email(
|
|
self, session: AsyncSession, user_service: UserService
|
|
) -> None:
|
|
"""Test creating a new user with a duplicate email."""
|
|
|
|
# Create first user
|
|
user = m.User(email=f"test@example.com", password="password123")
|
|
session.add(user)
|
|
await session.commit()
|
|
|
|
# Create second user
|
|
user = m.User(email=f"test@example.com", password="password12345")
|
|
|
|
with pytest.raises(IntegrityError) as exc_info:
|
|
session.add(user)
|
|
await session.commit()
|
|
|
|
assert "violates unique constraint" in str(exc_info.value)
|