mirror of
https://github.com/xlorepdarkhelm/numinar-coding-project.git
synced 2026-09-09 11:19:49 -04:00
599 lines
20 KiB
Python
599 lines
20 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
|
|
from unittest.mock import patch
|
|
|
|
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.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
|
|
|
|
# Third-party imports
|
|
from backend.types import RoomData
|
|
|
|
|
|
# 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: list[Room]
|
|
) -> 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: list[Room] = 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: list[Room] = 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))
|
|
|
|
|
|
class DummyError(Exception):
|
|
"""Custom exception for generic error branch coverage in service tests."""
|
|
|
|
pass
|
|
|
|
|
|
# Parametrized test for get_room and delete_room error cases
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"service_func, session_attr, arg, exc_type",
|
|
[
|
|
(get_room, "scalar", "sample_room", SQLAlchemyError),
|
|
(
|
|
get_room,
|
|
"scalar",
|
|
"sample_room",
|
|
DummyError,
|
|
), # custom exception for coverage
|
|
(delete_room, "execute", "sample_room", SQLAlchemyError),
|
|
],
|
|
ids=["get_room_sqlalchemy", "get_room_dummy", "delete_room_sqlalchemy"],
|
|
)
|
|
async def test_get_room_and_delete_room_database_error(
|
|
async_session: AsyncSession,
|
|
service_func: Callable[[AsyncSession, int], Any],
|
|
session_attr: str,
|
|
arg: str,
|
|
exc_type: type[Exception],
|
|
request: pytest.FixtureRequest,
|
|
) -> None:
|
|
"""Test error handling and finally block coverage for get_room and delete_room.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session fixture.
|
|
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.
|
|
exc_type: The exception type to raise.
|
|
request: The pytest request object for fixture access.
|
|
|
|
Asserts:
|
|
- Exception of exc_type is raised.
|
|
- Session method is called once.
|
|
- Rollback is called once for delete_room.
|
|
- Commit is not called for delete_room on error.
|
|
- Logger debug is called (finally block).
|
|
"""
|
|
obj = request.getfixturevalue(arg)
|
|
room_id = obj.id
|
|
setattr(
|
|
async_session,
|
|
session_attr,
|
|
AsyncMock(side_effect=exc_type("fail")),
|
|
)
|
|
if service_func is delete_room:
|
|
async_session.rollback = AsyncMock() # type: ignore [method-assign]
|
|
async_session.commit = AsyncMock() # type: ignore [method-assign]
|
|
with patch("backend.services.rooms.logger.debug") as mock_debug:
|
|
with pytest.raises(exc_type):
|
|
await service_func(async_session, room_id)
|
|
assert mock_debug.called
|
|
session_method = getattr(async_session, session_attr)
|
|
session_method.assert_called_once()
|
|
if service_func is delete_room:
|
|
assert isinstance(async_session.rollback, AsyncMock)
|
|
async_session.rollback.assert_called_once()
|
|
assert session_method.call_args.args[0].compare(
|
|
delete(Room).where(Room.id == room_id)
|
|
)
|
|
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, exc_type",
|
|
[
|
|
("sample_room", SQLAlchemyError),
|
|
("sample_room", DummyError), # custom exception for coverage
|
|
],
|
|
indirect=["sample_room"],
|
|
ids=["sqlalchemy_error", "dummy_error"],
|
|
)
|
|
async def test_new_room_database_error(
|
|
async_session: AsyncSession,
|
|
sample_room: Room,
|
|
sample_room_data: RoomData,
|
|
exc_type: type[Exception],
|
|
) -> None:
|
|
"""Test error handling and finally block coverage for new_room.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session fixture.
|
|
sample_room: The sample Room object fixture.
|
|
sample_room_data: The data for creating the room.
|
|
exc_type: The exception type to raise.
|
|
|
|
Asserts:
|
|
- Exception of exc_type is raised.
|
|
- add called once with sample_room.
|
|
- commit called once.
|
|
- rollback called once.
|
|
- Logger debug is called (finally block).
|
|
"""
|
|
async_session.add = MagicMock() # type: ignore [method-assign]
|
|
async_session.commit = AsyncMock( # type: ignore [method-assign]
|
|
side_effect=exc_type("fail")
|
|
)
|
|
async_session.rollback = AsyncMock() # type: ignore [method-assign]
|
|
with patch("backend.services.rooms.logger.debug") as mock_debug:
|
|
with pytest.raises(exc_type):
|
|
await new_room(async_session, **sample_room_data)
|
|
assert mock_debug.called
|
|
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)
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_room_commits_errors(
|
|
async_session: AsyncSession, sample_room: Room, sample_room_data: RoomData
|
|
) -> None:
|
|
"""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.
|
|
sample_room: The sample Room object fixture.
|
|
sample_room_data: The data for creating the room.
|
|
|
|
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_data)
|
|
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()
|
|
|
|
|
|
# Patch logger.debug for finally block coverage using correct import path
|
|
LOGGER_PATH = "backend.services.rooms.logger.debug"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"service_func, args, session_attr",
|
|
[
|
|
(get_rooms, (), "scalars"),
|
|
(get_room, (ROOM_NOT_FOUND_ID,), "scalar"),
|
|
(
|
|
new_room,
|
|
(Room(id=999, name="fail", location="", equipment="", capacity=0),),
|
|
"commit",
|
|
),
|
|
],
|
|
)
|
|
async def test_services_rooms_finally_and_exception_logging(
|
|
async_session: AsyncSession,
|
|
service_func: Callable[..., Any],
|
|
args: tuple[Any, ...],
|
|
session_attr: str,
|
|
) -> None:
|
|
"""Test that finally blocks and generic exception logging are covered.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
service_func: The service function to test.
|
|
args: The arguments to pass to the service function.
|
|
session_attr: The session method to mock.
|
|
|
|
Asserts:
|
|
- Exception is raised
|
|
- Logger error is called
|
|
- Logger debug is called
|
|
"""
|
|
with patch(LOGGER_PATH) as mock_debug, patch(
|
|
"backend.services.rooms.logger.error"
|
|
) as mock_error, patch("backend.services.rooms.logger.info"):
|
|
# Patch session methods to raise Exception
|
|
if service_func is get_rooms:
|
|
setattr(
|
|
async_session, session_attr, AsyncMock(side_effect=Exception("fail"))
|
|
)
|
|
elif service_func is get_room:
|
|
setattr(
|
|
async_session, session_attr, AsyncMock(side_effect=Exception("fail"))
|
|
)
|
|
elif service_func is new_room:
|
|
async_session.add = MagicMock() # type: ignore [method-assign]
|
|
async_session.commit = AsyncMock( # type: ignore [method-assign]
|
|
side_effect=Exception("fail")
|
|
)
|
|
async_session.rollback = AsyncMock() # type: ignore [method-assign]
|
|
with pytest.raises(Exception): # noqa: B017
|
|
await service_func(async_session, *args)
|
|
# Check that error and debug logging were called
|
|
assert mock_error.called
|
|
assert mock_debug.called
|