182 lines
5.8 KiB
Python
182 lines
5.8 KiB
Python
"""Unit tests for database wiring helpers."""
|
|
|
|
from typing import Any, cast
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
|
|
|
|
import backend.database as database
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_database_singletons(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Reset database singletons before each test."""
|
|
monkeypatch.setattr(database, "_ENGINE", None)
|
|
monkeypatch.setattr(database, "_SESSIONMAKER", None)
|
|
|
|
|
|
def test_get_database_url_requires_configuration(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Database URL helper should fail when DATABASE_URL is not set."""
|
|
monkeypatch.delenv("DATABASE_URL", raising=False)
|
|
|
|
with pytest.raises(database.DatabaseConfigurationError):
|
|
database.get_database_url()
|
|
|
|
|
|
def test_get_database_url_rewrites_postgresql_scheme(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Database URL helper should normalize URLs for SQLAlchemy async psycopg."""
|
|
monkeypatch.setenv("DATABASE_URL", "postgresql://user:pass@db:5432/app")
|
|
|
|
assert database.get_database_url() == "postgresql+psycopg://user:pass@db:5432/app"
|
|
|
|
|
|
def test_get_database_url_keeps_existing_scheme(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Database URL helper should keep already-normalized URLs unchanged."""
|
|
monkeypatch.setenv("DATABASE_URL", "postgresql+psycopg://user:pass@db:5432/app")
|
|
|
|
assert database.get_database_url() == "postgresql+psycopg://user:pass@db:5432/app"
|
|
|
|
|
|
def test_get_engine_is_singleton(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Engine helper should create engine once and reuse it."""
|
|
created: list[tuple[str, bool]] = []
|
|
engine = cast("AsyncEngine", AsyncMock(spec=AsyncEngine))
|
|
|
|
def fake_create_async_engine(url: str, *, pool_pre_ping: bool) -> AsyncEngine:
|
|
created.append((url, pool_pre_ping))
|
|
return engine
|
|
|
|
monkeypatch.setenv("DATABASE_URL", "postgresql://user:pass@db:5432/app")
|
|
monkeypatch.setattr(database, "create_async_engine", fake_create_async_engine)
|
|
|
|
first = database.get_engine()
|
|
second = database.get_engine()
|
|
|
|
assert first is engine
|
|
assert second is engine
|
|
assert created == [("postgresql+psycopg://user:pass@db:5432/app", True)]
|
|
|
|
|
|
def test_get_sessionmaker_is_singleton(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Sessionmaker helper should create one factory and reuse it."""
|
|
engine = cast("AsyncEngine", AsyncMock(spec=AsyncEngine))
|
|
monkeypatch.setattr(database, "get_engine", lambda: engine)
|
|
|
|
first = database.get_sessionmaker()
|
|
second = database.get_sessionmaker()
|
|
|
|
assert first is second
|
|
assert first.kw["bind"] is engine
|
|
assert first.kw["expire_on_commit"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session_raises_503_when_db_not_configured(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Session dependency should translate config errors to HTTP 503."""
|
|
|
|
def raise_config_error() -> Any:
|
|
raise database.DatabaseConfigurationError("missing")
|
|
|
|
monkeypatch.setattr(database, "get_sessionmaker", raise_config_error)
|
|
|
|
generator = database.get_session()
|
|
with pytest.raises(HTTPException) as exc:
|
|
await anext(generator)
|
|
|
|
assert exc.value.status_code == 503
|
|
assert exc.value.detail == {"status": "unhealthy", "database": "not_configured"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_session_yields_scoped_session(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Session dependency should yield one session and close context afterwards."""
|
|
fake_session = cast("AsyncSession", AsyncMock(spec=AsyncSession))
|
|
|
|
class SessionContextManager:
|
|
"""Minimal async context manager used by the fake session factory."""
|
|
|
|
exited = False
|
|
|
|
async def __aenter__(self) -> AsyncSession:
|
|
return fake_session
|
|
|
|
async def __aexit__(self, *args: object) -> None:
|
|
self.exited = True
|
|
|
|
context_manager = SessionContextManager()
|
|
|
|
class SessionFactory:
|
|
"""Callable session factory returning an async context manager."""
|
|
|
|
def __call__(self) -> SessionContextManager:
|
|
return context_manager
|
|
|
|
monkeypatch.setattr(database, "get_sessionmaker", lambda: SessionFactory())
|
|
|
|
generator = database.get_session()
|
|
yielded = await anext(generator)
|
|
|
|
assert yielded is fake_session
|
|
|
|
with pytest.raises(StopAsyncIteration):
|
|
await anext(generator)
|
|
|
|
assert context_manager.exited is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_database_success() -> None:
|
|
"""Database probe should return True when SELECT 1 succeeds."""
|
|
session = cast("AsyncSession", AsyncMock(spec=AsyncSession))
|
|
session.execute = AsyncMock(return_value=1)
|
|
|
|
assert await database.probe_database(session) is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_database_failure() -> None:
|
|
"""Database probe should return False when SQLAlchemy raises an error."""
|
|
session = cast("AsyncSession", AsyncMock(spec=AsyncSession))
|
|
session.execute = AsyncMock(side_effect=SQLAlchemyError("db unavailable"))
|
|
|
|
assert await database.probe_database(session) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dispose_engine_resets_global_state(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Dispose helper should call engine dispose and reset cached globals."""
|
|
|
|
class FakeEngine:
|
|
"""Fake async engine exposing a dispose coroutine."""
|
|
|
|
def __init__(self) -> None:
|
|
self.disposed = False
|
|
|
|
async def dispose(self) -> None:
|
|
self.disposed = True
|
|
|
|
engine = FakeEngine()
|
|
monkeypatch.setattr(database, "_ENGINE", engine)
|
|
monkeypatch.setattr(database, "_SESSIONMAKER", "sessionmaker")
|
|
|
|
await database.dispose_engine()
|
|
|
|
assert engine.disposed is True
|
|
assert database._ENGINE is None
|
|
assert database._SESSIONMAKER is None
|