How to mock mongodb when it is called from another function?

Question

I need help while mocking mongodb. I am using mongomock to mock mongodb.

My project structure is:

-- my_mongo.py
-- code.py
-- my_test.py

my_mongo.py has :

from pymongo import MongoClient

 def get_db():
   client = MongoClient(os.environ['MONGODB_URI'])
   db = client['my_db']
   return db

 def insert(id, data):
     return get_db().results.insert_one(
          {
          "id": id,
          "data":df.to_json(),
          }).id

and code.py has

import my_mongo

def action():
    #do somethings
    my_mongo.insert(id, data)

and my_test.py has

import mongomock
import my_mongo
from unittest import mock


with patch.object(my_mongo.get_db().client, "client",  mongomock.MongoClient()):
    import code

def test_action_1():
    my_mongo.insert = mock.Mock(return_value=1)
    code.action()   

def test_action_2():
     with patch.object(my_mongo.get_db(), "get_db", mongomock.MongoClient().db):
     code.action()

It throws pymongo.errors.ServerSelectionTimeoutError for both tests. So, It still goes into the insert_one() method in my_mongo.py. I expect in test_action_1 my_mongo.insert returns 1, but it doesn't.

What am I missing?

Answer 1

I'm not entirely sure what mongomock is for, but it looks like it's for mocking an entire mongo database and not actually using python mocking. I'm going to answer without including mongomock since I don't think you really need it, so you can take that for what it's worth.

There were a few issues:

1. Calling patch.object will patch the given method on whatever object you give it. If you call get_db in the test, then code.action calls get_db, those are 2 different objects. Maybe this works? But I'm skeptical, so I just changed it.

2. Don't use code as your module name. That's already a module included with python.

3. code.action was missing args and a return statement.

You'll also notice that I changed how and what was being mocked to illustrate different ways to accomplish the mocking. Test 1 mocks the insert call with a function decorator. Test 2 mocks the get_db call with a contextmanager. Either is correct, just showing that you have options.

Here's the finished product:

---

my_mongo.py:

from pymongo import MongoClient

def get_db():
    client = MongoClient(os.environ['MONGODB_URI'])
    db = client['my_db']
    return db

def insert(id, data):
    return get_db().results.insert_one({"id": id, "data":data.to_json()}).id  # df was undefined, updated to data

---

my_code.py:

import my_mongo

# I added id and data args. They were undefined
def action(id, data):
    return my_mongo.insert(id, data)  # I added a return here

---

my_test.py

from unittest import mock

import my_code

# I removed the contextmanager import. Nothing is being evaluated here that would
# need to be patched, so I'm pretty certain it has no effect

@mock.patch('my_mongo.insert')
def test_action_1(mock_insert):
    expected_id = 1
    mock_insert.return_value = expected_id
    ret = my_code.action(expected_id, mock.Mock())
    assert ret == expected_id

def test_action_2():
    with mock.patch('my_mongo.get_db') as mock_get_db:
        expected_id = 'some id'
        mock_db = mock.Mock()
        mock_db.results.insert_one.return_value.id = expected_id
        mock_get_db.return_value = mock_db
        ret = my_code.action(expected_id, mock.Mock())
    assert ret == expected_id

Answer 2

That line of code for patching mongodb is wrong. Instead of using patch.object(my_mongo.get_db(), "get_db", mongomock.MongoClient().db), you should use patch.object("my_mongo.get_db", return_value=mongomock.MongoClient()['my_db']).

---

Following is the complete runable code for your example:

my_test.py

import mongomock
from unittest.mock import patch

import my_code
import my_mongo


def test_action_2():
    mocked_mongo = mongomock.MongoClient()
    with patch("my_mongo.get_db", return_value=mongomock.MongoClient()['my_db']):
        my_code.action()
        assert mocked_mongo.my_db.results.count_documents({'id': 'some_id'}) == 1

my_mongo.py

from pymongo import MongoClient

def get_db():
    client = MongoClient(os.environ['MONGODB_URI'])
    db = client['my_db']
    return db

def insert(id, data):
    return get_db().results.insert_one(
        {
            "id": id,
            "data": data,
        })

my_code.py

import my_mongo

def action():
    #do somethings
    return my_mongo.insert('some_id', '{"a": 3}')