I am trying to mock the fetchall()
from dbconnection
cursor object. I am trying the following code with expected return value. However, it was not returning the value. I have the answer now and edited the unit test to include the answer also
db.py
def query_db(db_connection, query):
"""
:param db_connection: dbconnection object
:param query: query to be executed
:return:
"""
cur = db_connection.cursor()
try:
logging.info(f"Query to be executed : {query}")
cur.execute(query)
results = cur.fetchall()
logging.info(f"Query Results : {results}")
except Exception:
logging.exception("Exception while executing \'query_db\' function")
raise Exception(f"Error while executing query : {query}. Please check the logs for details")
return results
Test Case:
def test_get_client_object(self):
dbconnection = Mock(name="dbconnection")
mycursor = Mock(name="mycursor")
mycursor.fetchall.return_value = "testing_return_value"
dbconnection.cursor.return_value = mycursor # I was doing dbconnection.cursor = mycursor . ... which caused the error
self.assertEqual("testing_return_value", utils.query_db(dbconnection, 12345))
I got the follwing assertion error. It returned a mock object instead of expected return value.
<Mock name='mycursor().fetchall()' id='4443879760'> != testing_return_value
Expected :testing_return_value
Actual :<Mock name='mycursor().fetchall()' id='4443879760'>
<Click to see difference>
Here is the unit test solution:
utils.py
:
def query_db(db_connection, query):
cur = db_connection.cursor()
try:
print(f"Query to be executed : {query}")
cur.execute(query)
results = cur.fetchall()
print(f"Query Results : {results}")
except Exception:
print("Exception while executing \'query_db\' function")
raise Exception(
f"Error while executing query : {query}. Please check the logs for details")
return results
test_utils.py
:
import unittest
from unittest.mock import Mock
import utils
class TestUtils(unittest.TestCase):
def test_get_client_object(self):
dbconnection = Mock(name="dbconnection")
mycursor = Mock(name="mycursor")
mycursor.fetchall.return_value = "testing_return_value"
dbconnection.cursor.return_value = mycursor
self.assertEqual("testing_return_value",
utils.query_db(dbconnection, 12345))
dbconnection.cursor.assert_called_once()
mycursor.execute.assert_called_once_with(12345)
mycursor.fetchall.assert_called_once()
def test_query_db_exception(self):
dbconnection = Mock(name="dbconnection")
mycursor = Mock(name="mycursor")
mycursor.fetchall.side_effect = Exception
dbconnection.cursor.return_value = mycursor
with self.assertRaises(Exception) as cm:
utils.query_db(dbconnection, 12345)
self.assertEqual(str(
cm.exception), 'Error while executing query : 12345. Please check the logs for details')
if __name__ == '__main__':
unittest.main(verbosity=2)
Unit test result with 100 coverage report:
test_get_client_object (__main__.TestUtils) ... Query to be executed : 12345
Query Results : testing_return_value
ok
test_query_db_exception (__main__.TestUtils) ... Query to be executed : 12345
Exception while executing 'query_db' function
ok
----------------------------------------------------------------------
Ran 2 tests in 0.002s
OK
Name Stmts Miss Cover Missing
------------------------------------------------------------------------
src/stackoverflow/59226762/test_utils.py 22 0 100%
src/stackoverflow/59226762/utils.py 11 0 100%
------------------------------------------------------------------------
TOTAL 33 0 100%
Source code: https://github.com/mrdulin/python-codelab/tree/master/src/stackoverflow/59226762