mirror of
https://github.com/xlorepdarkhelm/numinar-coding-project.git
synced 2026-09-05 23:58:25 -04:00
506 lines
15 KiB
Python
506 lines
15 KiB
Python
"""Test configuration and fixtures for the backend tests."""
|
|
|
|
import logging
|
|
from collections.abc import AsyncGenerator
|
|
from collections.abc import Generator
|
|
from datetime import datetime
|
|
from datetime import timedelta
|
|
from datetime import timezone
|
|
from pathlib import Path
|
|
from typing import Awaitable
|
|
from typing import Callable
|
|
from typing import TypeVar
|
|
from unittest.mock import AsyncMock
|
|
from unittest.mock import MagicMock
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from dotenv import load_dotenv
|
|
from fastapi import Request
|
|
from fastapi import Response
|
|
from fastapi.responses import JSONResponse
|
|
from httpx import ASGITransport
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.exc import NoResultFound
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
|
|
# Ensure .env is loaded from base or parent directory before any project
|
|
# imports
|
|
def _load_env_from_base_or_parent() -> None:
|
|
base = Path(__file__).parent.parent # project root
|
|
env_path = base / ".env"
|
|
if env_path.exists():
|
|
load_dotenv(dotenv_path=env_path)
|
|
else:
|
|
parent_env = base.parent / ".env"
|
|
if parent_env.exists():
|
|
load_dotenv(dotenv_path=parent_env)
|
|
|
|
|
|
_load_env_from_base_or_parent()
|
|
from backend.main import app # noqa: E402
|
|
from backend.models import Booking # noqa: E402
|
|
from backend.models import BookingList # noqa: E402
|
|
from backend.models import Room # noqa: E402
|
|
from backend.models import RoomList # noqa: E402
|
|
from backend.models import User # noqa: E402
|
|
from backend.models import UserList # noqa: E402
|
|
from backend.schemas.bookings import BookingCreate # noqa: E402
|
|
from tests.types import BookingCreateData # noqa: E402
|
|
from tests.types import BookingUpdateData # noqa: E402
|
|
from tests.types import RoomData # noqa: E402
|
|
from tests.types import RoomUpdateData # noqa: E402
|
|
from tests.types import UserData # noqa: E402
|
|
|
|
|
|
# Modern handler type for exception handlers
|
|
ExcT = TypeVar("ExcT", bound=Exception)
|
|
HandlerType = Callable[[Request, ExcT], Response | Awaitable[Response]]
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_dependency_overrides() -> Generator[None, None, None]:
|
|
"""Fixture to clear FastAPI app dependency overrides before and after each test.
|
|
|
|
Yields:
|
|
None
|
|
"""
|
|
app.dependency_overrides.clear()
|
|
yield
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def async_session() -> AsyncMock:
|
|
"""Fixture to provide a mocked AsyncSession for database interactions.
|
|
|
|
Returns:
|
|
A mocked AsyncSession object for database interactions.
|
|
"""
|
|
return AsyncMock(spec=AsyncSession)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def client() -> AsyncGenerator[AsyncClient, None]:
|
|
"""Fixture to provide an AsyncClient for async endpoint testing.
|
|
|
|
Yields:
|
|
An HTTPX AsyncClient instance for making async HTTP requests.
|
|
"""
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
|
yield ac
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_logger(request: pytest.FixtureRequest) -> Generator[MagicMock, None, None]:
|
|
"""Fixture to mock a logger for a specified module path.
|
|
|
|
Args:
|
|
request: Pytest request object, used to pass the module path for patching.
|
|
|
|
Yields:
|
|
MagicMock: A mocked logger instance for the specified module path.
|
|
|
|
Raises:
|
|
ValueError: If the module path parameter is not provided.
|
|
|
|
Usage:
|
|
To use this fixture in a test module, parameterize it with the module path
|
|
where the logger is defined.
|
|
"""
|
|
# Require explicit module path for logger
|
|
if not hasattr(request, "param"):
|
|
raise ValueError(
|
|
"mock_logger fixture requires a module path parameter, e.g."
|
|
" 'backend.routers.invitees'"
|
|
)
|
|
module_path = request.param + ".logger"
|
|
|
|
mock_logger_instance = MagicMock(spec=logging.Logger)
|
|
with patch(module_path, mock_logger_instance):
|
|
yield mock_logger_instance
|
|
|
|
|
|
@pytest.fixture
|
|
def event_publisher() -> AsyncMock:
|
|
"""Fixture for an AsyncMock event publisher for event publishing tests.
|
|
|
|
Returns:
|
|
An async mock event publisher callable.
|
|
"""
|
|
return AsyncMock()
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_users() -> UserList:
|
|
"""Fixture to provide sample User objects for testing.
|
|
|
|
Returns:
|
|
A list of sample User objects.
|
|
"""
|
|
return [
|
|
User(email="user1@example.com", name="User One"),
|
|
User(email="user2@example.com", name="User Two"),
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_users_data(sample_users: UserList) -> list[UserData]:
|
|
"""Fixture to provide sample user data dictionaries for testing.
|
|
|
|
Args:
|
|
sample_users: The list of sample User objects.
|
|
|
|
Returns:
|
|
A list of dictionaries representing user data.
|
|
"""
|
|
return [{"email": user.email, "name": user.name} for user in sample_users]
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_rooms() -> RoomList:
|
|
"""Fixture to provide sample Room objects for testing.
|
|
|
|
Returns:
|
|
A list of sample Room objects.
|
|
"""
|
|
return [
|
|
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,
|
|
),
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_room(sample_rooms: RoomList) -> Room:
|
|
"""Fixture to provide a sample Room object for testing.
|
|
|
|
Args:
|
|
sample_rooms: The list of sample Room objects.
|
|
|
|
Returns:
|
|
A single sample Room object.
|
|
"""
|
|
return sample_rooms[0]
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_room_data(sample_rooms: RoomList) -> RoomData:
|
|
"""Fixture to provide sample room creation data for testing.
|
|
|
|
Args:
|
|
sample_rooms: The list of sample Room objects.
|
|
|
|
Returns:
|
|
A dictionary with sample room creation data.
|
|
"""
|
|
room = sample_rooms[0]
|
|
return {
|
|
"id": room.id,
|
|
"name": room.name,
|
|
"location": room.location,
|
|
"equipment": room.equipment,
|
|
"capacity": room.capacity,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def room_update_data() -> RoomUpdateData:
|
|
"""Fixture to provide sample room update data for testing.
|
|
|
|
Returns:
|
|
A dictionary with sample room update data.
|
|
"""
|
|
return {
|
|
"name": "Updated Room Name",
|
|
"location": "Updated Location",
|
|
"equipment": "Updated Equipment",
|
|
"capacity": 20,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def updated_room(sample_room: Room, room_update_data: RoomUpdateData) -> Room:
|
|
"""Fixture to provide an updated Room object for testing.
|
|
|
|
Args:
|
|
sample_room: The original sample Room object.
|
|
room_update_data: The dictionary with updated room data.
|
|
|
|
Returns:
|
|
An updated Room object with new data applied.
|
|
"""
|
|
updated = Room(
|
|
id=sample_room.id,
|
|
name=room_update_data["name"],
|
|
location=room_update_data["location"],
|
|
equipment=room_update_data["equipment"],
|
|
capacity=room_update_data["capacity"],
|
|
)
|
|
return updated
|
|
|
|
|
|
@pytest.fixture
|
|
def utcnow() -> datetime:
|
|
"""Fixture to provide a reference UTC datetime for tests.
|
|
|
|
Returns:
|
|
A datetime object representing the current UTC time.
|
|
"""
|
|
return datetime.now(timezone.utc)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_bookings(utcnow: datetime, sample_room: Room) -> BookingList:
|
|
"""Fixture to provide sample Booking objects for testing.
|
|
|
|
Args:
|
|
utcnow: The reference UTC datetime.
|
|
sample_room: The sample Room object to attach to bookings.
|
|
|
|
Returns:
|
|
A list of sample Booking objects with attached room relationships.
|
|
"""
|
|
# Always set bookings in the future (current month or later)
|
|
booking1 = Booking(
|
|
room_id=1,
|
|
start_time=utcnow + timedelta(days=1, hours=3),
|
|
end_time=utcnow + timedelta(days=1, hours=4),
|
|
title="Sample Booking 1",
|
|
invitees=["user1@example.com", "user2@example.com"],
|
|
)
|
|
booking1.id = 1
|
|
booking1.room = sample_room
|
|
booking2 = Booking(
|
|
room_id=1,
|
|
start_time=utcnow + timedelta(days=1, hours=12),
|
|
end_time=utcnow + timedelta(days=1, hours=13),
|
|
title="Sample Booking 2",
|
|
invitees=["user1@example.com", "user2@example.com", "user3@example.com"],
|
|
)
|
|
booking2.id = 2
|
|
booking2.room = sample_room
|
|
return [booking1, booking2]
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_booking(sample_bookings: BookingList) -> Booking:
|
|
"""Fixture to provide a sample Booking object for testing.
|
|
|
|
Args:
|
|
sample_bookings: The list of sample Booking objects.
|
|
|
|
Returns:
|
|
A single sample Booking object.
|
|
"""
|
|
return sample_bookings[0]
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_booking_data(sample_booking: Booking) -> BookingCreateData:
|
|
"""Fixture to provide sample booking creation data for testing.
|
|
|
|
Args:
|
|
sample_booking: The sample Booking object to base the data on.
|
|
|
|
Returns:
|
|
A dictionary with sample booking creation data.
|
|
"""
|
|
return {
|
|
"room_id": sample_booking.room_id,
|
|
"start_time": sample_booking.start_time.isoformat(),
|
|
"end_time": sample_booking.end_time.isoformat(),
|
|
"title": sample_booking.title,
|
|
"invitees": sample_booking.invitees,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def conflict_booking(sample_booking: Booking) -> Booking:
|
|
"""Fixture to provide a Booking object that conflicts with sample_booking.
|
|
|
|
Args:
|
|
sample_booking: The sample Booking object to base the conflict on.
|
|
|
|
Returns:
|
|
A Booking object that overlaps in time with sample_booking.
|
|
"""
|
|
conflict = Booking(
|
|
room_id=sample_booking.room_id,
|
|
start_time=sample_booking.start_time + timedelta(minutes=1),
|
|
end_time=sample_booking.end_time + timedelta(minutes=1),
|
|
title="Conflicting Booking",
|
|
invitees=["user3@example.com", "user4@example.com"],
|
|
)
|
|
conflict.id = 3
|
|
conflict.room = sample_booking.room # Preserve room relationship
|
|
return conflict
|
|
|
|
|
|
@pytest.fixture
|
|
def conflict_booking_data(conflict_booking: Booking) -> BookingCreateData:
|
|
"""Helper function to generate booking creation data for a conflicting booking.
|
|
|
|
Args:
|
|
conflict_booking: The conflicting Booking object.
|
|
|
|
Returns:
|
|
A dictionary with booking creation data that would cause a conflict.
|
|
"""
|
|
return {
|
|
"room_id": conflict_booking.room_id,
|
|
"start_time": conflict_booking.start_time.isoformat(),
|
|
"end_time": conflict_booking.end_time.isoformat(),
|
|
"title": conflict_booking.title,
|
|
"invitees": conflict_booking.invitees,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def booking_update_data(utcnow: datetime, sample_booking: Booking) -> BookingUpdateData:
|
|
"""Fixture to provide sample booking update data for testing.
|
|
|
|
Args:
|
|
utcnow: The reference UTC datetime.
|
|
sample_booking: The sample Booking object to base the data on.
|
|
|
|
Returns:
|
|
A dictionary with sample booking update data.
|
|
"""
|
|
return {
|
|
"id": sample_booking.id,
|
|
"title": "Updated Booking Title",
|
|
"start_time": (utcnow + timedelta(hours=5)).isoformat(),
|
|
"end_time": (utcnow + timedelta(hours=6)).isoformat(),
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def updated_booking(
|
|
sample_booking: Booking, booking_update_data: BookingUpdateData
|
|
) -> Booking:
|
|
"""Fixture to provide an updated Booking object for testing.
|
|
|
|
Args:
|
|
sample_booking: The original sample Booking object.
|
|
booking_update_data: The dictionary with updated booking data.
|
|
|
|
Returns:
|
|
An updated Booking object with new data applied.
|
|
"""
|
|
updated = Booking(
|
|
room_id=2, # Match the update data in the test
|
|
title=booking_update_data.get("title", sample_booking.title),
|
|
start_time=datetime.fromisoformat(
|
|
booking_update_data.get("start_time", sample_booking.start_time.isoformat())
|
|
),
|
|
end_time=datetime.fromisoformat(
|
|
booking_update_data.get("end_time", sample_booking.end_time.isoformat())
|
|
),
|
|
invitees=sample_booking.invitees,
|
|
)
|
|
updated.id = sample_booking.id # Preserve ID
|
|
updated.room = sample_booking.room # Preserve room relationship
|
|
return updated
|
|
|
|
|
|
@pytest.fixture
|
|
def conflict_booking_update_data(sample_bookings: BookingList) -> BookingUpdateData:
|
|
"""Fixture to provide booking update data that would cause a conflict.
|
|
|
|
Args:
|
|
sample_bookings: The list of sample Booking objects.
|
|
|
|
Returns:
|
|
A dictionary with booking update data that would cause a conflict.
|
|
"""
|
|
return {
|
|
"id": sample_bookings[0].id,
|
|
"title": "Conflict Update",
|
|
"room_id": sample_bookings[1].room_id,
|
|
"start_time": sample_bookings[1].start_time.isoformat(),
|
|
"end_time": sample_bookings[1].end_time.isoformat(),
|
|
}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def override_exception_handler() -> Generator[None, None, None]:
|
|
"""Override exception handlers for testing to ensure correct status codes.
|
|
|
|
Yields:
|
|
None
|
|
"""
|
|
original_handlers: dict[type[Exception], HandlerType[Exception] | None] = {
|
|
Exception: app.exception_handlers.get(Exception),
|
|
NoResultFound: app.exception_handlers.get(NoResultFound),
|
|
ValueError: app.exception_handlers.get(ValueError),
|
|
SQLAlchemyError: app.exception_handlers.get(SQLAlchemyError),
|
|
}
|
|
|
|
async def test_exception_handler(request: Request, exc: Exception) -> JSONResponse:
|
|
return JSONResponse(
|
|
status_code=500, content={"detail": "Internal server error"}
|
|
)
|
|
|
|
async def not_found_handler(request: Request, exc: Exception) -> JSONResponse:
|
|
return JSONResponse(status_code=404, content={"detail": "Booking not found"})
|
|
|
|
async def value_error_handler(request: Request, exc: Exception) -> JSONResponse:
|
|
# Use 409 for booking conflict, 400 for other value errors
|
|
detail = str(exc)
|
|
if "overlap" in detail:
|
|
return JSONResponse(status_code=409, content={"detail": detail})
|
|
return JSONResponse(status_code=400, content={"detail": detail})
|
|
|
|
async def sqlalchemy_error_handler(
|
|
request: Request, exc: Exception
|
|
) -> JSONResponse:
|
|
return JSONResponse(status_code=500, content={"detail": "Database error"})
|
|
|
|
app.add_exception_handler(Exception, test_exception_handler)
|
|
app.add_exception_handler(NoResultFound, not_found_handler)
|
|
app.add_exception_handler(ValueError, value_error_handler)
|
|
app.add_exception_handler(SQLAlchemyError, sqlalchemy_error_handler)
|
|
yield
|
|
# Restore original handlers
|
|
for exc_type, handler in original_handlers.items():
|
|
if handler is not None:
|
|
app.add_exception_handler(exc_type, handler)
|
|
else:
|
|
if exc_type in app.exception_handlers:
|
|
del app.exception_handlers[exc_type]
|
|
|
|
|
|
@pytest.fixture
|
|
def booking_create(sample_booking: Booking) -> BookingCreate:
|
|
"""Fixture for BookingCreate input data for create tests, derived from sample_booking.
|
|
|
|
Args:
|
|
sample_booking: The sample Booking object to base the creation data on.
|
|
|
|
Returns:
|
|
A BookingCreate object with the relevant data.
|
|
"""
|
|
return BookingCreate(
|
|
room_id=sample_booking.room_id,
|
|
start_time=sample_booking.start_time,
|
|
end_time=sample_booking.end_time,
|
|
title=sample_booking.title,
|
|
invitees=list(sample_booking.invitees),
|
|
)
|