231 lines
8.3 KiB
Python
231 lines
8.3 KiB
Python
"""Tests for authentication service."""
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from fastapi import HTTPException
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
|
|
from app.models.plan import Plan
|
|
from app.models.user import User
|
|
from app.schemas.auth import UserLoginRequest, UserRegisterRequest
|
|
from app.services.auth import AuthService
|
|
from app.utils.auth import PasswordUtils
|
|
|
|
|
|
class TestAuthService:
|
|
"""Test authentication service operations."""
|
|
|
|
@pytest_asyncio.fixture
|
|
def auth_service(self, test_session: AsyncSession) -> AuthService:
|
|
"""Create an auth service instance."""
|
|
return AuthService(test_session)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_success(
|
|
self,
|
|
auth_service: AuthService,
|
|
ensure_plans: tuple[Plan, Plan],
|
|
test_user_data: dict[str, str],
|
|
) -> None:
|
|
"""Test successful user registration."""
|
|
request = UserRegisterRequest(**test_user_data)
|
|
|
|
response = await auth_service.register(request)
|
|
|
|
# Check user data
|
|
assert response.user.email == test_user_data["email"]
|
|
assert response.user.name == test_user_data["name"]
|
|
assert response.user.role == "admin" # First user gets admin role
|
|
assert response.user.is_active is True
|
|
# First user gets pro plan
|
|
free_plan, pro_plan = ensure_plans
|
|
assert response.user.credits == pro_plan.credits
|
|
assert response.user.plan["code"] == pro_plan.code
|
|
|
|
# Check token
|
|
assert response.token.access_token is not None
|
|
assert response.token.token_type == "bearer"
|
|
assert response.token.expires_in > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_duplicate_email(
|
|
self, auth_service: AuthService, test_user: User
|
|
) -> None:
|
|
"""Test registration with duplicate email."""
|
|
request = UserRegisterRequest(
|
|
email=test_user.email, password="password123", name="Another User"
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await auth_service.register(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "Email address is already registered" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_success(
|
|
self,
|
|
auth_service: AuthService,
|
|
test_user: User,
|
|
test_login_data: dict[str, str],
|
|
) -> None:
|
|
"""Test successful user login."""
|
|
request = UserLoginRequest(**test_login_data)
|
|
|
|
response = await auth_service.login(request)
|
|
|
|
# Check user data
|
|
assert response.user.id == test_user.id
|
|
assert response.user.email == test_user.email
|
|
assert response.user.name == test_user.name
|
|
assert response.user.role == test_user.role
|
|
assert response.user.is_active == test_user.is_active
|
|
|
|
# Check token
|
|
assert response.token.access_token is not None
|
|
assert response.token.token_type == "bearer"
|
|
assert response.token.expires_in > 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_invalid_email(self, auth_service: AuthService) -> None:
|
|
"""Test login with invalid email."""
|
|
request = UserLoginRequest(
|
|
email="nonexistent@example.com", password="password123"
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await auth_service.login(request)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert "Invalid email or password" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_invalid_password(
|
|
self, auth_service: AuthService, test_user: User
|
|
) -> None:
|
|
"""Test login with invalid password."""
|
|
request = UserLoginRequest(email=test_user.email, password="wrongpassword")
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await auth_service.login(request)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert "Invalid email or password" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_inactive_user(
|
|
self, auth_service: AuthService, test_user: User, test_session: AsyncSession
|
|
) -> None:
|
|
"""Test login with inactive user."""
|
|
# Store the email before deactivating
|
|
user_email = test_user.email
|
|
|
|
# Deactivate the user
|
|
test_user.is_active = False
|
|
await test_session.commit()
|
|
|
|
request = UserLoginRequest(email=user_email, password="testpassword123")
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await auth_service.login(request)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert "Account is deactivated" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_user_without_password(
|
|
self, auth_service: AuthService, test_user: User, test_session: AsyncSession
|
|
) -> None:
|
|
"""Test login with user that has no password hash."""
|
|
# Store the email before removing password
|
|
user_email = test_user.email
|
|
|
|
# Remove password hash
|
|
test_user.password_hash = None
|
|
await test_session.commit()
|
|
|
|
request = UserLoginRequest(email=user_email, password="anypassword")
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await auth_service.login(request)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert "Invalid email or password" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_current_user_success(
|
|
self, auth_service: AuthService, test_user: User
|
|
) -> None:
|
|
"""Test getting current user successfully."""
|
|
user = await auth_service.get_current_user(test_user.id)
|
|
|
|
assert user.id == test_user.id
|
|
assert user.email == test_user.email
|
|
assert user.name == test_user.name
|
|
assert user.is_active == test_user.is_active
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_current_user_not_found(self, auth_service: AuthService) -> None:
|
|
"""Test getting current user when user doesn't exist."""
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await auth_service.get_current_user(99999)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "User not found" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_current_user_inactive(
|
|
self, auth_service: AuthService, test_user: User, test_session: AsyncSession
|
|
) -> None:
|
|
"""Test getting current user when user is inactive."""
|
|
# Store the user ID before deactivating
|
|
user_id = test_user.id
|
|
|
|
# Deactivate the user
|
|
test_user.is_active = False
|
|
await test_session.commit()
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await auth_service.get_current_user(user_id)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert "Account is deactivated" in exc_info.value.detail
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_access_token(
|
|
self, auth_service: AuthService, test_user: User
|
|
) -> None:
|
|
"""Test access token creation."""
|
|
token_response = auth_service._create_access_token(test_user)
|
|
|
|
assert token_response.access_token is not None
|
|
assert token_response.token_type == "bearer"
|
|
assert token_response.expires_in > 0
|
|
|
|
# Verify token contains correct data
|
|
from app.utils.auth import JWTUtils
|
|
|
|
decoded = JWTUtils.decode_access_token(token_response.access_token)
|
|
assert decoded["sub"] == str(test_user.id)
|
|
assert decoded["email"] == test_user.email
|
|
assert decoded["role"] == test_user.role
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_user_response(
|
|
self, auth_service: AuthService, test_user: User, test_session: AsyncSession
|
|
) -> None:
|
|
"""Test user response creation."""
|
|
# Ensure plan relationship is loaded
|
|
await test_session.refresh(test_user, ["plan"])
|
|
|
|
user_response = await auth_service.create_user_response(test_user)
|
|
|
|
assert user_response.id == test_user.id
|
|
assert user_response.email == test_user.email
|
|
assert user_response.name == test_user.name
|
|
assert user_response.role == test_user.role
|
|
assert user_response.credits == test_user.credits
|
|
assert user_response.is_active == test_user.is_active
|
|
assert user_response.plan["id"] == test_user.plan.id
|
|
assert user_response.plan["code"] == test_user.plan.code
|