Public Access
mirror of
https://github.com/xlorepdarkhelm/numinar-coding-project.git
synced 2026-09-10 18:22:42 -04:00
85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
"""Unit tests for the backend.services.users module."""
|
|
|
|
from unittest.mock import AsyncMock
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from backend.models import User # Add this import for the User model
|
|
from backend.services.users import get_users
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.users"], indirect=True)
|
|
async def test_get_users_success(
|
|
async_session: AsyncSession, sample_users: list[User], mock_logger: MagicMock
|
|
) -> None:
|
|
"""Test successful retrieval of all users from the database.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
sample_users: The list of sample user objects to return.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
mock_scalars_result = AsyncMock()
|
|
mock_scalars_result.all = MagicMock(return_value=sample_users)
|
|
mock_scalars = AsyncMock(return_value=mock_scalars_result)
|
|
async_session.scalars = mock_scalars # type: ignore [method-assign]
|
|
|
|
result: list[User] = await get_users(async_session)
|
|
|
|
assert isinstance(result, list)
|
|
assert len(result) == 2
|
|
assert result == sample_users
|
|
async_session.scalars.assert_called_once()
|
|
assert async_session.scalars.call_args.args[0].compare(select(User))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.users"], indirect=True)
|
|
async def test_get_users_empty(
|
|
async_session: AsyncSession, mock_logger: MagicMock
|
|
) -> None:
|
|
"""Test retrieval of users when the database is empty.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
mock_scalars_result = AsyncMock()
|
|
mock_scalars_result.all = MagicMock(return_value=[])
|
|
mock_scalars = AsyncMock(return_value=mock_scalars_result)
|
|
async_session.scalars = mock_scalars # type: ignore [method-assign]
|
|
|
|
result: list[User] = await get_users(async_session)
|
|
|
|
assert isinstance(result, list)
|
|
assert len(result) == 0
|
|
async_session.scalars.assert_called_once()
|
|
assert async_session.scalars.call_args.args[0].compare(select(User))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.users"], indirect=True)
|
|
async def test_get_users_database_error(
|
|
async_session: AsyncSession, mock_logger: MagicMock
|
|
) -> None:
|
|
"""Test handling of database errors in get_users.
|
|
|
|
Verifies that get_users raises an exception on database failure.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
async_session.scalars = AsyncMock( # type: ignore [method-assign]
|
|
side_effect=SQLAlchemyError("Database error")
|
|
)
|
|
|
|
with pytest.raises(SQLAlchemyError):
|
|
await get_users(async_session)
|
|
async_session.scalars.assert_called_once()
|