128 lines
5.3 KiB
Python
128 lines
5.3 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
from sqlalchemy.exc import IntegrityError, OperationalError
|
|
|
|
from app.crud.user import get_user_by_email, create_user
|
|
from app.schemas.user import UserCreate
|
|
from app.models import User as UserModel
|
|
from app.core.exceptions import (
|
|
UserCreationError,
|
|
EmailAlreadyRegisteredError,
|
|
DatabaseConnectionError,
|
|
DatabaseIntegrityError,
|
|
DatabaseQueryError,
|
|
DatabaseTransactionError
|
|
)
|
|
|
|
# Fixtures
|
|
@pytest.fixture
|
|
def mock_db_session():
|
|
session = AsyncMock()
|
|
session.begin = AsyncMock()
|
|
session.begin_nested = AsyncMock()
|
|
session.commit = AsyncMock()
|
|
session.rollback = AsyncMock()
|
|
session.refresh = AsyncMock()
|
|
session.add = MagicMock()
|
|
session.delete = MagicMock()
|
|
session.execute = AsyncMock()
|
|
session.get = AsyncMock()
|
|
session.flush = AsyncMock()
|
|
session.in_transaction = MagicMock(return_value=False)
|
|
return session
|
|
|
|
@pytest.fixture
|
|
def user_create_data():
|
|
return UserCreate(email="test@example.com", password="password123", name="Test User")
|
|
|
|
@pytest.fixture
|
|
def existing_user_data():
|
|
return UserModel(id=1, email="exists@example.com", password_hash="hashed_password", name="Existing User")
|
|
|
|
# Tests for get_user_by_email
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_by_email_found(mock_db_session, existing_user_data):
|
|
mock_result = AsyncMock()
|
|
mock_result.scalars.return_value.first.return_value = existing_user_data
|
|
mock_db_session.execute.return_value = mock_result
|
|
|
|
user = await get_user_by_email(mock_db_session, "exists@example.com")
|
|
assert user is not None
|
|
assert user.email == "exists@example.com"
|
|
mock_db_session.execute.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_by_email_not_found(mock_db_session):
|
|
mock_result = AsyncMock()
|
|
mock_result.scalars.return_value.first.return_value = None
|
|
mock_db_session.execute.return_value = mock_result
|
|
|
|
user = await get_user_by_email(mock_db_session, "nonexistent@example.com")
|
|
assert user is None
|
|
mock_db_session.execute.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_by_email_db_connection_error(mock_db_session):
|
|
mock_db_session.execute.side_effect = OperationalError("mock_op_error", "params", "orig")
|
|
with pytest.raises(DatabaseConnectionError):
|
|
await get_user_by_email(mock_db_session, "test@example.com")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_user_by_email_db_query_error(mock_db_session):
|
|
# Simulate a generic SQLAlchemyError that is not OperationalError
|
|
mock_db_session.execute.side_effect = IntegrityError("mock_sql_error", "params", "orig") # Using IntegrityError as an example of SQLAlchemyError
|
|
with pytest.raises(DatabaseQueryError):
|
|
await get_user_by_email(mock_db_session, "test@example.com")
|
|
|
|
|
|
# Tests for create_user
|
|
@pytest.mark.asyncio
|
|
async def test_create_user_success(mock_db_session, user_create_data):
|
|
mock_result = AsyncMock()
|
|
mock_result.scalar_one_or_none.return_value = UserModel(
|
|
id=1,
|
|
email=user_create_data.email,
|
|
name=user_create_data.name,
|
|
password_hash="hashed_password" # This would be set by the actual hash_password function
|
|
)
|
|
mock_db_session.execute.return_value = mock_result
|
|
|
|
created_user = await create_user(mock_db_session, user_create_data)
|
|
mock_db_session.add.assert_called_once()
|
|
mock_db_session.flush.assert_called_once()
|
|
assert created_user is not None
|
|
assert created_user.email == user_create_data.email
|
|
assert created_user.name == user_create_data.name
|
|
assert created_user.id == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_user_email_already_registered(mock_db_session, user_create_data):
|
|
mock_db_session.flush.side_effect = IntegrityError("mock error (unique constraint)", "params", "orig")
|
|
with pytest.raises(EmailAlreadyRegisteredError):
|
|
await create_user(mock_db_session, user_create_data)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_user_db_integrity_error_not_unique(mock_db_session, user_create_data):
|
|
# Simulate an IntegrityError that is not related to a unique constraint
|
|
mock_db_session.flush.side_effect = IntegrityError("mock error (not unique constraint)", "params", "orig")
|
|
with pytest.raises(DatabaseIntegrityError):
|
|
await create_user(mock_db_session, user_create_data)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_user_db_connection_error(mock_db_session, user_create_data):
|
|
mock_db_session.begin.side_effect = OperationalError("mock_op_error", "params", "orig")
|
|
with pytest.raises(DatabaseConnectionError):
|
|
await create_user(mock_db_session, user_create_data)
|
|
# also test OperationalError on flush
|
|
mock_db_session.begin.side_effect = None # reset side effect
|
|
mock_db_session.flush.side_effect = OperationalError("mock_op_error", "params", "orig")
|
|
with pytest.raises(DatabaseConnectionError):
|
|
await create_user(mock_db_session, user_create_data)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_user_db_transaction_error(mock_db_session, user_create_data):
|
|
# Simulate a generic SQLAlchemyError on flush that is not IntegrityError or OperationalError
|
|
mock_db_session.flush.side_effect = UserCreationError("Simulated non-specific SQLAlchemyError") # Or any other SQLAlchemyError
|
|
with pytest.raises(DatabaseTransactionError):
|
|
await create_user(mock_db_session, user_create_data) |