Public Access
mirror of
https://github.com/xlorepdarkhelm/numinar-coding-project.git
synced 2026-09-10 18:22:42 -04:00
497 lines
17 KiB
Python
497 lines
17 KiB
Python
"""Unit tests for backend.services.rooms.
|
|
|
|
This test module covers all CRUD service logic for Room objects, including:
|
|
- Retrieval of all rooms and single rooms
|
|
- Creation, update, and deletion of rooms
|
|
- Error handling for not found and database exceptions
|
|
|
|
Style conventions:
|
|
- Google-style docstrings with "Asserts:" sections
|
|
- All test data provided via fixtures from conftest.py
|
|
- Constants used for error messages and status codes
|
|
- Parameterized tests for error scenarios and data variations
|
|
- Consistent blank lines and organized imports
|
|
- Explicit type annotations for all function signatures
|
|
|
|
All test data is managed through fixtures in conftest.py for maintainability and reuse.
|
|
"""
|
|
|
|
# Standard library imports
|
|
from typing import Any
|
|
from typing import Callable
|
|
from unittest.mock import AsyncMock
|
|
from unittest.mock import MagicMock
|
|
|
|
# Third-party imports
|
|
import pytest
|
|
from sqlalchemy import delete
|
|
from sqlalchemy import select
|
|
from sqlalchemy import update
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
# Local imports
|
|
from backend.models import Room
|
|
from backend.models import RoomList
|
|
from backend.services.rooms import delete_room
|
|
from backend.services.rooms import get_room
|
|
from backend.services.rooms import get_rooms
|
|
from backend.services.rooms import new_room
|
|
from backend.services.rooms import update_room
|
|
|
|
|
|
# Error message constants
|
|
ROOM_NOT_FOUND_MSG = "Room not found"
|
|
ROOM_DB_ERROR_MSG = "Database error"
|
|
|
|
# Not found ID constant for tests
|
|
ROOM_NOT_FOUND_ID = 999
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_rooms_returns_all_rooms(
|
|
async_session: AsyncSession, sample_rooms: RoomList
|
|
) -> None:
|
|
"""Test successful retrieval of all rooms from the database.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
sample_rooms: The sample list of Room objects.
|
|
|
|
Asserts:
|
|
- result is a list matching sample_rooms
|
|
- async_session.scalars called once with ``select(Room)``
|
|
"""
|
|
mock_scalars_result = AsyncMock()
|
|
mock_scalars_result.all = MagicMock(return_value=sample_rooms)
|
|
mock_scalars = AsyncMock(return_value=mock_scalars_result)
|
|
async_session.scalars = mock_scalars # type: ignore [method-assign]
|
|
|
|
result: RoomList = await get_rooms(async_session)
|
|
|
|
assert isinstance(result, list)
|
|
assert len(result) == 2
|
|
assert result == sample_rooms
|
|
async_session.scalars.assert_called_once()
|
|
assert async_session.scalars.call_args.args[0].compare(select(Room))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_rooms_returns_empty_list(async_session: AsyncSession) -> None:
|
|
"""Test retrieval of rooms when the database is empty.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
|
|
Asserts:
|
|
- result is an empty list
|
|
- async_session.scalars called once with ``select(Room)``
|
|
"""
|
|
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: RoomList = await get_rooms(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(Room))
|
|
|
|
|
|
# Parametrized test for get_room and delete_room error cases
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"service_func, session_attr, arg",
|
|
[
|
|
(get_room, "scalar", "sample_room"),
|
|
(delete_room, "execute", "sample_room"),
|
|
],
|
|
ids=["get_room", "delete_room"],
|
|
)
|
|
async def test_get_room_and_delete_room_database_error(
|
|
async_session: AsyncSession,
|
|
service_func: Callable[[AsyncSession, int], Any],
|
|
session_attr: str,
|
|
arg: str,
|
|
request: pytest.FixtureRequest,
|
|
) -> None:
|
|
"""Test handling of database errors for get_room and delete_room.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
service_func: The service function to test (get_room or delete_room).
|
|
session_attr: The session method to mock ("scalar" or "execute").
|
|
arg: The fixture name for the room object.
|
|
request: The pytest request object for fixture access.
|
|
|
|
Asserts:
|
|
- SQLAlchemyError is raised
|
|
- session method is called once
|
|
- rollback is called once for delete_room
|
|
"""
|
|
obj = request.getfixturevalue(arg)
|
|
room_id = obj.id
|
|
setattr(
|
|
async_session,
|
|
session_attr,
|
|
AsyncMock(side_effect=SQLAlchemyError(ROOM_DB_ERROR_MSG)),
|
|
)
|
|
if service_func is delete_room:
|
|
async_session.rollback = AsyncMock() # type: ignore [method-assign]
|
|
async_session.commit = AsyncMock() # type: ignore [method-assign]
|
|
with pytest.raises(SQLAlchemyError):
|
|
await service_func(async_session, room_id)
|
|
session_method = getattr(async_session, session_attr)
|
|
session_method.assert_called_once()
|
|
if service_func is delete_room:
|
|
# Assert rollback called
|
|
assert isinstance(async_session.rollback, AsyncMock)
|
|
async_session.rollback.assert_called_once()
|
|
# Assert the SQL statement is correct
|
|
assert session_method.call_args.args[0].compare(
|
|
delete(Room).where(Room.id == room_id)
|
|
)
|
|
# Assert commit not called due to error
|
|
if isinstance(async_session.commit, AsyncMock):
|
|
assert async_session.commit.call_count == 0
|
|
|
|
|
|
# Parametrized test for new_room error case
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("sample_room", ["sample_room"], indirect=True)
|
|
async def test_new_room_database_error(
|
|
async_session: AsyncSession, sample_room: Room
|
|
) -> None:
|
|
"""Test handling of database errors for new_room using fixture.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
sample_room: The sample Room object fixture.
|
|
|
|
Asserts:
|
|
- SQLAlchemyError is raised
|
|
- async_session.add called once with ``room``
|
|
- async_session.commit called once
|
|
- async_session.rollback called once
|
|
"""
|
|
async_session.add = MagicMock() # type: ignore [method-assign]
|
|
async_session.commit = AsyncMock( # type: ignore [method-assign]
|
|
side_effect=SQLAlchemyError(ROOM_DB_ERROR_MSG)
|
|
)
|
|
async_session.rollback = AsyncMock() # type: ignore [method-assign]
|
|
with pytest.raises(SQLAlchemyError):
|
|
await new_room(async_session, sample_room)
|
|
async_session.add.assert_called_once_with(sample_room)
|
|
async_session.commit.assert_called_once()
|
|
async_session.rollback.assert_called_once()
|
|
|
|
|
|
# Parametrized test for update_room error case
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"sample_room, room_update_data",
|
|
[("sample_room", "room_update_data")],
|
|
indirect=True,
|
|
)
|
|
async def test_update_room_database_error_param(
|
|
async_session: AsyncSession, sample_room: Room, room_update_data: dict[str, Any]
|
|
) -> None:
|
|
"""Test handling of database errors for update_room using fixtures.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
sample_room: The sample Room object fixture.
|
|
room_update_data: The update data for the room.
|
|
|
|
Asserts:
|
|
- SQLAlchemyError is raised
|
|
- async_session.execute called once
|
|
- async_session.rollback called once
|
|
"""
|
|
async_session.execute = AsyncMock( # type: ignore [method-assign]
|
|
side_effect=SQLAlchemyError(ROOM_DB_ERROR_MSG)
|
|
)
|
|
async_session.rollback = AsyncMock() # type: ignore [method-assign]
|
|
with pytest.raises(SQLAlchemyError):
|
|
await update_room(async_session, sample_room.id, **room_update_data)
|
|
async_session.execute.assert_called_once()
|
|
async_session.rollback.assert_called_once()
|
|
|
|
|
|
# Parametrized test for get_room_success (for all sample_rooms)
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"sample_room",
|
|
[
|
|
pytest.param(room, id=f"room_{room.id}")
|
|
for room in [
|
|
Room(
|
|
id=1,
|
|
name="Room 1",
|
|
location="Building A",
|
|
equipment="Projector",
|
|
capacity=10,
|
|
),
|
|
Room(
|
|
id=2,
|
|
name="Room 2",
|
|
location="Building B",
|
|
equipment="Whiteboard",
|
|
capacity=15,
|
|
),
|
|
]
|
|
],
|
|
indirect=True,
|
|
)
|
|
async def test_get_room_returns_room(
|
|
async_session: AsyncSession, sample_room: Room
|
|
) -> None:
|
|
"""Test successful retrieval of a room by ID using fixture data.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
sample_room: The sample Room object fixture.
|
|
|
|
Asserts:
|
|
- result matches sample_room
|
|
- async_session.scalar called once with
|
|
``select(Room).where(Room.id == sample_room.id)``
|
|
"""
|
|
async_session.scalar = AsyncMock( # type: ignore [method-assign]
|
|
return_value=sample_room
|
|
)
|
|
result: Room = await get_room(async_session, sample_room.id)
|
|
assert result == sample_room
|
|
async_session.scalar.assert_called_once()
|
|
assert async_session.scalar.call_args.args[0].compare(
|
|
select(Room).where(Room.id == sample_room.id)
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_room_returns_room_fixture(
|
|
async_session: AsyncSession, sample_room: Room
|
|
) -> None:
|
|
"""Test successful retrieval of a room by ID using fixture data.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
sample_room: The sample Room object fixture.
|
|
|
|
Asserts:
|
|
- result matches sample_room
|
|
- async_session.scalar called once with
|
|
``select(Room).where(Room.id == sample_room.id)``
|
|
"""
|
|
async_session.scalar = AsyncMock(return_value=sample_room) # type: ignore [method-assign]
|
|
|
|
result: Room = await get_room(async_session, sample_room.id)
|
|
|
|
assert result == sample_room
|
|
async_session.scalar.assert_called_once()
|
|
assert async_session.scalar.call_args.args[0].compare(
|
|
select(Room).where(Room.id == sample_room.id)
|
|
)
|
|
"""Test handling of database errors in new_room.
|
|
|
|
Verifies that new_room rolls back the session on database failure.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
|
|
Asserts:
|
|
- SQLAlchemyError is raised
|
|
- async_session.add called once with room
|
|
- async_session.commit called once
|
|
- async_session.rollback called once
|
|
"""
|
|
async_session.add = MagicMock() # type: ignore [method-assign]
|
|
async_session.commit = AsyncMock( # type: ignore [method-assign]
|
|
side_effect=SQLAlchemyError(ROOM_DB_ERROR_MSG)
|
|
)
|
|
async_session.rollback = AsyncMock() # type: ignore [method-assign]
|
|
|
|
with pytest.raises(SQLAlchemyError):
|
|
await new_room(async_session, sample_room)
|
|
async_session.add.assert_called_once_with(sample_room)
|
|
async_session.commit.assert_called_once()
|
|
async_session.rollback.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_room_returns_updated_room(
|
|
async_session: AsyncSession,
|
|
sample_room: Room,
|
|
room_update_data: dict[str, Any],
|
|
updated_room: Room,
|
|
) -> None:
|
|
"""Test successful update of a room.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
sample_room: The sample Room object fixture.
|
|
room_update_data: The update data for the room.
|
|
updated_room: The expected updated Room object.
|
|
|
|
Asserts:
|
|
- result matches updated_room
|
|
- async_session.execute called once with
|
|
``update(Room).where(Room.id == room_id).values(**update_params)``
|
|
- async_session.commit called once
|
|
- async_session.scalar called once with ``select(Room).where(Room.id == room_id)``
|
|
"""
|
|
room_id = sample_room.id
|
|
update_params = room_update_data
|
|
mock_execute_result = MagicMock(rowcount=1)
|
|
async_session.execute = AsyncMock( # type: ignore [method-assign]
|
|
return_value=mock_execute_result
|
|
)
|
|
async_session.commit = AsyncMock() # type: ignore [method-assign]
|
|
async_session.scalar = AsyncMock( # type: ignore [method-assign]
|
|
return_value=updated_room
|
|
)
|
|
|
|
result: Room = await update_room(async_session, room_id, **update_params)
|
|
|
|
assert result == updated_room
|
|
async_session.execute.assert_called_once()
|
|
assert async_session.execute.call_args.args[0].compare(
|
|
update(Room).where(Room.id == room_id).values(**update_params)
|
|
)
|
|
async_session.commit.assert_called_once()
|
|
async_session.scalar.assert_called_once()
|
|
assert async_session.scalar.call_args.args[0].compare(
|
|
select(Room).where(Room.id == room_id)
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_room_raises_not_found(async_session: AsyncSession) -> None:
|
|
"""Test handling of non-existent room in update_room.
|
|
|
|
Verifies that update_room raises ValueError when the room is not found.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
|
|
Asserts:
|
|
- ValueError is raised
|
|
- async_session.execute called once with
|
|
``update(Room).where(Room.id == room_id).values(**update_params)``
|
|
- async_session.rollback called once
|
|
"""
|
|
room_id = ROOM_NOT_FOUND_ID
|
|
update_params: dict[str, Any] = {
|
|
"name": "Updated Room",
|
|
"location": "Building 2",
|
|
"equipment": "Whiteboard",
|
|
"capacity": 15,
|
|
}
|
|
mock_execute_result = MagicMock(rowcount=0)
|
|
async_session.execute = AsyncMock( # type: ignore [method-assign]
|
|
return_value=mock_execute_result
|
|
)
|
|
async_session.rollback = AsyncMock() # type: ignore [method-assign]
|
|
|
|
with pytest.raises(ValueError):
|
|
await update_room(async_session, room_id, **update_params)
|
|
async_session.execute.assert_called_once()
|
|
assert async_session.execute.call_args.args[0].compare(
|
|
update(Room).where(Room.id == room_id).values(**update_params)
|
|
)
|
|
async_session.rollback.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_room_database_error(
|
|
async_session: AsyncSession, sample_room: Room, room_update_data: dict[str, Any]
|
|
) -> None:
|
|
"""Test handling of database errors in update_room.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
sample_room: The sample Room object fixture.
|
|
room_update_data: The update data for the room.
|
|
|
|
Asserts:
|
|
- SQLAlchemyError is raised
|
|
- async_session.execute called once
|
|
- async_session.rollback called once
|
|
"""
|
|
room_id = sample_room.id
|
|
update_params = room_update_data
|
|
async_session.execute = AsyncMock( # type: ignore [method-assign]
|
|
side_effect=SQLAlchemyError(ROOM_DB_ERROR_MSG)
|
|
)
|
|
async_session.rollback = AsyncMock() # type: ignore [method-assign]
|
|
|
|
with pytest.raises(SQLAlchemyError):
|
|
await update_room(async_session, room_id, **update_params)
|
|
async_session.execute.assert_called_once()
|
|
assert async_session.execute.call_args.args[0].compare(
|
|
update(Room).where(Room.id == room_id).values(**update_params)
|
|
)
|
|
async_session.rollback.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_room_commits_success(
|
|
async_session: AsyncSession, sample_room: Room
|
|
) -> None:
|
|
"""Test successful deletion of a room.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
sample_room: The sample Room object fixture.
|
|
|
|
Asserts:
|
|
- async_session.execute called once with ``delete(Room).where(Room.id == room_id)``
|
|
- async_session.commit called once
|
|
"""
|
|
room_id = sample_room.id
|
|
mock_execute_result = MagicMock(rowcount=1)
|
|
async_session.execute = AsyncMock( # type: ignore [method-assign]
|
|
return_value=mock_execute_result
|
|
)
|
|
async_session.commit = AsyncMock() # type: ignore [method-assign]
|
|
|
|
await delete_room(async_session, room_id)
|
|
|
|
async_session.execute.assert_called_once()
|
|
assert async_session.execute.call_args.args[0].compare(
|
|
delete(Room).where(Room.id == room_id)
|
|
)
|
|
async_session.commit.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_room_raises_not_found(async_session: AsyncSession) -> None:
|
|
"""Test handling of non-existent room in delete_room.
|
|
|
|
Verifies that delete_room raises ValueError when the room is not found.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
|
|
Asserts:
|
|
- ValueError is raised
|
|
- async_session.execute called once with ``delete(Room).where(Room.id == room_id)``
|
|
- async_session.rollback called once
|
|
"""
|
|
room_id = ROOM_NOT_FOUND_ID
|
|
mock_execute_result = MagicMock(rowcount=0)
|
|
async_session.execute = AsyncMock( # type: ignore [method-assign]
|
|
return_value=mock_execute_result
|
|
)
|
|
async_session.rollback = AsyncMock() # type: ignore [method-assign]
|
|
|
|
with pytest.raises(ValueError):
|
|
await delete_room(async_session, room_id)
|
|
async_session.execute.assert_called_once()
|
|
assert async_session.execute.call_args.args[0].compare(
|
|
delete(Room).where(Room.id == room_id)
|
|
)
|
|
async_session.rollback.assert_called_once()
|