Patching PyMongo Client

If your application code instantiates pymongo.MongoClient internally rather than accepting it as an injected dependency, you can intercept these instantiations using Mongomock's runtime patching framework.

How @mongomock.patch Works

The @mongomock.patch utility dynamically swaps the real pymongo.MongoClient class definition with a virtual client. This interceptor routes database calls to Mongomock's in-memory engine while preserving connection string configuration routing.

import pymongo
import mongomock

# Application Function
def fetch_database_status(connection_string):
    # Connection is established internally
    client = pymongo.MongoClient(connection_string)
    return client.admin.command("ping")

# Test Definition
@mongomock.patch(servers=(('cluster.mongo.net', 27017),))
def test_internal_client_initialization():
    # The real MongoClient is mocked during test execution
    response = fetch_database_status("mongodb://cluster.mongo.net:27017")
    assert response.get("ok") == 1.0

Server Routing and the on_new Policy

You can configure how Mongomock behaves when your application attempts to connect to a server address that was not defined in the servers parameter of @mongomock.patch:

@mongomock.patch(servers=(('localhost', 27017),), on_new='error')

on_new Policy Reference

• 'error' (Default): Raises a ValueError immediately. This is the strictest setting and is useful for catching misspelled hostnames or missing test configuration settings.
• 'create': Dynamically spins up a new virtual in-memory server state. This is useful for systems with dynamically generated connection strings.

• 'timeout': Simulates network partitions or offline database servers. The client will block for the duration of the timeout before raising a ServerSelectionTimeoutError:

  import pymongo
  import mongomock
  from unittest import mock
  import time
  
  @mongomock.patch(on_new='timeout')
  @mock.patch('time.sleep') # Mock sleep to prevent test lag
  def test_database_unreachable(mock_sleep):
      with pytest.raises(pymongo.errors.ServerSelectionTimeoutError):
          pymongo.MongoClient("mongodb://non-existent-host:27017")

• 'pymongo': Redirects connection requests to a real running MongoDB database via PyMongo. This is useful for hybrid integration tests where some services must hit a physical instance.

Decorator vs. Context Manager Syntax

You can apply patching using decorators or as inline context managers within individual test runs:

Using the Decorator

@mongomock.patch(servers=(('test-db', 27017),))
def test_with_decorator():
    client = pymongo.MongoClient("test-db")
    client.db.coll.insert_one({"key": "value"})

Using the Context Manager

def test_with_context_manager():
    # PyMongo operates normally outside of this block
    with mongomock.patch(servers=(('test-db', 27017),)):
        client = pymongo.MongoClient("test-db")
        client.db.coll.insert_one({"key": "value"})

Common Patching Pitfalls

Import Binding issues

Ensure your application code instantiates the database client via pymongo.MongoClient rather than binding it to a local namespace.

• Incorrect (Will NOT be patched correctly):

  # my_module.py
  from pymongo import MongoClient
  
  # The local binding 'MongoClient' is resolved before the patcher runs
  client = MongoClient("mongodb://test-db")

• Correct (Will be patched correctly):

  # my_module.py
  import pymongo
  
  # Resolved dynamically at instantiation
  client = pymongo.MongoClient("mongodb://test-db")

Next, explore Mongomock's support for aggregation pipelines and query operators in the Aggregation Guide.