darkhelm/conference-room-booking-system
Public Access
1
0
Fork 0
mirror of https://github.com/xlorepdarkhelm/numinar-coding-project.git synced 2026-09-09 17:45:35 -04:00
Code Issues Packages Projects Releases Wiki Activity

Fixing stuff.

Browse Source 
Signed-off-by: Cliff Hill <xlorep@darkhelm.org>
This commit is contained in:
Xlorep DarkHelm
2025-08-27 12:16:24 -04:00
parent 9fb8a8642e
commit 1d29e0bbd1
5 changed files with 305 additions and 304 deletions
Download Patch File Download Diff File Expand all files Collapse all files

0
backend/.bandit → .bandit
View File

6
backend/src/backend/services/bookings.py
View File

@@ -275,9 +275,9 @@ async def update_booking(
)
try:
current = await get_booking(session, booking_id)
new_room_id = kwargs.get("room_id", current.room_id)
new_start = kwargs.get("start_time", current.start_time)
new_end = kwargs.get("end_time", current.end_time)
new_room_id = cast(int, kwargs.get("room_id", current.room_id))
new_start = cast(datetime, kwargs.get("start_time", current.start_time))
new_end = cast(datetime, kwargs.get("end_time", current.end_time))
room = await _validate_room_exists(session, new_room_id, get_room)
now = datetime.now(timezone.utc)

10
backend/src/backend/services/invitees.py
View File

@@ -37,7 +37,15 @@ async def get_invitees_for_booking(
try:
stmt = select(Invitee).where(Invitee.booking_id == booking_id)
result = await session.scalars(stmt)
invitees = cast(InviteeList, result.all())
invitees: InviteeList
all_result = result.all()
if hasattr(all_result, "__await__"):
invitees = cast(
InviteeList,
await all_result, # pyright: ignore[reportGeneralTypeIssues]
)
else:
invitees = cast(InviteeList, all_result)
logger.info(
f"Successfully retrieved {len(invitees)} invitees for booking_id: {booking_id}"
)

459
backend/tests/services/test_bookings.py
View File

@@ -47,17 +47,15 @@ async def test_get_bookings_for_room_success(
mock_scalars_result = AsyncMock()
mock_scalars_result.all = MagicMock(return_value=sample_bookings)
mock_scalars = AsyncMock(return_value=mock_scalars_result)
async_session.scalars = mock_scalars # type: ignore [method-assign]
result: BookingList = await get_bookings_for_room(async_session, room_id)
assert isinstance(result, list)
assert len(result) == 2
assert result == sample_bookings
async_session.scalars.assert_called_once()
assert async_session.scalars.call_args.args[0].compare(
select(Booking).where(Booking.room_id == room_id)
)
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
@@ -77,16 +75,14 @@ async def test_get_bookings_for_room_empty(
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: BookingList = await get_bookings_for_room(async_session, room_id)
assert isinstance(result, list)
assert len(result) == 0
async_session.scalars.assert_called_once()
assert async_session.scalars.call_args.args[0].compare(
select(Booking).where(Booking.room_id == room_id)
)
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
@@ -103,13 +99,14 @@ async def test_get_bookings_for_room_database_error(
mock_logger: The mocked logger instance.
"""
room_id = 1
async_session.scalars = AsyncMock( # type: ignore [method-assign]
side_effect=SQLAlchemyError("Database error")
)
with pytest.raises(SQLAlchemyError):
await get_bookings_for_room(async_session, room_id)
async_session.scalars.assert_called_once()
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
@@ -234,23 +231,24 @@ async def test_new_booking_success(
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 config to allow future
with patch(
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 = lambda key, **kwargs: ( # type: ignore
12
if key == "BOOKING_MAX_MONTHS"
else kwargs.get("default") # type: ignore
)
# Patch session.scalars to simulate no overlap
mock_scalars_result = AsyncMock()
mock_scalars_result.first = MagicMock(return_value=None)
async_session.scalars = AsyncMock(return_value=mock_scalars_result)
) as mock_config,
):
mock_config.side_effect = lambda key, **kwargs: (
12 if key == "BOOKING_MAX_MONTHS" else kwargs.get("default")
)
# 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",
@@ -339,28 +337,32 @@ async def test_new_booking_constraints(
if room_exists
else AsyncMock(side_effect=Exception("no room"))
)
with patch("backend.services.bookings.get_room", new=get_room_patch):
# Patch config to set max months
with patch(
with (
patch("backend.services.bookings.get_room", new=get_room_patch),
patch(
"backend.services.bookings.config", wraps=sys.modules["backend"].config
) as mock_config:
mock_config.side_effect = lambda key, *args, **kwargs: ( # type: ignore
max_months
if key == "BOOKING_MAX_MONTHS"
else kwargs.get("default") # type: ignore
)
# Patch session.scalars to simulate no overlap
mock_scalars_result = AsyncMock()
mock_scalars_result.first = MagicMock(return_value=None)
async_session.scalars = AsyncMock(return_value=mock_scalars_result)
# Patch get_invitees_for_booking to return 0 invitees
with patch(
) as mock_config,
):
mock_config.side_effect = lambda key, *args, **kwargs: (
max_months if key == "BOOKING_MAX_MONTHS" else kwargs.get("default")
)
# 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)
),
):
with pytest.raises(ValueError) as exc:
await new_booking(async_session, booking)
assert expected_error in str(exc.value)
@pytest.mark.asyncio
@@ -378,29 +380,30 @@ async def test_new_booking_overlap(
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 config to allow future
with patch(
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 = lambda key, **kwargs: ( # type: ignore
12
if key == "BOOKING_MAX_MONTHS"
else kwargs.get("default") # type: ignore
)
# 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)
async_session.scalars = AsyncMock(return_value=mock_scalars_result)
) as mock_config,
):
mock_config.side_effect = lambda key, **kwargs: (
12 if key == "BOOKING_MAX_MONTHS" else kwargs.get("default")
)
# 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),
@@ -427,37 +430,39 @@ async def test_new_booking_attendees_exceed_capacity(
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 config to allow future
with patch(
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 = lambda key, **kwargs: ( # type: ignore
12
if key == "BOOKING_MAX_MONTHS"
else kwargs.get("default") # type: ignore
)
# Patch session.scalars to simulate no overlap
mock_scalars_result = AsyncMock()
mock_scalars_result.first = MagicMock(return_value=None)
async_session.scalars = AsyncMock(return_value=mock_scalars_result)
# Patch get_invitees_for_booking to return 2 invitees (over capacity)
with patch(
) as mock_config,
):
mock_config.side_effect = lambda key, **kwargs: (
12 if key == "BOOKING_MAX_MONTHS" else kwargs.get("default")
)
# 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)
),
):
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
@@ -486,33 +491,35 @@ async def test_new_booking_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 config to allow future
with patch(
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 = lambda key, **kwargs: ( # type: ignore
12
if key == "BOOKING_MAX_MONTHS"
else kwargs.get("default") # type: ignore
)
# Patch session.scalars to simulate no overlap
mock_scalars_result = AsyncMock()
mock_scalars_result.first = MagicMock(return_value=None)
async_session.scalars = AsyncMock(return_value=mock_scalars_result)
# Patch get_invitees_for_booking to return 0 invitees
with patch(
) as mock_config,
):
mock_config.side_effect = lambda key, **kwargs: (
12 if key == "BOOKING_MAX_MONTHS" else kwargs.get("default")
)
# 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()
),
):
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
@@ -547,37 +554,34 @@ async def test_update_booking_success(
)
# Patch get_booking to return a booking
with patch(
"backend.services.bookings.get_booking",
new=AsyncMock(return_value=updated_booking),
):
# Patch _validate_no_overlap to do nothing
with patch(
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 _validate_room_exists to return a room with capacity
with patch(
"backend.services.bookings._validate_room_exists",
new=AsyncMock(return_value=MagicMock(capacity=10)),
):
# 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 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
),
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
@@ -606,41 +610,38 @@ async def test_update_booking_not_found(
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"],
)
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 _validate_no_overlap to do nothing
with patch(
patch(
"backend.services.bookings._validate_no_overlap",
new=AsyncMock(return_value=None),
):
# Patch _validate_room_exists to return a room with capacity
with patch(
"backend.services.bookings._validate_room_exists",
new=AsyncMock(return_value=MagicMock(capacity=10)),
):
# Patch get_invitees_for_booking to return 0 invitees
with patch(
"backend.services.bookings.get_invitees_for_booking",
new=AsyncMock(return_value=[]),
):
with pytest.raises(ValueError):
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()
),
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(ValueError):
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
@@ -667,42 +668,38 @@ async def test_update_booking_database_error(
)
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"],
)
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 _validate_no_overlap to do nothing
with patch(
patch(
"backend.services.bookings._validate_no_overlap",
new=AsyncMock(return_value=None),
):
# Patch _validate_room_exists to return a room with capacity
with patch(
"backend.services.bookings._validate_room_exists",
new=AsyncMock(return_value=MagicMock(capacity=10)),
):
# Patch get_invitees_for_booking to return 0 invitees
with 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()
),
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

134
backend/tests/services/test_invitees.py
View File

@@ -36,15 +36,13 @@ async def test_get_invitees_for_booking_success(
mock_scalars_result = AsyncMock()
mock_scalars_result.all = MagicMock(return_value=sample_invitees)
mock_scalars = AsyncMock(return_value=mock_scalars_result)
async_session.scalars = mock_scalars # type: ignore [method-assign]
result: InviteeList = await get_invitees_for_booking(async_session, booking_id)
with patch.object(async_session, "scalars", mock_scalars):
result = await get_invitees_for_booking(async_session, booking_id)
assert isinstance(result, list)
assert len(result) == 2
assert result == sample_invitees
async_session.scalars.assert_called_once()
assert async_session.scalars.call_args.args[0].compare(
mock_scalars.assert_called_once()
assert mock_scalars.call_args.args[0].compare(
select(Invitee).where(Invitee.booking_id == booking_id)
)
@@ -64,16 +62,14 @@ async def test_get_invitees_for_booking_empty(
"""
booking_id = 1
mock_scalars_result = AsyncMock()
mock_scalars_result.all = MagicMock(return_value=[])
mock_scalars_result.all = AsyncMock(return_value=[])
mock_scalars = AsyncMock(return_value=mock_scalars_result)
async_session.scalars = mock_scalars # type: ignore [method-assign]
result: InviteeList = await get_invitees_for_booking(async_session, booking_id)
with patch.object(async_session, "scalars", mock_scalars):
result = await get_invitees_for_booking(async_session, booking_id)
assert isinstance(result, list)
assert len(result) == 0
async_session.scalars.assert_called_once()
assert async_session.scalars.call_args.args[0].compare(
mock_scalars.assert_called_once()
assert mock_scalars.call_args.args[0].compare(
select(Invitee).where(Invitee.booking_id == booking_id)
)
@@ -92,13 +88,11 @@ async def test_get_invitees_for_booking_database_error(
mock_logger: The mocked logger instance.
"""
booking_id = 1
async_session.scalars = AsyncMock( # type: ignore [method-assign]
side_effect=SQLAlchemyError("Database error")
)
with pytest.raises(SQLAlchemyError):
await get_invitees_for_booking(async_session, booking_id)
async_session.scalars.assert_called_once()
mock_scalars = AsyncMock(side_effect=SQLAlchemyError("Database error"))
with patch.object(async_session, "scalars", mock_scalars):
with pytest.raises(SQLAlchemyError):
await get_invitees_for_booking(async_session, booking_id)
mock_scalars.assert_called_once()
@pytest.mark.asyncio
@@ -118,35 +112,35 @@ async def test_add_invitee_to_booking_success(
sample_invitee: The mocked invitee object.
mock_logger: The mocked logger instance.
"""
async_session.add = MagicMock() # type: ignore [method-assign]
async_session.commit = AsyncMock() # type: ignore [method-assign]
# Patch session.scalars().first() to return None (no existing invitee)
mock_scalars_result = AsyncMock()
mock_scalars_result.first = MagicMock(return_value=None)
mock_scalars_result.all = MagicMock(return_value=[])
async_session.scalars = AsyncMock(return_value=mock_scalars_result)
# Patch get_booking and get_room to return objects with correct attributes
with patch(
"backend.services.bookings.get_booking",
new=AsyncMock(return_value=MagicMock(room_id=1)),
):
with patch(
with (
patch.object(async_session, "add", new_callable=MagicMock) as mock_add,
patch.object(async_session, "commit", new_callable=AsyncMock) as mock_commit,
patch.object(
async_session, "scalars", AsyncMock(return_value=mock_scalars_result)
),
patch(
"backend.services.bookings.get_booking",
new=AsyncMock(return_value=MagicMock(room_id=1)),
),
patch(
"backend.services.rooms.get_room",
new=AsyncMock(return_value=MagicMock(capacity=10)),
):
with patch(
"backend.services.invitees.get_invitees_for_booking",
new=AsyncMock(return_value=[]),
):
result: Invitee = await add_invitee_to_booking(
async_session, sample_invitee.booking_id, sample_invitee.user_email
)
assert result.booking_id == sample_invitee.booking_id
assert result.user_email == sample_invitee.user_email
async_session.add.assert_called_once()
async_session.commit.assert_called_once()
),
patch(
"backend.services.invitees.get_invitees_for_booking",
new=AsyncMock(return_value=[]),
),
):
result = await add_invitee_to_booking(
async_session, sample_invitee.booking_id, sample_invitee.user_email
)
assert result.booking_id == sample_invitee.booking_id
assert result.user_email == sample_invitee.user_email
mock_add.assert_called_once()
mock_commit.assert_called_once()
@pytest.mark.asyncio
@@ -165,36 +159,38 @@ async def test_add_invitee_to_booking_database_error(
booking_id = 1
user_email = "user1@example.com"
invitee = Invitee(booking_id=booking_id, user_email=user_email)
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 session.scalars().first() to return None (no existing invitee)
mock_scalars_result = AsyncMock()
mock_scalars_result.first = MagicMock(return_value=None)
mock_scalars_result.all = MagicMock(return_value=[])
async_session.scalars = AsyncMock(return_value=mock_scalars_result)
# Patch get_booking and get_room to return objects with correct attributes
with patch(
"backend.services.bookings.get_booking",
new=AsyncMock(return_value=MagicMock(room_id=1)),
):
with patch(
with (
patch.object(async_session, "add", MagicMock()) as mock_add,
patch.object(
async_session,
"commit",
AsyncMock(side_effect=SQLAlchemyError("Database error")),
) as mock_commit,
patch.object(async_session, "rollback", AsyncMock()) as mock_rollback,
patch.object(
async_session, "scalars", AsyncMock(return_value=mock_scalars_result)
),
patch(
"backend.services.bookings.get_booking",
new=AsyncMock(return_value=MagicMock(room_id=1)),
),
patch(
"backend.services.rooms.get_room",
new=AsyncMock(return_value=MagicMock(capacity=10)),
):
with patch(
"backend.services.invitees.get_invitees_for_booking",
new=AsyncMock(return_value=[]),
):
with pytest.raises(SQLAlchemyError):
await add_invitee_to_booking(async_session, booking_id, user_email)
async_session.add.assert_called_once_with(invitee)
async_session.commit.assert_called_once()
async_session.rollback.assert_called_once()
),
patch(
"backend.services.invitees.get_invitees_for_booking",
new=AsyncMock(return_value=[]),
),
):
with pytest.raises(SQLAlchemyError):
await add_invitee_to_booking(async_session, booking_id, user_email)
mock_add.assert_called_once_with(invitee)
mock_commit.assert_called_once()
mock_rollback.assert_called_once()
@pytest.mark.asyncio