mirror of
https://github.com/xlorepdarkhelm/numinar-coding-project.git
synced 2026-09-06 23:38:24 -04:00
454 lines
15 KiB
Python
454 lines
15 KiB
Python
"""Unit tests for the backend.services.rooms module."""
|
|
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from sqlalchemy import delete
|
|
from sqlalchemy import select
|
|
from sqlalchemy import update
|
|
from sqlalchemy.exc import NoResultFound
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.rooms"], indirect=True)
|
|
async def test_get_rooms_success(
|
|
async_session: AsyncSession, sample_rooms: RoomList, mock_logger: MagicMock
|
|
) -> None:
|
|
"""Test successful retrieval of all rooms from the database.
|
|
|
|
Verifies that get_rooms returns the expected list of rooms and constructs
|
|
the correct SQLAlchemy query.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
sample_rooms: The mocked list of rooms to return.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
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
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.rooms"], indirect=True)
|
|
async def test_get_rooms_empty(
|
|
async_session: AsyncSession, mock_logger: MagicMock
|
|
) -> None:
|
|
"""Test retrieval of rooms when the database is empty.
|
|
|
|
Verifies that get_rooms returns an empty list when no rooms are found.
|
|
|
|
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: 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))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.rooms"], indirect=True)
|
|
async def test_get_rooms_database_error(
|
|
async_session: AsyncSession, mock_logger: MagicMock
|
|
) -> None:
|
|
"""Test handling of database errors in get_rooms.
|
|
|
|
Verifies that get_rooms 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_rooms(async_session)
|
|
async_session.scalars.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.rooms"], indirect=True)
|
|
@pytest.mark.parametrize(
|
|
"room_id, expected_name",
|
|
[
|
|
(1, "Room A"),
|
|
(2, "Room B"),
|
|
],
|
|
ids=["room1", "room2"],
|
|
)
|
|
async def test_get_room_success(
|
|
async_session: AsyncSession,
|
|
room_id: int,
|
|
expected_name: str,
|
|
mock_logger: MagicMock,
|
|
) -> None:
|
|
"""Test successful retrieval of a room by ID.
|
|
|
|
Verifies that get_room returns the correct room and constructs the correct query.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
room_id: The ID of the room to retrieve.
|
|
expected_name: The expected name of the room.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
room = Room(
|
|
id=room_id,
|
|
name=expected_name,
|
|
location="Building 1",
|
|
equipment="Projector",
|
|
capacity=10,
|
|
)
|
|
async_session.scalar = AsyncMock(return_value=room) # type: ignore [method-assign]
|
|
|
|
result: Room = await get_room(async_session, room_id)
|
|
|
|
assert result.id == room_id
|
|
assert result.name == expected_name
|
|
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
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.rooms"], indirect=True)
|
|
@pytest.mark.parametrize(
|
|
"room_id",
|
|
[999, -1],
|
|
ids=["nonexistent_id", "invalid_id"],
|
|
)
|
|
async def test_get_room_not_found(
|
|
async_session: AsyncSession, mock_logger: MagicMock, room_id: int
|
|
) -> None:
|
|
"""Test handling of non-existent room in get_room.
|
|
|
|
Verifies that get_room raises NoResultFound when the room is not found.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
mock_logger: The mocked logger instance.
|
|
room_id: The ID of the room to retrieve.
|
|
"""
|
|
async_session.scalar = AsyncMock(return_value=None) # type: ignore [method-assign]
|
|
|
|
with pytest.raises(NoResultFound):
|
|
await get_room(async_session, room_id)
|
|
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
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.rooms"], indirect=True)
|
|
async def test_get_room_database_error(
|
|
async_session: AsyncSession, mock_logger: MagicMock
|
|
) -> None:
|
|
"""Test handling of database errors in get_room.
|
|
|
|
Verifies that get_room raises an exception on database failure.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
room_id: int = 1
|
|
async_session.scalar = AsyncMock( # type: ignore [method-assign]
|
|
side_effect=SQLAlchemyError("Database error")
|
|
)
|
|
|
|
with pytest.raises(SQLAlchemyError):
|
|
await get_room(async_session, room_id)
|
|
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
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.rooms"], indirect=True)
|
|
async def test_new_room_success(
|
|
async_session: AsyncSession, mock_logger: MagicMock
|
|
) -> None:
|
|
"""Test successful creation of a new room.
|
|
|
|
Verifies that new_room adds the room, commits the session, and returns the room.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
room = Room(
|
|
id=1, name="Room A", location="Building 1", equipment="Projector", capacity=10
|
|
)
|
|
async_session.add = MagicMock() # type: ignore [method-assign]
|
|
async_session.commit = AsyncMock() # type: ignore [method-assign]
|
|
|
|
result: Room = await new_room(async_session, room)
|
|
|
|
assert result == room
|
|
async_session.add.assert_called_once_with(room)
|
|
async_session.commit.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.rooms"], indirect=True)
|
|
async def test_new_room_database_error(
|
|
async_session: AsyncSession, mock_logger: MagicMock
|
|
) -> 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.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
room = Room(
|
|
id=1, name="Room A", location="Building 1", equipment="Projector", capacity=10
|
|
)
|
|
async_session.add = MagicMock() # type: ignore [method-assign]
|
|
async_session.commit = AsyncMock( # type: ignore [method-assign]
|
|
side_effect=SQLAlchemyError("Database error")
|
|
)
|
|
async_session.rollback = AsyncMock() # type: ignore [method-assign]
|
|
|
|
with pytest.raises(SQLAlchemyError):
|
|
await new_room(async_session, room)
|
|
async_session.add.assert_called_once_with(room)
|
|
async_session.commit.assert_called_once()
|
|
async_session.rollback.assert_called_once()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("mock_logger", ["backend.routers.rooms"], indirect=True)
|
|
async def test_update_room_success(
|
|
async_session: AsyncSession, mock_logger: MagicMock
|
|
) -> None:
|
|
"""Test successful update of a room.
|
|
|
|
Verifies that update_room updates the room, commits the session, and returns
|
|
the updated room.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
room_id = 1
|
|
update_params: dict[str, Any] = {
|
|
"name": "Updated Room",
|
|
"location": "Building 2",
|
|
"equipment": "Whiteboard",
|
|
"capacity": 15,
|
|
}
|
|
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]
|
|
updated_room = Room(id=room_id, **update_params)
|
|
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
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.rooms"], indirect=True)
|
|
async def test_update_room_not_found(
|
|
async_session: AsyncSession, mock_logger: MagicMock
|
|
) -> 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.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
room_id = 999
|
|
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
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.rooms"], indirect=True)
|
|
async def test_update_room_database_error(
|
|
async_session: AsyncSession, mock_logger: MagicMock
|
|
) -> None:
|
|
"""Test handling of database errors in update_room.
|
|
|
|
Verifies that update_room rolls back the session on database failure.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
room_id = 1
|
|
update_params: dict[str, Any] = {
|
|
"name": "Updated Room",
|
|
"location": "Building 2",
|
|
"equipment": "Whiteboard",
|
|
"capacity": 15,
|
|
}
|
|
async_session.execute = AsyncMock( # type: ignore [method-assign]
|
|
side_effect=SQLAlchemyError("Database error")
|
|
)
|
|
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
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.rooms"], indirect=True)
|
|
async def test_delete_room_success(
|
|
async_session: AsyncSession, mock_logger: MagicMock
|
|
) -> None:
|
|
"""Test successful deletion of a room.
|
|
|
|
Verifies that delete_room deletes the room and commits the session.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
room_id = 1
|
|
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
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.rooms"], indirect=True)
|
|
async def test_delete_room_not_found(
|
|
async_session: AsyncSession, mock_logger: MagicMock
|
|
) -> 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.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
room_id = 999
|
|
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()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("mock_logger", ["backend.services.rooms"], indirect=True)
|
|
async def test_delete_room_database_error(
|
|
async_session: AsyncSession, mock_logger: MagicMock
|
|
) -> None:
|
|
"""Test handling of database errors in delete_room.
|
|
|
|
Verifies that delete_room rolls back the session on database failure.
|
|
|
|
Args:
|
|
async_session: The asynchronous database session.
|
|
mock_logger: The mocked logger instance.
|
|
"""
|
|
room_id = 1
|
|
async_session.execute = AsyncMock( # type: ignore [method-assign]
|
|
side_effect=SQLAlchemyError("Database error")
|
|
)
|
|
async_session.rollback = AsyncMock() # type: ignore [method-assign]
|
|
|
|
with pytest.raises(SQLAlchemyError):
|
|
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()
|