Testing and Integration Utilities

def patch(servers='localhost', on_new='error'):
    """
    Patch pymongo.MongoClient to use mongomock.

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

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

    Raises:
    ImportError: if pymongo not available and on_new is 'pymongo'
    """

import mongomock

# Use as context manager
with mongomock.patch():
    import pymongo
    client = pymongo.MongoClient()  # Actually creates mongomock.MongoClient
    db = client.testdb
    collection = db.testcol
    collection.insert_one({'test': 'data'})
    assert collection.count_documents({}) == 1

# Use as decorator
@mongomock.patch()
def test_with_mock_client():
    import pymongo
    client = pymongo.MongoClient('mongodb://localhost:27017/')
    # This is actually a mongomock client
    assert isinstance(client, mongomock.MongoClient)

# Use with unittest
import unittest

class TestMongoDB(unittest.TestCase):
    @mongomock.patch()
    def test_database_operations(self):
        import pymongo
        client = pymongo.MongoClient()
        db = client.testdb
        
        # All operations use mongomock
        result = db.users.insert_one({'name': 'Alice'})
        self.assertIsNotNone(result.inserted_id)
        
        user = db.users.find_one({'name': 'Alice'})
        self.assertEqual(user['name'], 'Alice')

# Patch with error on new client creation
with mongomock.patch(on_new='error'):
    import pymongo
    client1 = pymongo.MongoClient()  # Works
    # client2 = pymongo.MongoClient()  # Would raise error

# Patch with timeout simulation
with mongomock.patch(on_new='timeout'):
    import pymongo
    try:
        client = pymongo.MongoClient()
    except pymongo.errors.ServerSelectionTimeoutError:
        print("Simulated connection timeout")

# Patch specific servers
with mongomock.patch(servers=['mongodb1:27017', 'mongodb2:27017']):
    import pymongo
    client = pymongo.MongoClient('mongodb1:27017')  # Uses mock

def ignore_feature(feature):
    """
    Ignore a not-implemented feature instead of raising NotImplementedError.

    Parameters:
    - feature: str, feature name to ignore
      - 'collation': collation support in queries
      - 'session': client session support

    Returns:
    None
    """

def warn_on_feature(feature):
    """
    Re-enable warnings for a feature.

    Parameters:
    - feature: str, feature name to warn about

    Returns:
    None
    """

import mongomock

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

client = mongomock.MongoClient()
collection = client.db.test

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

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

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

# Feature management in test setup
class TestMongoDB(unittest.TestCase):
    def setUp(self):
        # Ignore features not relevant to tests
        mongomock.ignore_feature('collation')
        mongomock.ignore_feature('session')
        
        self.client = mongomock.MongoClient()
        self.db = self.client.testdb
        
    def test_operations_ignore_unsupported_features(self):
        # These operations won't fail due to unsupported features
        result = self.db.collection.find({}, collation={'locale': 'en'})
        self.assertIsNotNone(result)

def enable_gridfs_integration():
    """
    Enable GridFS support with mongomock.
    
    This function sets up the necessary components to use
    GridFS operations with mongomock collections.

    Returns:
    None
    """

import mongomock
import gridfs

# Enable GridFS support
mongomock.enable_gridfs_integration()

client = mongomock.MongoClient()
db = client.gridfs_test

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

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

# Retrieve file
retrieved_file = fs.get(file_id)
content = retrieved_file.read()
filename = retrieved_file.filename

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

# List files
for file_doc in fs.find():
    print(f"File: {file_doc.filename}, Size: {file_doc.length}")

import mongomock
import unittest

class TestDatabaseOperations(unittest.TestCase):
    def setUp(self):
        """Set up isolated test database."""
        self.client = mongomock.MongoClient()
        self.db = self.client.test_db
        self.collection = self.db.test_collection
        
    def tearDown(self):
        """Clean up test data."""
        # Drop test database to ensure isolation
        self.client.drop_database('test_db')
    
    def test_insert_and_find(self):
        """Test basic operations in isolated environment."""
        # Insert test data
        self.collection.insert_one({'name': 'Test User', 'age': 25})
        
        # Verify data exists
        user = self.collection.find_one({'name': 'Test User'})
        self.assertIsNotNone(user)
        self.assertEqual(user['age'], 25)
    
    def test_collection_isolation(self):
        """Test that collections are properly isolated."""
        # Create data in test collection
        self.collection.insert_one({'test': 'data'})
        
        # Verify other collections are empty
        other_collection = self.db.other_collection
        count = other_collection.count_documents({})
        self.assertEqual(count, 0)

class TestWithSharedData(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        """Set up shared test data for all test methods."""
        cls.client = mongomock.MongoClient()
        cls.db = cls.client.shared_test_db
        cls.users = cls.db.users
        
        # Insert shared test data
        cls.test_users = [
            {'name': 'Alice', 'age': 30, 'department': 'Engineering'},
            {'name': 'Bob', 'age': 25, 'department': 'Marketing'},
            {'name': 'Charlie', 'age': 35, 'department': 'Engineering'}
        ]
        cls.users.insert_many(cls.test_users)
        
        # Create indexes for testing
        cls.users.create_index('email', unique=True)
        cls.users.create_index([('department', 1), ('age', -1)])
    
    @classmethod
    def tearDownClass(cls):
        """Clean up shared test data."""
        cls.client.drop_database('shared_test_db')
    
    def test_query_shared_data(self):
        """Test querying shared test data."""
        engineers = list(self.users.find({'department': 'Engineering'}))
        self.assertEqual(len(engineers), 2)
    
    def test_aggregation_on_shared_data(self):
        """Test aggregation on shared test data."""
        pipeline = [
            {'$group': {
                '_id': '$department',
                'avg_age': {'$avg': '$age'},
                'count': {'$sum': 1}
            }}
        ]
        results = list(self.users.aggregate(pipeline))
        self.assertEqual(len(results), 2)

def assert_document_exists(collection, filter_doc, msg=None):
    """Assert that a document matching filter exists."""
    doc = collection.find_one(filter_doc)
    if doc is None:
        raise AssertionError(msg or f"Document {filter_doc} not found")
    return doc

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

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

# Usage in tests
class TestAssertionHelpers(unittest.TestCase):
    def setUp(self):
        self.collection = mongomock.MongoClient().db.test
        
    def test_with_helpers(self):
        # Insert test data
        self.collection.insert_many([
            {'name': 'Alice', 'status': 'active'},
            {'name': 'Bob', 'status': 'inactive'},
            {'name': 'Charlie', 'status': 'active'}
        ])
        
        # Use assertion helpers
        assert_document_exists(self.collection, {'name': 'Alice'})
        assert_document_count(self.collection, {'status': 'active'}, 2)
        
        # Create and verify index
        self.collection.create_index('name')
        assert_index_exists(self.collection, 'name_1')

import time
import mongomock

def benchmark_operation(operation, iterations=1000):
    """Benchmark a mongomock operation."""
    start_time = time.time()
    
    for _ in range(iterations):
        operation()
    
    end_time = time.time()
    total_time = end_time - start_time
    avg_time = total_time / iterations
    
    return {
        'total_time': total_time,
        'avg_time': avg_time,
        'operations_per_second': iterations / total_time
    }

# Usage example
def test_insert_performance():
    """Test insert operation performance."""
    collection = mongomock.MongoClient().db.perf_test
    
    def insert_op():
        collection.insert_one({'data': 'test', 'timestamp': time.time()})
    
    results = benchmark_operation(insert_op, 1000)
    print(f"Insert performance: {results['operations_per_second']:.2f} ops/sec")
    
    # Verify all documents were inserted
    count = collection.count_documents({})
    assert count == 1000

def test_query_performance():
    """Test query operation performance."""
    collection = mongomock.MongoClient().db.perf_test
    
    # Set up test data
    test_data = [{'id': i, 'value': f'item_{i}'} for i in range(10000)]
    collection.insert_many(test_data)
    collection.create_index('id')
    
    def query_op():
        result = collection.find_one({'id': 5000})
        return result
    
    results = benchmark_operation(query_op, 1000)
    print(f"Query performance: {results['operations_per_second']:.2f} ops/sec")

import mongomock
import unittest
from unittest.mock import patch

class IntegrationTestExample(unittest.TestCase):
    """Example integration test using mongomock."""
    
    def setUp(self):
        """Set up test environment."""
        self.client = mongomock.MongoClient()
        self.db = self.client.integration_test
        
    def test_full_user_workflow(self):
        """Test complete user management workflow."""
        users = self.db.users
        
        # Create user
        user_data = {
            'username': 'testuser',
            'email': 'test@example.com',
            'created_at': datetime.utcnow(),
            'status': 'active'
        }
        result = users.insert_one(user_data)
        user_id = result.inserted_id
        
        # Verify user creation
        created_user = users.find_one({'_id': user_id})
        self.assertEqual(created_user['username'], 'testuser')
        
        # Update user
        update_result = users.update_one(
            {'_id': user_id},
            {'$set': {'last_login': datetime.utcnow()}}
        )
        self.assertEqual(update_result.modified_count, 1)
        
        # Query with complex filter
        active_users = list(users.find({
            'status': 'active',
            'created_at': {'$gte': datetime.utcnow() - timedelta(days=1)}
        }))
        self.assertEqual(len(active_users), 1)
        
        # Aggregation
        stats = list(users.aggregate([
            {'$group': {
                '_id': '$status',
                'count': {'$sum': 1}
            }}
        ]))
        self.assertEqual(len(stats), 1)
        self.assertEqual(stats[0]['count'], 1)
        
        # Delete user
        delete_result = users.delete_one({'_id': user_id})
        self.assertEqual(delete_result.deleted_count, 1)
        
        # Verify deletion
        deleted_user = users.find_one({'_id': user_id})
        self.assertIsNone(deleted_user)

    @mongomock.patch()
    def test_with_real_pymongo_imports(self):
        """Test using real PyMongo imports with mongomock patch."""
        import pymongo
        
        # This actually creates a mongomock client
        client = pymongo.MongoClient()
        db = client.test_db
        
        # All operations use mongomock
        collection = db.test_collection
        result = collection.insert_one({'patched': True})
        
        found = collection.find_one({'patched': True})
        self.assertIsNotNone(found)
        self.assertTrue(found['patched'])

# Always use unique database names for tests
def get_test_db_name():
    return f"test_{uuid.uuid4().hex[:8]}"

# Clean up after tests
class CleanTestCase(unittest.TestCase):
    def setUp(self):
        self.client = mongomock.MongoClient()
        self.db_name = get_test_db_name()
        self.db = self.client[self.db_name]
        
    def tearDown(self):
        self.client.drop_database(self.db_name)

# Use realistic data structures and volumes
def create_realistic_test_data():
    return [
        {
            'user_id': str(ObjectId()),
            'username': f'user_{i}',
            'email': f'user{i}@example.com',
            'profile': {
                'age': random.randint(18, 80),
                'interests': random.sample(['sports', 'music', 'art', 'tech'], 2)
            },
            'created_at': datetime.utcnow() - timedelta(days=random.randint(1, 365))
        }
        for i in range(1000)
    ]