Use Case 14: Testing Guide — Mocking Motor, Cache, Registry & Transactions
Scenario: Write unit tests that never touch a real MongoDB — mock the collection, the cache backend, and the client, exactly like the library's own test suite (tests/).
📦 What's New?
| Component | Description |
|---|---|
AsyncMock collections |
Stub find_one, insert_one, find_one_and_update, cursor chains. |
Patching MongoConnectionManager.get_database |
Gives repositories a mock collection without a live connection. |
monkeypatch on get_client |
Fakes sessions for TransactionManager tests. |
| Reference tests | tests/test_populating_repository.py, tests/test_cache.py, tests/test_registry.py, tests/test_transactions.py. |
🚀 Example Boilerplate
Python
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
from bson import ObjectId
from mongo_ops.cache import CacheConfig, InMemoryCacheBackend
from mongo_ops.cache.repository import CachedBaseRepository
from mongo_ops.models import BaseDocument
from mongo_ops.registry import ModelRegistry
from mongo_ops.populate import PopulateRule, PopulationEngine
from mongo_ops.repository import PopulatingRepository
# ----------------------------------------------------------------------
# 1. Models (same shape as the library tests)
# ----------------------------------------------------------------------
class Profile(BaseDocument):
avatar_url: str = ""
class User(BaseDocument):
name: str = ""
profile: Profile | None = None # ObjectId in DB, Profile in memory
# ----------------------------------------------------------------------
# 2. Fixtures
# ----------------------------------------------------------------------
@pytest.fixture
def mock_collection():
return AsyncMock()
@pytest.fixture
def engine():
profile_repo = AsyncMock()
return PopulationEngine({"profiles": profile_repo})
@pytest.fixture
def repo(mock_collection, engine):
with patch("mongo_ops.repository.MongoConnectionManager.get_database") as mock_db:
mock_db.return_value.__getitem__.return_value = mock_collection
r = PopulatingRepository(
"users",
User,
population_engine=engine,
populate_rules=[PopulateRule(field_name="profile", collection_name="profiles")],
)
r.collection = mock_collection
return r
# ----------------------------------------------------------------------
# 3. Population — get_by_id resolves the reference
# ----------------------------------------------------------------------
@pytest.mark.asyncio
async def test_get_by_id_populates(repo, mock_collection, engine):
uid, pid = ObjectId(), ObjectId()
mock_collection.find_one.return_value = {
"_id": uid,
"name": "Alice",
"profile": pid, # ObjectId stored in DB
"created_at": "2024-01-01T00:00:00",
"updated_at": "2024-01-01T00:00:00",
}
engine._repos["profiles"].get_by_id.return_value = Profile(id=pid, avatar_url="pic.png")
result = await repo.get_by_id(uid)
assert result is not None
assert result.name == "Alice"
assert isinstance(result.profile, Profile)
assert result.profile.avatar_url == "pic.png"
# ----------------------------------------------------------------------
# 4. Patch FK guard
# ----------------------------------------------------------------------
@pytest.mark.asyncio
async def test_patch_rejects_fk_field(repo):
with pytest.raises(ValueError, match="Cannot patch FK fields"):
await repo.patch(ObjectId(), {"profile": ObjectId()})
# ----------------------------------------------------------------------
# 5. Registry — index specs pass through to create_index
# ----------------------------------------------------------------------
@pytest.mark.asyncio
async def test_initialize_all_creates_indexes():
ModelRegistry.register("users", User, indexes=[("email", 1)])
fake_collection = AsyncMock()
await ModelRegistry.initialize_all(db={"users": fake_collection})
fake_collection.create_index.assert_awaited_once_with(("email", 1))
# ----------------------------------------------------------------------
# 6. Cached repository — cache-first reads + invalidation
# ----------------------------------------------------------------------
@pytest.mark.asyncio
async def test_cached_get_by_id_populates_cache():
backend = InMemoryCacheBackend(
max_entries=100, default_ttl=300, cleanup_interval=9999
)
await backend.initialize()
try:
with patch("mongo_ops.repository.MongoConnectionManager.get_database") as mock_db:
mock_collection = AsyncMock()
mock_db.return_value.__getitem__.return_value = mock_collection
repo = CachedBaseRepository(
"users", User, backend, CacheConfig(enabled=True)
)
repo.collection = mock_collection
oid = ObjectId()
mock_collection.find_one.return_value = {
"_id": oid,
"name": "cached",
"created_at": "2024-01-01T00:00:00",
"updated_at": "2024-01-01T00:00:00",
}
first = await repo.get_by_id(oid)
assert first is not None
mock_collection.find_one.return_value = None # DB now "empty"
second = await repo.get_by_id(oid) # served from cache
assert second is not None
assert second.name == "cached"
mock_collection.find_one.assert_awaited_once() # only one DB read
finally:
await backend.shutdown()
# ----------------------------------------------------------------------
# 7. Transactions — fake the client's start_session
# ----------------------------------------------------------------------
from mongo_ops.transactions import TransactionManager
@pytest.mark.asyncio
async def test_execute_transaction(monkeypatch):
session_ctx = AsyncMock()
session_ctx.__aenter__.return_value = AsyncMock()
session_ctx.__aexit__.return_value = None
client = MagicMock()
client.start_session = AsyncMock(return_value=session_ctx)
# NOTE: start_transaction must return a context manager, not a coroutine.
async_session = session_ctx.__aenter__.return_value
async_session.start_transaction = lambda **_: session_ctx
monkeypatch.setattr(
"mongo_ops.transactions.MongoConnectionManager.get_client",
lambda: client,
)
async def fake_op(session):
return "ok"
results = await TransactionManager.execute_transaction([fake_op])
assert results == ["ok"]
💡 Tips
- pytest-asyncio is already configured in
pyproject.toml(asyncio_mode = "auto"), so@pytest.mark.asynciotests work out of the box. Run withpytest(coverage reports are enabled there too). - Never hit the network. Keep the patches in fixtures (or a
conftest.py) and reuse them. - Mock
cursorchains withMagicMock()+.to_list = AsyncMock(...), exactly liketests/test_repository.py. - For an optional integration check (real Mongo), use
MongoConnectionManager.lifespanagainst a local replica set and drop the test database in teardown — keep it separate from the unit suite. - The pattern works symmetrically for Redis: mock the
RedisCacheBackendmethods (get,set,delete) — no Redis process required.