7.3MongoDB Testing

MongoDB stores data as JSON-like documents (BSON). Documents in the same collection can have different structures — which is both a feature and a testing challenge.

Document Structure Validation

from pymongo import MongoClient
import pytest

@pytest.fixture(scope="session")
def mongo_db():
    client = MongoClient("mongodb://localhost:27017")
    db = client["testdb"]
    yield db
    client.close()

def test_user_document_structure(mongo_db):
    """Verify user documents have expected fields."""
    mongo_db.users.insert_one({
        "name": "Alice",
        "email": "alice@test.com",
        "tags": ["admin"],
        "profile": {"bio": "Test user", "avatar_url": None}
    })

    user = mongo_db.users.find_one({"email": "alice@test.com"})
    assert user is not None
    assert isinstance(user["tags"], list)
    assert "admin" in user["tags"]
    assert isinstance(user["profile"], dict)
    assert "bio" in user["profile"]

def test_schema_validation_enforced(mongo_db):
    """If schema validation is configured, invalid documents should be rejected."""
    # Set up schema validation
    mongo_db.command({
        "collMod": "strict_users",
        "validator": {
            "$jsonSchema": {
                "bsonType": "object",
                "required": ["name", "email"],
                "properties": {
                    "name": {"bsonType": "string"},
                    "email": {"bsonType": "string", "pattern": "^.+@.+$"}
                }
            }
        }
    })

    # Valid document should succeed
    mongo_db.strict_users.insert_one({"name": "Valid", "email": "valid@test.com"})

    # Invalid document should fail
    from pymongo.errors import WriteError
    with pytest.raises(WriteError):
        mongo_db.strict_users.insert_one({"name": "Invalid"})  # Missing email

MongoDB Aggregation Testing

def test_aggregation_pipeline(mongo_db):
    """Test an aggregation pipeline that calculates user statistics."""
    # Insert test data
    mongo_db.orders.insert_many([
        {"user_id": "user-1", "total": 50.00, "status": "completed"},
        {"user_id": "user-1", "total": 75.00, "status": "completed"},
        {"user_id": "user-1", "total": 25.00, "status": "cancelled"},
        {"user_id": "user-2", "total": 100.00, "status": "completed"},
    ])

    # Run aggregation
    pipeline = [
        {"$match": {"status": "completed"}},
        {"$group": {
            "_id": "$user_id",
            "total_spent": {"$sum": "$total"},
            "order_count": {"$sum": 1}
        }},
        {"$sort": {"total_spent": -1}}
    ]
    results = list(mongo_db.orders.aggregate(pipeline))

    assert len(results) == 2
    assert results[0]["_id"] == "user-1"
    assert results[0]["total_spent"] == 125.00  # 50 + 75 (cancelled excluded)
    assert results[0]["order_count"] == 2