使用 python 模拟 dbconnection fetchall

Mocking dbconnection fetchall using python

我正在尝试从 dbconnection 游标对象模拟 fetchall()。我正在尝试使用预期 return 值的以下代码。但是,它不是 returning 值。我现在有了答案并编辑了单元测试以包含答案 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

测试用例:

    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))

我得到以下断言错误。它 return 模拟对象而不是预期的 return 值。

<Mock name='mycursor().fetchall()' id='4443879760'> != testing_return_value

Expected :testing_return_value
Actual   :<Mock name='mycursor().fetchall()' id='4443879760'>
<Click to see difference>

单元测试解决方案如下:

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)

包含 100 个覆盖率报告的单元测试结果:

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/Whosebug/59226762/test_utils.py      22      0   100%
src/Whosebug/59226762/utils.py           11      0   100%
------------------------------------------------------------------------
TOTAL                                         33      0   100%

源代码:https://github.com/mrdulin/python-codelab/tree/master/src/Whosebug/59226762

接受的答案对我不起作用,我仍然收到对 fetchall() 的链接调用。我也模拟了所有三个级别,包括 fetchall,然后它起作用了,也许它对某人有帮助:

测试:

expected = ["{\"parameter\":\"1337\"}"]
myconnection = mocker.Mock(name="dbconnection")
mycursor = mocker.Mock(name="mycursor")
myfetchall = mocker.Mock(name="myfetchall")  # I needed to mock this as well
myfetchall.fetchall.return_value = expected
mycursor.execute.return_value = myfetchall
myconnection.cursor.return_value = mycursor

函数:

cursor = connection.cursor()
return cursor.execute(sqlstatement).fetchall()