python-3.x - 使用python模拟dbconnection fetchall
问题描述
我正在尝试模拟fetchall()
fromdbconnection
光标对象。我正在尝试使用预期返回值的以下代码。但是,它没有返回值。我现在有了答案并编辑了单元测试以包含答案 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>
解决方案
这是单元测试解决方案:
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
推荐阅读
- python - 将 python 正则表达式模式转换为 lua
- java - Java 特殊符号日志记录后的“JVM 控制台输出中的无效多字节序列:无效或不完整的多字节或宽字符”
- azure - 在 azure devops 管道中执行 gulp 脚本
- javascript - 实现函数并返回新的列表js
- c++ - 如何解决:else 语句未被调用 c++
- for-loop - 为什么我的 `for-select` 语句在我关闭频道后仍不断收到来自我的频道的信息?
- python - Mongo GridFS 发现投影问题
- soot - 如何在 jimple (soot) 中保留原始变量名?
- wordpress - 在wordpress上单击菜单项后如何隐藏菜单项
- json - 通过 Angular 从 IndexedDB 获取/读取 JSON 文件