使用python模拟dbconnection

Question

我试图从fetchall()游标对象中模拟dbconnection。我正在尝试以下具有预期返回值的代码。但是，它没有返回值。我现在有了答案，并编辑了单元测试，以包括答案也包括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))

我得到了后续断言错误。它返回一个模拟对象，而不是预期的返回值。

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

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

Answer 1

以下是单元测试解决方案：

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

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

Answer 2

接受的答案对我不起作用，而且我仍然在被链式的调用-- 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()