Use Case 14: Testing Guide – Mocking Motor & Population Engine

Scenario: You want to write unit tests for repositories that use async Motor collections and the PopulationEngine without connecting to a real MongoDB instance.

---

📦 What’s New?

Component	Description
AsyncMock for collections	Allows you to stub find_one, insert_one, etc., and control returned data.
Patching MongoConnectionManager.get_database	Redirects repository initialization to a mock collection.
Mocking PopulationEngine	Replace the real engine with a lightweight stub that returns pre‑crafted objects.
Example test files	tests/test_populating_repository.py and tests/test_transactions.py are used as reference implementations.

---

🚀 Example Test Boilerplate

Python

import pytest
from unittest.mock import AsyncMock, patch
from bson import ObjectId

from mongo_ops.models import BaseDocument
from mongo_ops.populate import PopulateRule, PopulationEngine
from mongo_ops.repository import PopulatingRepository

# ----------------------------------------------------------------------
# 1️⃣ Define simple models (same as in the library tests)
# ----------------------------------------------------------------------
class Profile(BaseDocument):
    avatar_url: str = ""

class User(BaseDocument):
    name: str = ""
    profile_id: ObjectId | None = None
    profile: Profile | None = None

# ----------------------------------------------------------------------
# 2️⃣ Fixture – mock Motor collection
# ----------------------------------------------------------------------
@pytest.fixture
def mock_collection():
    return AsyncMock()

# ----------------------------------------------------------------------
# 3️⃣ Fixture – mock PopulationEngine (optional)
# ----------------------------------------------------------------------
@pytest.fixture
def engine():
    # Provide a real engine but replace the repo for "profiles" with a mock
    profile_repo = AsyncMock()
    return PopulationEngine({"profiles": profile_repo})

# ----------------------------------------------------------------------
# 4️⃣ Fixture – repository under test
# ----------------------------------------------------------------------
@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
        repo = PopulatingRepository(
            "users",
            User,
            population_engine=engine,
            populate_rules=[
                PopulateRule(
                    field_name="profile",
                    collection_name="profiles",
                    ref_field="profile_id",
                )
            ],
        )
        repo.collection = mock_collection
        return repo

# ----------------------------------------------------------------------
# 5️⃣ Test – `get_by_id` populates the related document
# ----------------------------------------------------------------------
@pytest.mark.asyncio
async def test_get_by_id_populates(repo, mock_collection, engine):
    # Arrange – user document returned from the DB
    uid = ObjectId()
    pid = ObjectId()
    mock_collection.find_one.return_value = {
        "_id": uid,
        "name": "Alice",
        "profile_id": pid,
        "created_at": "2024-01-01T00:00:00",
        "updated_at": "2024-01-01T00:00:00",
    }
    # Mock the profile repo to return a concrete Profile instance
    engine._repos["profiles"].get_by_id.return_value = Profile(id=pid, avatar_url="pic.png")

    # Act
    result = await repo.get_by_id(uid)

    # Assert
    assert result is not None
    assert result.name == "Alice"
    assert result.profile is not None
    assert result.profile.avatar_url == "pic.png"

# ----------------------------------------------------------------------
# 6️⃣ Test – transaction helper uses the real `TransactionManager`
# ----------------------------------------------------------------------
from mongo_ops.transactions import TransactionManager

@pytest.mark.asyncio
async def test_execute_transaction(monkeypatch):
    # Fake a Motor client session that records calls
    async def fake_op(session):
        return "ok"
    # Execute with the helper – it should return a list with the result
    results = await TransactionManager.execute_transaction([fake_op])
    assert results == ["ok"]

---

💡 Tips

• Never hit the network: All DB calls are mocked, so tests run instantly.
• Reuse fixtures: Keep the mock_collection and engine fixtures in a conftest.py file for other repo tests.
• Coverage: The same pattern works for CachedBaseRepository – just mock cache_backend methods (get, set, delete).
• AsyncTestCase: If you prefer unittest style, use IsolatedAsyncioTestCase from Python 3.8+.

---