127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
"""Unit tests for DB session singleton pattern (issue 001).
|
|
|
|
Verifies that the engine is created only once per process and that sessions
|
|
from the API dependency reuse the shared factory instead of building engines
|
|
on every request.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from contract_check.api.deps import get_db_session
|
|
from contract_check.core.db import session as session_mod
|
|
from contract_check.core.db.session import (
|
|
_get_engine,
|
|
_get_session_factory,
|
|
create_engine,
|
|
dispose_engine,
|
|
get_session,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def reset_singletons() -> Any:
|
|
"""Isolate the module-level singletons between tests."""
|
|
session_mod._engine = None
|
|
session_mod._session_factory = None
|
|
yield
|
|
session_mod._engine = None
|
|
session_mod._session_factory = None
|
|
|
|
|
|
def _mock_session() -> AsyncMock:
|
|
session = AsyncMock()
|
|
session.__aenter__.return_value = session
|
|
session.__aexit__.return_value = False
|
|
return session
|
|
|
|
|
|
def _mock_factory(session: AsyncMock) -> MagicMock:
|
|
factory = MagicMock()
|
|
factory.return_value = session
|
|
return factory
|
|
|
|
|
|
async def test_engine_created_once_across_many_requests() -> None:
|
|
"""N get_db_session invocations must build exactly one engine (AC)."""
|
|
with (
|
|
patch("contract_check.core.db.session.create_async_engine") as mock_create,
|
|
patch("contract_check.core.db.session.async_sessionmaker") as mock_maker,
|
|
):
|
|
mock_create.return_value = AsyncMock()
|
|
mock_maker.return_value = _mock_factory(_mock_session())
|
|
|
|
for _ in range(5):
|
|
agen = get_db_session()
|
|
await agen.__anext__()
|
|
await agen.aclose()
|
|
|
|
mock_create.assert_called_once()
|
|
|
|
|
|
async def test_session_factory_singleton() -> None:
|
|
with patch("contract_check.core.db.session._get_engine") as mock_get_engine:
|
|
mock_get_engine.return_value = MagicMock()
|
|
|
|
factory1 = _get_session_factory()
|
|
factory2 = _get_session_factory()
|
|
|
|
assert factory1 is factory2
|
|
mock_get_engine.assert_called_once()
|
|
|
|
|
|
async def test_get_session_uses_shared_factory() -> None:
|
|
session = _mock_session()
|
|
with patch("contract_check.core.db.session._get_session_factory") as mock_get_factory:
|
|
mock_get_factory.return_value = _mock_factory(session)
|
|
|
|
agen = get_session()
|
|
assert await agen.__anext__() is session
|
|
await agen.aclose()
|
|
|
|
mock_get_factory.assert_called_once()
|
|
|
|
|
|
async def test_get_session_accepts_explicit_factory() -> None:
|
|
session = _mock_session()
|
|
factory = _mock_factory(session)
|
|
|
|
with patch("contract_check.core.db.session._get_session_factory") as mock_get_factory:
|
|
agen = get_session(factory=factory)
|
|
assert await agen.__anext__() is session
|
|
await agen.aclose()
|
|
|
|
mock_get_factory.assert_not_called()
|
|
|
|
|
|
async def test_dispose_engine_disposes_and_resets_singletons() -> None:
|
|
with patch("contract_check.core.db.session.create_async_engine") as mock_create:
|
|
engine1 = AsyncMock()
|
|
engine2 = AsyncMock()
|
|
mock_create.side_effect = [engine1, engine2]
|
|
|
|
assert _get_engine() is engine1
|
|
await dispose_engine()
|
|
assert _get_engine() is engine2
|
|
|
|
engine1.dispose.assert_awaited_once()
|
|
assert session_mod._session_factory is None
|
|
assert mock_create.call_count == 2
|
|
|
|
|
|
async def test_dispose_engine_noop_when_never_built() -> None:
|
|
with patch("contract_check.core.db.session.create_async_engine") as mock_create:
|
|
await dispose_engine()
|
|
mock_create.assert_not_called()
|
|
|
|
|
|
async def test_explicit_create_still_builds_fresh_engines() -> None:
|
|
with patch("contract_check.core.db.session.create_async_engine") as mock_create:
|
|
mock_create.side_effect = [AsyncMock(), AsyncMock()]
|
|
|
|
assert create_engine() is not create_engine()
|
|
assert mock_create.call_count == 2
|