python - python模拟sqlalchemy连接
问题描述
我有一个连接到数据库并获取一些数据的简单函数。
db.py
from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool
def _create_engine(app):
impac_engine = create_engine(
app['DB'],
poolclass=NullPool # this setting enables NOT to use Pooling, preventing from timeout issues.
)
return impac_engine
def get_all_pos(app):
engine = _create_engine(app)
qry = """SELECT DISTINCT id, name FROM p_t ORDER BY name ASC"""
try:
cursor = engine.execute(qry)
rows = cursor.fetchall()
return rows
except Exception as re:
raise re
我试图通过模拟这个连接来编写一些测试用例 -
测试.py
import unittest
from db import get_all_pos
from unittest.mock import patch
from unittest.mock import Mock
class TestPosition(unittest.TestCase):
@patch('db.sqlalchemy')
def test_get_all_pos(self, mock_sqlalchemy):
mock_sqlalchemy.create_engine = Mock()
get_all_pos({'DB': 'test'})
if __name__ == '__main__':
unittest.main()
当我运行上述文件 python tests.py 时,我收到以下错误 -
"Could not parse rfc1738 URL from string '%s'" % name
sqlalchemy.exc.ArgumentError: Could not parse rfc1738 URL from string 'test'
不应该 mock_sqlalchemy.create_engine = Mock() 给我一个模拟对象并绕过 URL 检查。
解决方案
另一种选择是模拟您的_create_engine
功能。由于这是一个单元测试,我们想要测试get_all_pos
我们不应该依赖 的行为_create_engine
,所以我们可以像这样修补它。
import unittest
import db
from unittest.mock import patch
class TestPosition(unittest.TestCase):
@patch.object(db, '_create_engine')
def test_get_all_pos(self, mock_sqlalchemy):
args = {'DB': 'test'}
db.get_all_pos(args)
mock_sqlalchemy.assert_called_once()
mock_sqlalchemy.assert_called_with({'DB': 'test'})
if __name__ == '__main__':
unittest.main()
如果要测试某些结果,则需要正确设置所有相应的属性。我建议不要将它链接到一个调用中,以便它更具可读性,如下所示。
import unittest
import db
from unittest.mock import patch
from unittest.mock import Mock
class Cursor:
def __init__(self, vals):
self.vals = vals
def fetchall(self):
return self.vals
class TestPosition(unittest.TestCase):
@patch.object(db, '_create_engine')
def test_get_all_pos(self, mock_sqlalchemy):
to_test = [1, 2, 3]
mock_cursor = Mock()
cursor_attrs = {'fetchall.return_value': to_test}
mock_cursor.configure_mock(**cursor_attrs)
mock_execute = Mock()
engine_attrs = {'execute.return_value': mock_cursor}
mock_execute.configure_mock(**engine_attrs)
mock_sqlalchemy.return_value = mock_execute
args = {'DB': 'test'}
rows = db.get_all_pos(args)
mock_sqlalchemy.assert_called_once()
mock_sqlalchemy.assert_called_with({'DB': 'test'})
self.assertEqual(to_test, rows)
推荐阅读
- javascript - 给定问题的替代和优化解决方案
- google-maps - 位置选择器给我空的位置(我标记为设置位置的位置坐标)
- eclipse - 如何在 Mac 上对 Java (JSF) XHTML 文件进行实时编辑?
- facebook - Facebook喜欢按钮点击
- css - 布尔玛汉堡菜单不显示项目
- php - 使用 php 将实体添加到数据存储区
- python - 列表理解,用于将列表中的每个字符串乘以给定范围内的数字
- sharepoint - 共享点域是什么意思
- swift - AVCaptureSession 不向 iPhone X 以下的 iPhone 提供数据
- python - Python - 将输入范围转换为序数字符