"""Unit tests for the backend.services.bookings module.""" import sys from collections.abc import Awaitable from datetime import datetime from datetime import timedelta from datetime import timezone 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 NoResultFound from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession from backend.models import Booking from backend.models import BookingList from backend.services.bookings import delete_booking from backend.services.bookings import get_booking from backend.services.bookings import get_bookings_for_room from backend.services.bookings import new_booking from backend.services.bookings import update_booking def _config_side_effect(key: str, *args: Any, **kwargs: Any) -> int | None: if key == "BOOKING_MAX_MONTHS": return 12 if "max_months" not in kwargs else kwargs["max_months"] return kwargs.get("default") @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_get_bookings_for_room_success( async_session: AsyncSession, sample_bookings: BookingList, mock_logger: MagicMock ) -> None: """Test retrieval of all bookings for a room with no date filter. Args: async_session: The asynchronous database session. sample_bookings: The mocked list of bookings to return. mock_logger: The mocked logger instance. """ room_id = 1 mock_scalars_result = AsyncMock() mock_scalars_result.all = MagicMock(return_value=sample_bookings) mock_scalars = AsyncMock(return_value=mock_scalars_result) with patch.object(async_session, "scalars", mock_scalars) as scalars_mock: result: BookingList = await get_bookings_for_room(async_session, room_id) assert isinstance(result, list) assert len(result) == 2 assert result == sample_bookings scalars_mock.assert_called_once() assert scalars_mock.call_args.args[0].compare( select(Booking).where(Booking.room_id == room_id) ) @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_get_bookings_for_room_with_date( async_session: AsyncSession, sample_bookings: BookingList, mock_logger: MagicMock ) -> None: """Test retrieval of bookings for a room filtered by date. Args: async_session: The asynchronous database session. sample_bookings: The mocked list of bookings to return. mock_logger: The mocked logger instance. """ room_id = 1 date = "2025-08-28" mock_scalars_result = AsyncMock() mock_scalars_result.all = MagicMock(return_value=sample_bookings) mock_scalars = AsyncMock(return_value=mock_scalars_result) with patch.object(async_session, "scalars", mock_scalars) as scalars_mock: result: BookingList = await get_bookings_for_room(async_session, room_id, date) assert isinstance(result, list) assert result == sample_bookings scalars_mock.assert_called_once() # Check the query includes the date filter stmt = scalars_mock.call_args.args[0] assert stmt.whereclause is not None assert str(stmt.whereclause).find("start_time >=") != -1 assert str(stmt.whereclause).find("start_time <") != -1 @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_get_bookings_for_room_invalid_date( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test that an invalid date string raises ValueError. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ room_id = 1 invalid_date = "not-a-date" with pytest.raises(ValueError): await get_bookings_for_room(async_session, room_id, invalid_date) @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_get_bookings_for_room_empty( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test retrieval of bookings when none exist for the room. Verifies that get_bookings_for_room returns an empty list when no bookings are found. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ room_id = 1 mock_scalars_result = AsyncMock() mock_scalars_result.all = MagicMock(return_value=[]) mock_scalars = AsyncMock(return_value=mock_scalars_result) with patch.object(async_session, "scalars", mock_scalars) as scalars_mock: result: BookingList = await get_bookings_for_room(async_session, room_id) assert isinstance(result, list) assert len(result) == 0 scalars_mock.assert_called_once() assert scalars_mock.call_args.args[0].compare( select(Booking).where(Booking.room_id == room_id) ) @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_get_bookings_for_room_database_error( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test handling of database errors in get_bookings_for_room. Verifies that get_bookings_for_room raises an exception on database failure. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ room_id = 1 with patch.object( async_session, "scalars", AsyncMock(side_effect=SQLAlchemyError("Database error")), ) as scalars_mock: with pytest.raises(SQLAlchemyError): await get_bookings_for_room(async_session, room_id) scalars_mock.assert_called_once() @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) @pytest.mark.parametrize( "booking_id, expected_room_id", [ (1, 1), (2, 2), ], ids=["booking1", "booking2"], ) async def test_get_booking_success( async_session: AsyncSession, booking_id: int, expected_room_id: int, mock_logger: MagicMock, ) -> None: """Test successful retrieval of a booking by ID. Verifies that get_booking returns the correct booking and constructs the correct query. Args: async_session: The asynchronous database session. booking_id: The ID of the booking to retrieve. expected_room_id: The expected room ID of the booking. mock_logger: The mocked logger instance. """ booking = Booking( room_id=expected_room_id, start_time=datetime.now(timezone.utc) + timedelta(hours=1), end_time=datetime.now(timezone.utc) + timedelta(hours=2), title=f"Service Test Booking {booking_id}", ) booking.id = booking_id async_session.scalar = AsyncMock(return_value=booking) # type: ignore [method-assign] result: Booking = await get_booking(async_session, booking_id) assert result.id == booking_id assert result.room_id == expected_room_id async_session.scalar.assert_called_once() assert async_session.scalar.call_args.args[0].compare( select(Booking).where(Booking.id == booking_id) ) @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) @pytest.mark.parametrize( "booking_id", [999, -1], ids=["nonexistent_id", "invalid_id"], ) async def test_get_booking_not_found( async_session: AsyncSession, mock_logger: MagicMock, booking_id: int ) -> None: """Test handling of non-existent booking in get_booking. Verifies that get_booking raises NoResultFound when the booking is not found. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. booking_id: The ID of the booking to retrieve. """ async_session.scalar = AsyncMock(return_value=None) # type: ignore [method-assign] with pytest.raises(NoResultFound): await get_booking(async_session, booking_id) async_session.scalar.assert_called_once() assert async_session.scalar.call_args.args[0].compare( select(Booking).where(Booking.id == booking_id) ) @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_get_booking_database_error( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test handling of database errors in get_booking. Verifies that get_booking raises an exception on database failure. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ booking_id = 1 async_session.scalar = AsyncMock( # type: ignore [method-assign] side_effect=SQLAlchemyError("Database error") ) with pytest.raises(SQLAlchemyError): await get_booking(async_session, booking_id) async_session.scalar.assert_called_once() assert async_session.scalar.call_args.args[0].compare( select(Booking).where(Booking.id == booking_id) ) @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_new_booking_success( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test successful creation of a new booking. Verifies that new_booking adds the booking, commits the session, and returns the booking. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ booking = Booking( room_id=1, start_time=datetime.now(timezone.utc) + timedelta(hours=1), end_time=datetime.now(timezone.utc) + timedelta(hours=2), title="Service Test Booking", ) booking.id = 1 async_session.add = MagicMock() # type: ignore [method-assign] async_session.commit = AsyncMock() # type: ignore [method-assign] # Patch get_room to always exist with ( patch( "backend.services.bookings.get_room", new=AsyncMock(return_value=MagicMock(capacity=10)), ), patch( "backend.services.bookings.config", wraps=sys.modules["backend"].config ) as mock_config, ): mock_config.side_effect = _config_side_effect # Patch session.scalars to simulate no overlap mock_scalars_result = AsyncMock() mock_scalars_result.first = MagicMock(return_value=None) with patch.object( async_session, "scalars", AsyncMock(return_value=mock_scalars_result) ): # Patch get_invitees_for_booking to return 0 invitees with patch( "backend.services.bookings.get_invitees_for_booking", new=AsyncMock(return_value=[]), ): result: Booking = await new_booking(async_session, booking) assert result == booking async_session.add.assert_called_once_with(booking) async_session.commit.assert_called_once() # --- Constraint tests --- @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) @pytest.mark.parametrize( "room_exists, start_time, end_time, max_months, expected_error", [ # Booking in the past ( True, datetime.now(timezone.utc) - timedelta(days=1), datetime.now(timezone.utc) + timedelta(hours=1), 12, "Bookings cannot be made in the past.", ), # Start >= end ( True, datetime.now(timezone.utc) + timedelta(days=1), datetime.now(timezone.utc) + timedelta(days=1), 12, "The booking start time must be before the end time.", ), # Too far in future ( True, datetime.now(timezone.utc) + timedelta(days=365 * 2), datetime.now(timezone.utc) + timedelta(days=365 * 2, hours=1), 12, "Bookings can only be made up to 12 months in advance.", ), # Non-existent room ( False, datetime.now(timezone.utc) + timedelta(days=1), datetime.now(timezone.utc) + timedelta(days=2), 12, "The selected room does not exist.", ), ], ids=["past", "start>=end", "future", "no_room"], ) async def test_new_booking_constraints( async_session: AsyncSession, mock_logger: MagicMock, room_exists: bool, start_time: datetime, end_time: datetime, max_months: int, expected_error: str, ) -> None: """Test booking constraints: past, start>=end, future, non-existent room. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. room_exists: Flag indicating if the room exists. start_time: The proposed start time for the booking. end_time: The proposed end time for the booking. max_months: The maximum number of months in advance bookings can be made. expected_error: The expected error message. """ from backend.models import Booking from backend.services.bookings import new_booking # Ensure start_time == end_time exactly for the start>=end test if expected_error == "The booking start time must be before the end time.": start_time = end_time = datetime.now(timezone.utc) + timedelta(days=1) booking: Booking = Booking(room_id=1, start_time=start_time, end_time=end_time) booking.id = 1 # Patch get_room to simulate room existence get_room_patch: Callable[[], Awaitable[MagicMock]] = ( AsyncMock(return_value=MagicMock(capacity=10)) if room_exists else AsyncMock(side_effect=Exception("no room")) ) with ( patch("backend.services.bookings.get_room", new=get_room_patch), patch( "backend.services.bookings.config", wraps=sys.modules["backend"].config ) as mock_config, ): def _config_side_effect_max_months( key: str, *args: Any, **kwargs: Any ) -> int | None: return max_months if key == "BOOKING_MAX_MONTHS" else kwargs.get("default") mock_config.side_effect = _config_side_effect_max_months # Patch session.scalars to simulate no overlap mock_scalars_result = AsyncMock() mock_scalars_result.first = MagicMock(return_value=None) with ( patch.object( async_session, "scalars", AsyncMock(return_value=mock_scalars_result), ), patch( "backend.services.bookings.get_invitees_for_booking", new=AsyncMock(return_value=[]), ), ): with pytest.raises(ValueError) as exc: await new_booking(async_session, booking) assert expected_error in str(exc.value) @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_new_booking_overlap( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test overlapping booking constraint. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ from backend.models import Booking from backend.services.bookings import new_booking # Patch get_room to always exist with ( patch( "backend.services.bookings.get_room", new=AsyncMock(return_value=MagicMock(capacity=10)), ), patch( "backend.services.bookings.config", wraps=sys.modules["backend"].config ) as mock_config, ): mock_config.side_effect = _config_side_effect # Patch session.scalars to simulate overlap overlap_booking: Booking = Booking( room_id=1, start_time=datetime.now(timezone.utc) + timedelta(days=1), end_time=datetime.now(timezone.utc) + timedelta(days=2), ) overlap_booking.id = 2 mock_scalars_result = AsyncMock() mock_scalars_result.first = MagicMock(return_value=overlap_booking) with patch.object( async_session, "scalars", AsyncMock(return_value=mock_scalars_result) ): booking: Booking = Booking( room_id=1, start_time=datetime.now(timezone.utc) + timedelta(days=1), end_time=datetime.now(timezone.utc) + timedelta(days=2), ) booking.id = 3 with pytest.raises(ValueError) as exc: await new_booking(async_session, booking) assert "overlap" in str(exc.value) @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_new_booking_attendees_exceed_capacity( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test attendee count exceeding room capacity. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ from backend.models import Booking from backend.services.bookings import new_booking # Patch get_room to have capacity 1 with ( patch( "backend.services.bookings.get_room", new=AsyncMock(return_value=MagicMock(capacity=1)), ), patch( "backend.services.bookings.config", wraps=sys.modules["backend"].config ) as mock_config, ): mock_config.side_effect = _config_side_effect # Patch session.scalars to simulate no overlap mock_scalars_result = AsyncMock() mock_scalars_result.first = MagicMock(return_value=None) with ( patch.object( async_session, "scalars", AsyncMock(return_value=mock_scalars_result) ), patch( "backend.services.bookings.get_invitees_for_booking", new=AsyncMock(return_value=[MagicMock(), MagicMock()]), ), ): booking: Booking = Booking( room_id=1, start_time=datetime.now(timezone.utc) + timedelta(days=1), end_time=datetime.now(timezone.utc) + timedelta(days=2), ) booking.id = 4 with pytest.raises(ValueError) as exc: await new_booking(async_session, booking) assert "exceeds the room capacity" in str(exc.value) @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_new_booking_database_error( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test handling of database errors in new_booking. Verifies that new_booking rolls back the session on database failure. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ booking = Booking( room_id=1, start_time=datetime.now(timezone.utc) + timedelta(hours=1), end_time=datetime.now(timezone.utc) + timedelta(hours=2), ) booking.id = 1 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] # Patch get_room to always exist with ( patch( "backend.services.bookings.get_room", new=AsyncMock(return_value=MagicMock(capacity=10)), ), patch( "backend.services.bookings.config", wraps=sys.modules["backend"].config ) as mock_config, ): mock_config.side_effect = _config_side_effect # Patch session.scalars to simulate no overlap mock_scalars_result = AsyncMock() mock_scalars_result.first = MagicMock(return_value=None) with ( patch.object( async_session, "scalars", AsyncMock(return_value=mock_scalars_result) ), patch( "backend.services.bookings.get_invitees_for_booking", new=AsyncMock(return_value=[]), ), ): with pytest.raises(SQLAlchemyError): await new_booking(async_session, booking) async_session.add.assert_called_once_with(booking) async_session.commit.assert_called_once() async_session.rollback.assert_called_once() @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_update_booking_success( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test successful update of a booking. Verifies that update_booking updates the booking, commits the session, and returns the updated booking. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ booking_id = 1 update_params: dict[str, Any] = { "room_id": 2, "start_time": datetime.now(timezone.utc) + timedelta(hours=1), "end_time": datetime.now(timezone.utc) + timedelta(hours=2), "title": "Service Updated Booking", } 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_booking = Booking(**update_params) updated_booking.id = booking_id async_session.scalar = AsyncMock( # type: ignore [method-assign] return_value=updated_booking ) # Patch get_booking to return a booking with ( patch( "backend.services.bookings.get_booking", new=AsyncMock(return_value=updated_booking), ), patch( "backend.services.bookings._validate_no_overlap", new=AsyncMock(return_value=None), ), patch( "backend.services.bookings._validate_room_exists", new=AsyncMock(return_value=MagicMock(capacity=10)), ), patch( "backend.services.bookings.get_invitees_for_booking", new=AsyncMock(return_value=[]), ), ): result: Booking = await update_booking( async_session, booking_id, **update_params ) assert result == updated_booking async_session.execute.assert_called_once() assert async_session.execute.call_args.args[0].compare( update(Booking).where(Booking.id == booking_id).values(**update_params) ) async_session.commit.assert_called_once() # Removed async_session.scalar.assert_called_once() and related assertion @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_update_booking_not_found( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test handling of non-existent booking in update_booking. Verifies that update_booking raises ValueError when the booking is not found. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ booking_id = 999 update_params: dict[str, Any] = { "room_id": 2, "start_time": datetime.now(timezone.utc) + timedelta(hours=1), "end_time": datetime.now(timezone.utc) + timedelta(hours=2), "title": "Service Updated Booking", } 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] # Patch get_booking to return a booking with ( patch( "backend.services.bookings.get_booking", new=AsyncMock( return_value=MagicMock( id=booking_id, room_id=2, start_time=update_params["start_time"], end_time=update_params["end_time"], ) ), ), patch( "backend.services.bookings._validate_no_overlap", new=AsyncMock(return_value=None), ), patch( "backend.services.bookings._validate_room_exists", new=AsyncMock(return_value=MagicMock(capacity=10)), ), patch( "backend.services.bookings.get_invitees_for_booking", new=AsyncMock(return_value=[]), ), ): from sqlalchemy.exc import NoResultFound with pytest.raises(NoResultFound): await update_booking(async_session, booking_id, **update_params) async_session.execute.assert_called_once() # Compare SQL string representations for reliability actual_sql = str(async_session.execute.call_args.args[0]) expected_sql = str( update(Booking).where(Booking.id == booking_id).values(**update_params) ) assert actual_sql == expected_sql async_session.rollback.assert_called_once() @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_update_booking_database_error( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test handling of database errors in update_booking. Verifies that update_booking rolls back the session on database failure. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ booking_id = 1 update_params: dict[str, Any] = { "room_id": 2, "start_time": datetime.now(timezone.utc) + timedelta(hours=1), "end_time": datetime.now(timezone.utc) + timedelta(hours=2), "title": "Service Updated Booking", } async_session.execute = AsyncMock( # type: ignore [method-assign] side_effect=SQLAlchemyError("Database error") ) async_session.rollback = AsyncMock() # type: ignore [method-assign] with ( patch( "backend.services.bookings.get_booking", new=AsyncMock( return_value=MagicMock( id=booking_id, room_id=2, start_time=update_params["start_time"], end_time=update_params["end_time"], ) ), ), patch( "backend.services.bookings._validate_no_overlap", new=AsyncMock(return_value=None), ), patch( "backend.services.bookings._validate_room_exists", new=AsyncMock(return_value=MagicMock(capacity=10)), ), patch( "backend.services.bookings.get_invitees_for_booking", new=AsyncMock(return_value=[]), ), ): with pytest.raises(SQLAlchemyError): await update_booking(async_session, booking_id, **update_params) async_session.execute.assert_called_once() assert async_session.execute.call_args.args[0].compare( update(Booking).where(Booking.id == booking_id).values(**update_params) ) async_session.rollback.assert_called_once() @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_delete_booking_success( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test successful deletion of a booking. Verifies that delete_booking deletes the booking and commits the session. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ booking_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_booking(async_session, booking_id) async_session.execute.assert_called_once() assert async_session.execute.call_args.args[0].compare( delete(Booking).where(Booking.id == booking_id) ) async_session.commit.assert_called_once() @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_delete_booking_not_found( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test handling of non-existent booking in delete_booking. Verifies that delete_booking raises ValueError when the booking is not found. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ booking_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_booking(async_session, booking_id) async_session.execute.assert_called_once() assert async_session.execute.call_args.args[0].compare( delete(Booking).where(Booking.id == booking_id) ) async_session.rollback.assert_called_once() @pytest.mark.asyncio @pytest.mark.parametrize("mock_logger", ["backend.services.bookings"], indirect=True) async def test_delete_booking_database_error( async_session: AsyncSession, mock_logger: MagicMock ) -> None: """Test handling of database errors in delete_booking. Verifies that delete_booking rolls back the session on database failure. Args: async_session: The asynchronous database session. mock_logger: The mocked logger instance. """ booking_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_booking(async_session, booking_id) async_session.execute.assert_called_once() assert async_session.execute.call_args.args[0].compare( delete(Booking).where(Booking.id == booking_id) ) async_session.rollback.assert_called_once()