0
# Testing and Integration Utilities
1

2
Utilities for test integration including patching, feature management, and development helpers for seamless testing workflows. These tools enable easy integration of mongomock into existing test suites.
3

4
## Capabilities
5

6
### MongoClient Patching
7

8
Replace PyMongo's MongoClient with mongomock for testing without code changes.
9

10
```python { .api }
11
def patch(servers='localhost', on_new='error'):
12
    """
13
    Patch pymongo.MongoClient to use mongomock.
14

15
    Parameters:
16
    - servers: str or list, server addresses to patch (ignored in mock)
17
    - on_new: str, behavior when new client created
18
      - 'error': raise exception for new clients
19
      - 'create': create new mock client
20
      - 'timeout': simulate connection timeout
21
      - 'pymongo': use real PyMongo client
22

23
    Returns:
24
    unittest.mock._patch: patch object for context manager or decorator usage
25

26
    Raises:
27
    ImportError: if pymongo not available and on_new is 'pymongo'
28
    """
29
```
30

31
**Usage Example:**
32

33
```python
34
import mongomock
35

36
# Use as context manager
37
with mongomock.patch():
38
    import pymongo
39
    client = pymongo.MongoClient()  # Actually creates mongomock.MongoClient
40
    db = client.testdb
41
    collection = db.testcol
42
    collection.insert_one({'test': 'data'})
43
    assert collection.count_documents({}) == 1
44

45
# Use as decorator
46
@mongomock.patch()
47
def test_with_mock_client():
48
    import pymongo
49
    client = pymongo.MongoClient('mongodb://localhost:27017/')
50
    # This is actually a mongomock client
51
    assert isinstance(client, mongomock.MongoClient)
52

53
# Use with unittest
54
import unittest
55

56
class TestMongoDB(unittest.TestCase):
57
    @mongomock.patch()
58
    def test_database_operations(self):
59
        import pymongo
60
        client = pymongo.MongoClient()
61
        db = client.testdb
62
        
63
        # All operations use mongomock
64
        result = db.users.insert_one({'name': 'Alice'})
65
        self.assertIsNotNone(result.inserted_id)
66
        
67
        user = db.users.find_one({'name': 'Alice'})
68
        self.assertEqual(user['name'], 'Alice')
69
```
70

71
**Advanced Patching Options:**
72

73
```python
74
# Patch with error on new client creation
75
with mongomock.patch(on_new='error'):
76
    import pymongo
77
    client1 = pymongo.MongoClient()  # Works
78
    # client2 = pymongo.MongoClient()  # Would raise error
79

80
# Patch with timeout simulation
81
with mongomock.patch(on_new='timeout'):
82
    import pymongo
83
    try:
84
        client = pymongo.MongoClient()
85
    except pymongo.errors.ServerSelectionTimeoutError:
86
        print("Simulated connection timeout")
87

88
# Patch specific servers
89
with mongomock.patch(servers=['mongodb1:27017', 'mongodb2:27017']):
90
    import pymongo
91
    client = pymongo.MongoClient('mongodb1:27017')  # Uses mock
92
```
93

94
### Feature Management
95

96
Control which MongoDB features are supported or generate warnings.
97

98
```python { .api }
99
def ignore_feature(feature):
100
    """
101
    Ignore a not-implemented feature instead of raising NotImplementedError.
102

103
    Parameters:
104
    - feature: str, feature name to ignore
105
      - 'collation': collation support in queries
106
      - 'session': client session support
107

108
    Returns:
109
    None
110
    """
111

112
def warn_on_feature(feature):
113
    """
114
    Re-enable warnings for a feature.
115

116
    Parameters:
117
    - feature: str, feature name to warn about
118

119
    Returns:
120
    None
121
    """
122
```
123

124
**Usage Example:**
125

126
```python
127
import mongomock
128

129
# Ignore collation features (don't raise NotImplementedError)
130
mongomock.ignore_feature('collation')
131

132
client = mongomock.MongoClient()
133
collection = client.db.test
134

135
# This won't raise an error anymore
136
result = collection.find({'name': 'Alice'}, collation={'locale': 'en_US'})
137

138
# Re-enable warnings for sessions
139
mongomock.warn_on_feature('session')
140

141
# This will generate warnings but not fail
142
with client.start_session() as session:
143
    collection.insert_one({'data': 'test'}, session=session)
144

145
# Feature management in test setup
146
class TestMongoDB(unittest.TestCase):
147
    def setUp(self):
148
        # Ignore features not relevant to tests
149
        mongomock.ignore_feature('collation')
150
        mongomock.ignore_feature('session')
151
        
152
        self.client = mongomock.MongoClient()
153
        self.db = self.client.testdb
154
        
155
    def test_operations_ignore_unsupported_features(self):
156
        # These operations won't fail due to unsupported features
157
        result = self.db.collection.find({}, collation={'locale': 'en'})
158
        self.assertIsNotNone(result)
159
```
160

161
### GridFS Integration
162

163
Enable GridFS support for file storage operations.
164

165
```python { .api }
166
def enable_gridfs_integration():
167
    """
168
    Enable GridFS support with mongomock.
169
    
170
    This function sets up the necessary components to use
171
    GridFS operations with mongomock collections.
172

173
    Returns:
174
    None
175
    """
176
```
177

178
**Usage Example:**
179

180
```python
181
import mongomock
182
import gridfs
183

184
# Enable GridFS support
185
mongomock.enable_gridfs_integration()
186

187
client = mongomock.MongoClient()
188
db = client.gridfs_test
189

190
# Use GridFS with mongomock
191
fs = gridfs.GridFS(db)
192

193
# Store file
194
file_id = fs.put(b"Hello, GridFS!", filename="test.txt")
195

196
# Retrieve file
197
retrieved_file = fs.get(file_id)
198
content = retrieved_file.read()
199
filename = retrieved_file.filename
200

201
print(f"Retrieved file '{filename}': {content}")
202

203
# List files
204
for file_doc in fs.find():
205
    print(f"File: {file_doc.filename}, Size: {file_doc.length}")
206
```
207

208
### Test Database Management
209

210
Utilities for managing test databases and collections.
211

212
**Database Isolation:**
213

214
```python
215
import mongomock
216
import unittest
217

218
class TestDatabaseOperations(unittest.TestCase):
219
    def setUp(self):
220
        """Set up isolated test database."""
221
        self.client = mongomock.MongoClient()
222
        self.db = self.client.test_db
223
        self.collection = self.db.test_collection
224
        
225
    def tearDown(self):
226
        """Clean up test data."""
227
        # Drop test database to ensure isolation
228
        self.client.drop_database('test_db')
229
    
230
    def test_insert_and_find(self):
231
        """Test basic operations in isolated environment."""
232
        # Insert test data
233
        self.collection.insert_one({'name': 'Test User', 'age': 25})
234
        
235
        # Verify data exists
236
        user = self.collection.find_one({'name': 'Test User'})
237
        self.assertIsNotNone(user)
238
        self.assertEqual(user['age'], 25)
239
    
240
    def test_collection_isolation(self):
241
        """Test that collections are properly isolated."""
242
        # Create data in test collection
243
        self.collection.insert_one({'test': 'data'})
244
        
245
        # Verify other collections are empty
246
        other_collection = self.db.other_collection
247
        count = other_collection.count_documents({})
248
        self.assertEqual(count, 0)
249
```
250

251
**Shared Test Data:**
252

253
```python
254
class TestWithSharedData(unittest.TestCase):
255
    @classmethod
256
    def setUpClass(cls):
257
        """Set up shared test data for all test methods."""
258
        cls.client = mongomock.MongoClient()
259
        cls.db = cls.client.shared_test_db
260
        cls.users = cls.db.users
261
        
262
        # Insert shared test data
263
        cls.test_users = [
264
            {'name': 'Alice', 'age': 30, 'department': 'Engineering'},
265
            {'name': 'Bob', 'age': 25, 'department': 'Marketing'},
266
            {'name': 'Charlie', 'age': 35, 'department': 'Engineering'}
267
        ]
268
        cls.users.insert_many(cls.test_users)
269
        
270
        # Create indexes for testing
271
        cls.users.create_index('email', unique=True)
272
        cls.users.create_index([('department', 1), ('age', -1)])
273
    
274
    @classmethod
275
    def tearDownClass(cls):
276
        """Clean up shared test data."""
277
        cls.client.drop_database('shared_test_db')
278
    
279
    def test_query_shared_data(self):
280
        """Test querying shared test data."""
281
        engineers = list(self.users.find({'department': 'Engineering'}))
282
        self.assertEqual(len(engineers), 2)
283
    
284
    def test_aggregation_on_shared_data(self):
285
        """Test aggregation on shared test data."""
286
        pipeline = [
287
            {'$group': {
288
                '_id': '$department',
289
                'avg_age': {'$avg': '$age'},
290
                'count': {'$sum': 1}
291
            }}
292
        ]
293
        results = list(self.users.aggregate(pipeline))
294
        self.assertEqual(len(results), 2)
295
```
296

297
### Test Assertions and Helpers
298

299
Helper functions for common test assertions and validations.
300

301
```python
302
def assert_document_exists(collection, filter_doc, msg=None):
303
    """Assert that a document matching filter exists."""
304
    doc = collection.find_one(filter_doc)
305
    if doc is None:
306
        raise AssertionError(msg or f"Document {filter_doc} not found")
307
    return doc
308

309
def assert_document_count(collection, filter_doc, expected_count, msg=None):
310
    """Assert that the number of matching documents equals expected count."""
311
    actual_count = collection.count_documents(filter_doc)
312
    if actual_count != expected_count:
313
        raise AssertionError(
314
            msg or f"Expected {expected_count} documents, found {actual_count}"
315
        )
316

317
def assert_index_exists(collection, index_name, msg=None):
318
    """Assert that an index exists on the collection."""
319
    indexes = collection.index_information()
320
    if index_name not in indexes:
321
        raise AssertionError(
322
            msg or f"Index '{index_name}' not found in {list(indexes.keys())}"
323
        )
324

325
# Usage in tests
326
class TestAssertionHelpers(unittest.TestCase):
327
    def setUp(self):
328
        self.collection = mongomock.MongoClient().db.test
329
        
330
    def test_with_helpers(self):
331
        # Insert test data
332
        self.collection.insert_many([
333
            {'name': 'Alice', 'status': 'active'},
334
            {'name': 'Bob', 'status': 'inactive'},
335
            {'name': 'Charlie', 'status': 'active'}
336
        ])
337
        
338
        # Use assertion helpers
339
        assert_document_exists(self.collection, {'name': 'Alice'})
340
        assert_document_count(self.collection, {'status': 'active'}, 2)
341
        
342
        # Create and verify index
343
        self.collection.create_index('name')
344
        assert_index_exists(self.collection, 'name_1')
345
```
346

347
### Performance Testing
348

349
Tools for testing performance characteristics and behavior under load.
350

351
```python
352
import time
353
import mongomock
354

355
def benchmark_operation(operation, iterations=1000):
356
    """Benchmark a mongomock operation."""
357
    start_time = time.time()
358
    
359
    for _ in range(iterations):
360
        operation()
361
    
362
    end_time = time.time()
363
    total_time = end_time - start_time
364
    avg_time = total_time / iterations
365
    
366
    return {
367
        'total_time': total_time,
368
        'avg_time': avg_time,
369
        'operations_per_second': iterations / total_time
370
    }
371

372
# Usage example
373
def test_insert_performance():
374
    """Test insert operation performance."""
375
    collection = mongomock.MongoClient().db.perf_test
376
    
377
    def insert_op():
378
        collection.insert_one({'data': 'test', 'timestamp': time.time()})
379
    
380
    results = benchmark_operation(insert_op, 1000)
381
    print(f"Insert performance: {results['operations_per_second']:.2f} ops/sec")
382
    
383
    # Verify all documents were inserted
384
    count = collection.count_documents({})
385
    assert count == 1000
386

387
def test_query_performance():
388
    """Test query operation performance."""
389
    collection = mongomock.MongoClient().db.perf_test
390
    
391
    # Set up test data
392
    test_data = [{'id': i, 'value': f'item_{i}'} for i in range(10000)]
393
    collection.insert_many(test_data)
394
    collection.create_index('id')
395
    
396
    def query_op():
397
        result = collection.find_one({'id': 5000})
398
        return result
399
    
400
    results = benchmark_operation(query_op, 1000)
401
    print(f"Query performance: {results['operations_per_second']:.2f} ops/sec")
402
```
403

404
### Integration Test Patterns
405

406
Common patterns for integration testing with mongomock.
407

408
```python
409
import mongomock
410
import unittest
411
from unittest.mock import patch
412

413
class IntegrationTestExample(unittest.TestCase):
414
    """Example integration test using mongomock."""
415
    
416
    def setUp(self):
417
        """Set up test environment."""
418
        self.client = mongomock.MongoClient()
419
        self.db = self.client.integration_test
420
        
421
    def test_full_user_workflow(self):
422
        """Test complete user management workflow."""
423
        users = self.db.users
424
        
425
        # Create user
426
        user_data = {
427
            'username': 'testuser',
428
            'email': 'test@example.com',
429
            'created_at': datetime.utcnow(),
430
            'status': 'active'
431
        }
432
        result = users.insert_one(user_data)
433
        user_id = result.inserted_id
434
        
435
        # Verify user creation
436
        created_user = users.find_one({'_id': user_id})
437
        self.assertEqual(created_user['username'], 'testuser')
438
        
439
        # Update user
440
        update_result = users.update_one(
441
            {'_id': user_id},
442
            {'$set': {'last_login': datetime.utcnow()}}
443
        )
444
        self.assertEqual(update_result.modified_count, 1)
445
        
446
        # Query with complex filter
447
        active_users = list(users.find({
448
            'status': 'active',
449
            'created_at': {'$gte': datetime.utcnow() - timedelta(days=1)}
450
        }))
451
        self.assertEqual(len(active_users), 1)
452
        
453
        # Aggregation
454
        stats = list(users.aggregate([
455
            {'$group': {
456
                '_id': '$status',
457
                'count': {'$sum': 1}
458
            }}
459
        ]))
460
        self.assertEqual(len(stats), 1)
461
        self.assertEqual(stats[0]['count'], 1)
462
        
463
        # Delete user
464
        delete_result = users.delete_one({'_id': user_id})
465
        self.assertEqual(delete_result.deleted_count, 1)
466
        
467
        # Verify deletion
468
        deleted_user = users.find_one({'_id': user_id})
469
        self.assertIsNone(deleted_user)
470

471
    @mongomock.patch()
472
    def test_with_real_pymongo_imports(self):
473
        """Test using real PyMongo imports with mongomock patch."""
474
        import pymongo
475
        
476
        # This actually creates a mongomock client
477
        client = pymongo.MongoClient()
478
        db = client.test_db
479
        
480
        # All operations use mongomock
481
        collection = db.test_collection
482
        result = collection.insert_one({'patched': True})
483
        
484
        found = collection.find_one({'patched': True})
485
        self.assertIsNotNone(found)
486
        self.assertTrue(found['patched'])
487
```
488

489
## Testing Best Practices
490

491
**Isolation and Cleanup:**
492

493
```python
494
# Always use unique database names for tests
495
def get_test_db_name():
496
    return f"test_{uuid.uuid4().hex[:8]}"
497

498
# Clean up after tests
499
class CleanTestCase(unittest.TestCase):
500
    def setUp(self):
501
        self.client = mongomock.MongoClient()
502
        self.db_name = get_test_db_name()
503
        self.db = self.client[self.db_name]
504
        
505
    def tearDown(self):
506
        self.client.drop_database(self.db_name)
507
```
508

509
**Realistic Test Data:**
510

511
```python
512
# Use realistic data structures and volumes
513
def create_realistic_test_data():
514
    return [
515
        {
516
            'user_id': str(ObjectId()),
517
            'username': f'user_{i}',
518
            'email': f'user{i}@example.com',
519
            'profile': {
520
                'age': random.randint(18, 80),
521
                'interests': random.sample(['sports', 'music', 'art', 'tech'], 2)
522
            },
523
            'created_at': datetime.utcnow() - timedelta(days=random.randint(1, 365))
524
        }
525
        for i in range(1000)
526
    ]
527
```