Files
conference-room-booking-system/backend/tests/routers/test_rooms.py
2025-10-02 22:26:30 -04:00

765 lines
25 KiB
Python

"""Unit tests for the backend.routers.rooms module."""
import asyncio
import re
from typing import Any
import asyncpg
import pytest
from httpx import AsyncClient
from httpx import Response
from sqlalchemy.exc import NoResultFound
from sqlalchemy.exc import SQLAlchemyError
from tests.types import RoomData
from tests.types import RoomUpdateData
from backend.dependencies.rooms import delete_room_service
from backend.dependencies.rooms import get_room_service
from backend.dependencies.rooms import get_rooms_service
from backend.dependencies.rooms import new_room_service
from backend.dependencies.rooms import update_room_service
from backend.main import app
from backend.models import Room
from backend.models import RoomList
from backend.routers.rooms import publish_room_availability_event
from backend.routers.rooms import room_availability_subscribers
# Useful constants for rooms router tests
ROOM_SUCCESS_STATUS = 200
ROOM_CREATED_STATUS = 201
ROOM_DELETED_STATUS = 204
ROOM_NOT_FOUND_STATUS = 404
BAD_REQUEST_STATUS = 400
INTERNAL_SERVER_ERROR_STATUS = 500
ROOM_DB_ERROR_DETAIL = "Database error"
ROOM_NOT_FOUND_DETAIL = "Room not found"
ROOM_UNEXPECTED_ERROR_DETAIL = "An unexpected error occurred"
ROOM_BAD_REQUEST_DETAIL = "bad request"
ROOM_NOT_FOUND_ID = 999
@pytest.mark.asyncio
@pytest.mark.parametrize(
"room_case", ["non_empty", "empty"], ids=["non_empty", "empty"]
)
async def test_get_rooms_parametrized_fixture(
client: AsyncClient,
sample_rooms: RoomList,
room_case: str,
) -> None:
"""Parametrized test for GET /rooms/ using fixture data only.
Args:
client: The FastAPI test client for making HTTP requests.
sample_rooms: The mocked list of rooms.
room_case: Indicates non-empty or empty test case.
Asserts:
- Response status code is ROOM_SUCCESS_STATUS.
- Returned rooms match expected fields (id, name, location, equipment, capacity).
"""
if room_case == "non_empty":
rooms = sample_rooms
else:
rooms = []
async def mock_get_rooms(*args: Any, **kwargs: Any) -> RoomList:
return rooms
app.dependency_overrides[get_rooms_service] = lambda: mock_get_rooms
response = await client.get("/rooms/")
assert response.status_code == ROOM_SUCCESS_STATUS
for returned_room, expected_room in zip(response.json(), rooms, strict=True):
assert returned_room["id"] == expected_room.id
assert returned_room["name"] == expected_room.name
assert returned_room["location"] == expected_room.location
assert returned_room["equipment"] == expected_room.equipment
assert returned_room["capacity"] == expected_room.capacity
@pytest.mark.asyncio
async def test_new_room_success(
client: AsyncClient,
sample_room: Room,
sample_room_data: RoomData,
) -> None:
"""Test POST /rooms/ for successful creation.
Args:
client: The FastAPI test client for making HTTP requests.
sample_room: The mocked Room object.
sample_room_data: The mocked room data dictionary.
Asserts:
- Response status code is ROOM_CREATED_STATUS.
- Response JSON matches sample_room_data.
"""
async def mock_new_room(*args: Any, **kwargs: Any) -> Room:
return sample_room
app.dependency_overrides[new_room_service] = lambda: mock_new_room
response = await client.post("/rooms/", json=sample_room_data)
assert response.status_code == ROOM_CREATED_STATUS
assert response.json() == sample_room_data
@pytest.mark.asyncio
async def test_update_room_success(
client: AsyncClient,
sample_rooms: RoomList,
room_update_data: RoomData,
updated_room: Room,
) -> None:
"""Test PATCH /rooms/{room_id} for successful update.
Args:
client: The FastAPI test client for making HTTP requests.
sample_rooms: The mocked list of rooms.
room_update_data: The mocked room update data dictionary.
updated_room: The mocked updated Room object.
Asserts:
- Response status code is ROOM_SUCCESS_STATUS.
- Response JSON matches updated_room fields (id, name, location, equipment, capacity).
"""
room_id = sample_rooms[0].id
async def mock_get_rooms(*args: Any, **kwargs: Any) -> RoomList:
return sample_rooms
async def mock_update_room(*args: Any, **kwargs: Any) -> Room:
return updated_room
app.dependency_overrides[get_rooms_service] = lambda: mock_get_rooms
app.dependency_overrides[update_room_service] = lambda: mock_update_room
response = await client.patch(f"/rooms/{room_id}", json=room_update_data)
assert response.status_code == ROOM_SUCCESS_STATUS
actual = response.json()
# Remove 'bookings' if present for comparison
actual.pop("bookings", None)
# Compare all relevant fields
assert actual["id"] == updated_room.id
assert actual["name"] == updated_room.name
assert actual["location"] == updated_room.location
assert actual["equipment"] == updated_room.equipment
assert actual["capacity"] == updated_room.capacity
@pytest.mark.asyncio
async def test_delete_room_success(
client: AsyncClient,
) -> None:
"""Test DELETE /rooms/{room_id} for successful deletion.
Args:
client: The FastAPI test client for making HTTP requests.
Asserts:
- Response status code is ROOM_DELETED_STATUS.
- Response text is empty.
"""
room_id = 1
async def mock_delete_room(*args: Any, **kwargs: Any) -> None:
return None
app.dependency_overrides[delete_room_service] = lambda: mock_delete_room
response = await client.delete(f"/rooms/{room_id}")
assert response.status_code == ROOM_DELETED_STATUS
assert response.text == ""
@pytest.mark.asyncio
async def test_delete_room_not_found(
client: AsyncClient,
) -> None:
"""Test DELETE /rooms/{room_id} for not found case.
Args:
client: The FastAPI test client for making HTTP requests.
Asserts:
- Response status code is ROOM_NOT_FOUND_STATUS.
- Response JSON contains 'detail'.
"""
room_id = ROOM_NOT_FOUND_ID
async def mock_delete_room(*args: Any, **kwargs: Any) -> None:
raise ValueError()
app.dependency_overrides[delete_room_service] = lambda: mock_delete_room
response = await client.delete(f"/rooms/{room_id}")
assert response.status_code == ROOM_NOT_FOUND_STATUS
assert "detail" in response.json()
@pytest.mark.asyncio
async def test_stream_room_availability(client: AsyncClient) -> None:
"""Test GET /rooms/availability/stream SSE endpoint for connection and response format.
Args:
client: The FastAPI test client for making HTTP requests.
Asserts:
- Response status code is ROOM_SUCCESS_STATUS.
- Response content-type is 'text/event-stream'.
- Response body contains keep-alive or SSE data.
"""
response = await client.get("/rooms/availability/stream?test_mode=true")
assert response.status_code == ROOM_SUCCESS_STATUS
assert response.headers["content-type"].startswith("text/event-stream")
# Check for keep-alive or SSE data in the response
body = response.text
assert re.search(r"(: keep-alive|data:)", body)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"exception,expected_status,expected_detail",
[
(
ValueError(ROOM_BAD_REQUEST_DETAIL),
BAD_REQUEST_STATUS,
ROOM_BAD_REQUEST_DETAIL,
),
(NoResultFound(), ROOM_NOT_FOUND_STATUS, ROOM_NOT_FOUND_DETAIL),
(SQLAlchemyError(), INTERNAL_SERVER_ERROR_STATUS, ROOM_DB_ERROR_DETAIL),
(Exception("fail"), INTERNAL_SERVER_ERROR_STATUS, ROOM_UNEXPECTED_ERROR_DETAIL),
],
ids=["bad_request", "not_found", "db_error", "unexpected_error"],
)
async def test_update_room_error_cases_post(
client: AsyncClient,
sample_rooms: RoomList,
room_update_data: RoomUpdateData,
exception: Exception,
expected_status: int,
expected_detail: str,
) -> None:
"""Test PATCH /rooms/{room_id} error scenarios.
Args:
client: The FastAPI test client for making HTTP requests.
sample_rooms: The mocked list of rooms.
room_update_data: The mocked room update data dictionary.
exception: The exception to raise in the dependency.
expected_status: The expected HTTP status code.
expected_detail: The expected error detail message.
Asserts:
- Response status code matches expected_status.
- Response JSON 'detail' contains expected_detail.
"""
room_id = sample_rooms[0].id
async def mock_update_room(*args: Any, **kwargs: Any) -> Room:
raise exception
app.dependency_overrides[update_room_service] = lambda: mock_update_room
response = await client.patch(f"/rooms/{room_id}", json=room_update_data)
assert response.status_code == expected_status
assert expected_detail.lower() in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"exception,expected_status,expected_detail",
[
(NoResultFound(), ROOM_NOT_FOUND_STATUS, ROOM_NOT_FOUND_DETAIL),
(SQLAlchemyError(), INTERNAL_SERVER_ERROR_STATUS, ROOM_DB_ERROR_DETAIL),
(Exception("fail"), INTERNAL_SERVER_ERROR_STATUS, ROOM_UNEXPECTED_ERROR_DETAIL),
],
ids=["not_found", "db_error", "unexpected_error"],
)
async def test_update_room_error_cases(
client: AsyncClient,
sample_rooms: RoomList,
room_update_data: RoomData,
exception: Exception,
expected_status: int,
expected_detail: str,
) -> None:
"""Test POST /rooms/ error scenarios.
Args:
client: The FastAPI test client for making HTTP requests.
sample_rooms: The mocked list of rooms.
room_update_data: The mocked room update data dictionary.
exception: The exception to raise in the dependency.
expected_status: The expected HTTP status code.
expected_detail: The expected error detail message.
Asserts:
- Response status code matches expected_status.
- Response JSON 'detail' contains expected_detail.
"""
room_id = sample_rooms[0].id
async def mock_update_room(*args: Any, **kwargs: Any) -> Room:
raise exception
app.dependency_overrides[update_room_service] = lambda: mock_update_room
response = await client.patch(f"/rooms/{room_id}", json=room_update_data)
assert response.status_code == expected_status
assert expected_detail.lower() in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"exception,expected_status,expected_detail",
[
(
ValueError(ROOM_NOT_FOUND_DETAIL),
ROOM_NOT_FOUND_STATUS,
ROOM_NOT_FOUND_DETAIL,
),
(NoResultFound(), ROOM_NOT_FOUND_STATUS, ROOM_NOT_FOUND_DETAIL),
(SQLAlchemyError(), INTERNAL_SERVER_ERROR_STATUS, ROOM_DB_ERROR_DETAIL),
(Exception("fail"), INTERNAL_SERVER_ERROR_STATUS, ROOM_UNEXPECTED_ERROR_DETAIL),
],
ids=["value_error", "not_found", "db_error", "unexpected_error"],
)
async def test_delete_room_error_cases(
client: AsyncClient,
exception: Exception,
expected_status: int,
expected_detail: str,
) -> None:
"""Test DELETE /rooms/{room_id} error scenarios.
Args:
client: The FastAPI test client for making HTTP requests.
exception: The exception to raise in the dependency.
expected_status: The expected HTTP status code.
expected_detail: The expected error detail message.
Asserts:
- Response status code matches expected_status.
- Response JSON 'detail' contains expected_detail.
"""
room_id = ROOM_NOT_FOUND_ID
async def mock_delete_room(*args: Any, **kwargs: Any) -> None:
raise exception
app.dependency_overrides[delete_room_service] = lambda: mock_delete_room
response = await client.delete(f"/rooms/{room_id}")
assert response.status_code == expected_status
assert expected_detail.lower() in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"exception,expected_status,expected_detail",
[
(
asyncpg.InterfaceError("iface error"),
INTERNAL_SERVER_ERROR_STATUS,
ROOM_DB_ERROR_DETAIL,
),
],
ids=["iface_error"],
)
async def test_update_room_asyncpg_error(
client: AsyncClient,
sample_rooms: RoomList,
room_update_data: RoomUpdateData,
exception: Exception,
expected_status: int,
expected_detail: str,
) -> None:
"""Test PATCH /rooms/{room_id} asyncpg.InterfaceError scenario.
Args:
client: The FastAPI test client for making HTTP requests.
sample_rooms: The list of sample Room objects.
room_update_data: The update data for the room.
exception: The exception to raise in the dependency override.
expected_status: The expected HTTP status code.
expected_detail: The expected error detail message.
Asserts:
- Response status code matches expected_status.
- Response detail matches expected_detail.
"""
room_id = sample_rooms[0].id
async def mock_update_room(*args: Any, **kwargs: Any) -> Room:
raise exception
app.dependency_overrides[update_room_service] = lambda: mock_update_room
response = await client.patch(f"/rooms/{room_id}", json=room_update_data)
assert response.status_code == expected_status
assert expected_detail.lower() in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"exception,expected_status,expected_detail",
[
(
asyncpg.InterfaceError("iface error"),
INTERNAL_SERVER_ERROR_STATUS,
ROOM_DB_ERROR_DETAIL,
),
],
ids=["iface_error"],
)
async def test_create_room_asyncpg_error(
client: AsyncClient,
sample_room_data: RoomData,
exception: Exception,
expected_status: int,
expected_detail: str,
) -> None:
"""Test POST /rooms/ asyncpg.InterfaceError scenario.
Args:
client: The FastAPI test client for making HTTP requests.
sample_room_data: The room data to post.
exception: The exception to raise in the dependency override.
expected_status: The expected HTTP status code.
expected_detail: The expected error detail message.
Asserts:
- Response status code matches expected_status.
- Response detail matches expected_detail.
"""
async def mock_new_room(*args: Any, **kwargs: Any) -> Room:
raise exception
app.dependency_overrides[new_room_service] = lambda: mock_new_room
response = await client.post("/rooms/", json=sample_room_data)
assert response.status_code == expected_status
assert expected_detail.lower() in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"exception,expected_status,expected_detail",
[
(
asyncpg.InterfaceError("iface error"),
INTERNAL_SERVER_ERROR_STATUS,
ROOM_DB_ERROR_DETAIL,
),
],
ids=["iface_error"],
)
async def test_delete_room_asyncpg_error(
client: AsyncClient,
exception: Exception,
expected_status: int,
expected_detail: str,
) -> None:
"""Test DELETE /rooms/{room_id} asyncpg.InterfaceError scenario.
Args:
client: The FastAPI test client for making HTTP requests.
exception: The exception to raise in the dependency override.
expected_status: The expected HTTP status code.
expected_detail: The expected error detail message.
Asserts:
- Response status code matches expected_status.
- Response detail matches expected_detail.
"""
room_id = ROOM_NOT_FOUND_ID
async def mock_delete_room(*args: Any, **kwargs: Any) -> None:
raise exception
app.dependency_overrides[delete_room_service] = lambda: mock_delete_room
response = await client.delete(f"/rooms/{room_id}")
assert response.status_code == expected_status
assert expected_detail.lower() in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"exception,expected_status,expected_detail",
[
(
asyncpg.InterfaceError("iface error"),
INTERNAL_SERVER_ERROR_STATUS,
ROOM_DB_ERROR_DETAIL,
),
],
)
async def test_get_rooms_asyncpg_error(
client: AsyncClient,
exception: Exception,
expected_status: int,
expected_detail: str,
) -> None:
"""Test GET /rooms/ asyncpg.InterfaceError scenario.
Args:
client: The FastAPI test client for making HTTP requests.
exception: The exception to raise in the dependency override.
expected_status: The expected HTTP status code.
expected_detail: The expected error detail message.
Asserts:
- Response status code matches expected_status.
- Response detail matches expected_detail.
"""
async def mock_get_rooms(*args: Any, **kwargs: Any) -> RoomList:
raise exception
app.dependency_overrides[get_rooms_service] = lambda: mock_get_rooms
response = await client.get("/rooms/")
assert response.status_code == expected_status
assert expected_detail.lower() in response.json()["detail"].lower()
def test_publish_room_availability_event_no_subscribers() -> None:
"""Test publish_room_availability_event with no subscribers.
Asserts:
- No error is raised when there are no subscribers.
"""
room_availability_subscribers.clear()
asyncio.run(publish_room_availability_event({"test": True}))
@pytest.mark.asyncio
async def test_patch_room_update_data_pop_id(
client: AsyncClient,
sample_rooms: RoomList,
room_update_data: RoomData,
updated_room: Room,
) -> None:
"""Test PATCH /rooms/{room_id} with 'id' in update_data covers pop logic.
Args:
client: The FastAPI test client for making HTTP requests.
sample_rooms: The mocked list of rooms.
room_update_data: The mocked room update data dictionary.
updated_room: The mocked updated Room object.
Asserts:
- Response status code is ROOM_SUCCESS_STATUS.
- Response JSON 'id' matches updated_room.id.
"""
room_id = sample_rooms[0].id
update_data = dict(room_update_data)
update_data["id"] = 1234
async def mock_update_room(*args: Any, **kwargs: Any) -> Room:
return updated_room
app.dependency_overrides[update_room_service] = lambda: mock_update_room
response = await client.patch(f"/rooms/{room_id}", json=update_data)
assert response.status_code == ROOM_SUCCESS_STATUS
assert response.json()["id"] == updated_room.id
@pytest.mark.asyncio
@pytest.mark.parametrize(
"mock_case, expected_status, expected_detail, exception",
[
("success", ROOM_SUCCESS_STATUS, None, None),
("not_found", ROOM_NOT_FOUND_STATUS, ROOM_NOT_FOUND_DETAIL, NoResultFound()),
(
"db_error",
INTERNAL_SERVER_ERROR_STATUS,
ROOM_DB_ERROR_DETAIL,
SQLAlchemyError(),
),
(
"iface_error",
INTERNAL_SERVER_ERROR_STATUS,
ROOM_DB_ERROR_DETAIL,
asyncpg.InterfaceError("iface error"),
),
(
"unexpected",
INTERNAL_SERVER_ERROR_STATUS,
ROOM_UNEXPECTED_ERROR_DETAIL,
Exception("fail"),
),
],
ids=[
"success",
"not_found",
"db_error",
"iface_error",
"unexpected",
],
)
async def test_read_room_parametrized(
client: AsyncClient,
sample_room: Room,
mock_case: str,
expected_status: int,
expected_detail: str | None,
exception: Exception | None,
) -> None:
"""Test GET /rooms/{room_id} endpoint for all branches.
Args:
client: The FastAPI test client for making HTTP requests.
sample_room: The mocked Room object for success case.
mock_case: The scenario to test (success, not_found, db_error, iface_error, unexpected).
expected_status: The expected HTTP status code.
expected_detail: The expected error detail message, if any.
exception: The exception to raise in the dependency for error cases.
Asserts:
- Response status code matches expected_status.
- On success, response JSON matches sample_room fields.
- On error, response JSON 'detail' matches expected_detail.
"""
room_id = sample_room.id
def mock_get_rooms(*args: Any, **kwargs: Any) -> list[Room]:
return [sample_room]
def mock_update_room(*args: Any, **kwargs: Any) -> Room:
return sample_room
def mock_new_room(*args: Any, **kwargs: Any) -> Room:
return sample_room
def mock_delete_room(*args: Any, **kwargs: Any) -> None:
return None
async def mock_get_room(*args: Any, **kwargs: Any) -> Room:
if mock_case == "success":
return sample_room
if exception is not None:
raise exception
return sample_room
app.dependency_overrides[get_rooms_service] = lambda: mock_get_rooms
app.dependency_overrides[update_room_service] = lambda: mock_update_room
app.dependency_overrides[new_room_service] = lambda: mock_new_room
app.dependency_overrides[delete_room_service] = lambda: mock_delete_room
app.dependency_overrides[get_room_service] = lambda: mock_get_room
response = await client.get(f"/rooms/{room_id}")
_assert_room_response(response, expected_status, sample_room, expected_detail)
def _assert_room_response(
response: Response,
expected_status: int,
sample_room: Room,
expected_detail: str | None,
) -> None:
"""Helper to assert room response for parametrized test.
Args:
response: The HTTP response to check.
expected_status: The expected HTTP status code.
sample_room: The expected Room object for success case.
expected_detail: The expected error detail message, if any.
Asserts:
- Response status code matches expected_status.
- On success, response JSON matches sample_room fields.
- On error, response JSON 'detail' contains expected_detail.
"""
assert response.status_code == expected_status
if expected_status == ROOM_SUCCESS_STATUS:
data = response.json()
assert data["id"] == sample_room.id
assert data["name"] == sample_room.name
assert data["location"] == sample_room.location
assert data["equipment"] == sample_room.equipment
assert data["capacity"] == sample_room.capacity
if expected_detail:
assert expected_detail.lower() in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"exception,expected_status,expected_detail",
[
(NoResultFound(), ROOM_NOT_FOUND_STATUS, ROOM_NOT_FOUND_DETAIL),
(SQLAlchemyError(), INTERNAL_SERVER_ERROR_STATUS, ROOM_DB_ERROR_DETAIL),
(
asyncpg.InterfaceError("iface error"),
INTERNAL_SERVER_ERROR_STATUS,
ROOM_DB_ERROR_DETAIL,
),
(Exception("fail"), INTERNAL_SERVER_ERROR_STATUS, ROOM_UNEXPECTED_ERROR_DETAIL),
],
)
async def test_get_rooms_error_branches_full(
client: AsyncClient,
exception: Exception,
expected_status: int,
expected_detail: str,
) -> None:
"""Test GET /rooms/ error branches for full coverage.
Args:
client: The FastAPI test client for making HTTP requests.
exception: The exception to raise in the dependency override.
expected_status: The expected HTTP status code.
expected_detail: The expected error detail message.
Asserts:
- Response status code matches expected_status.
- Response detail matches expected_detail.
"""
async def mock_get_rooms(*args: Any, **kwargs: Any) -> RoomList:
raise exception
app.dependency_overrides[get_rooms_service] = lambda: mock_get_rooms
response = await client.get("/rooms/")
assert response.status_code == expected_status
assert expected_detail.lower() in response.json()["detail"].lower()
@pytest.mark.asyncio
@pytest.mark.parametrize(
"exception,expected_status,expected_detail",
[
(NoResultFound(), ROOM_NOT_FOUND_STATUS, ROOM_NOT_FOUND_DETAIL),
(SQLAlchemyError(), INTERNAL_SERVER_ERROR_STATUS, ROOM_DB_ERROR_DETAIL),
(
asyncpg.InterfaceError("iface error"),
INTERNAL_SERVER_ERROR_STATUS,
ROOM_DB_ERROR_DETAIL,
),
(Exception("fail"), BAD_REQUEST_STATUS, ROOM_UNEXPECTED_ERROR_DETAIL),
],
)
async def test_create_room_error_branches_full(
client: AsyncClient,
sample_room_data: RoomData,
exception: Exception,
expected_status: int,
expected_detail: str,
) -> None:
"""Test POST /rooms/ error branches for full coverage.
Args:
client: The FastAPI test client for making HTTP requests.
sample_room_data: The room data to post.
exception: The exception to raise in the dependency override.
expected_status: The expected HTTP status code.
expected_detail: The expected error detail message.
Asserts:
- Response status code matches expected_status.
- Response detail matches expected_detail.
"""
async def mock_new_room(*args: Any, **kwargs: Any) -> Room:
raise exception
app.dependency_overrides[new_room_service] = lambda: mock_new_room
response = await client.post("/rooms/", json=sample_room_data)
assert response.status_code == expected_status
assert expected_detail.lower() in response.json()["detail"].lower()