python 模拟 sqlalchemy 连接

Question

我有一个简单的函数，可以连接到数据库并获取一些数据。

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

当我运行上述文件 pythontests.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 检查。

---

Answer 1

另一种选择是模拟你的_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)