Python单元测试:如何对包含数据库操作的模块进行单元测试?

17

我正在使用pymysql客户端库连接真实数据库。我在一个模块中有一个函数,在其中使用pymysql连接到数据库并仅执行数据库插入操作。如何在Python中对此函数进行单元测试,而不会触发真实数据库?

import pymysql

def connectDB(self):

# Connect to the database
connection = pymysql.connect(host='localhost',
                             user='user',
                             password='passwd',
                             db='db')

try:
    with connection.cursor() as cursor:
        # Create a new record
        sql = "INSERT INTO `users` (`email`, `password`) VALUES (%s, %s)"
        cursor.execute(sql, ('newuser@some.com', 'newpassword'))


    connection.commit()

我的 Python 版本是 2.7。


3
模拟数据库是一个选择https://docs.python.org/3/library/unittest.mock.html - mbieren
快速备注:也许 connectDB 不是一个同时执行查询的方法的最佳名称 :) - Eduardo
3个回答

17

您可以像这样使用patch

from unittest.mock import patch, MagicMock

@patch('mypackage.mymodule.pymysql')
def test(self, mock_sql):
    self.assertIs(mypackage.mymodule.pymysql, mock_sql)

    conn = Mock()
    mock_sql.connect.return_value = conn

    cursor      = MagicMock()
    mock_result = MagicMock()

    cursor.__enter__.return_value = mock_result
    cursor.__exit___              = MagicMock()

    conn.cursor.return_value = cursor

    connectDB()

    mock_sql.connect.assert_called_with(host='localhost',
                                        user='user',
                                        password='passwd',
                                        db='db')

    mock_result.execute.assert_called_with("sql request", ("user", "pass"))

我的Python版本是2.7。"unittest.mock"从Python 3.3开始可用。 - user2301
你可以在Python 2.7中使用Mock包:https://pypi.python.org/pypi/mock - uwevil
“mock_sql”是什么?代码中缺少更多的上下文信息。 - Banik
1
mock_sql 是您需要测试的包,其中包含您的 SQL 驱动程序。 - uwevil

1
您需要一系列名为存根(stubs)的虚假数据库,它们返回硬编码值。在测试期间,这些存根将代替真实数据库使用。我不熟悉Python,但在C++中可以通过使对象接收数据库作为构造函数参数来完成此操作。在生产代码中,您使用真实数据库参数,在测试中使用存根。这是因为构造函数期望一个指向公共基类的指针。即使它没有为Python编写,我建议阅读Roy Osherove的《单元测试的艺术》的前几章。该书清楚地解释了为什么这些虚假数据库是存根而不是模拟。

0
你刚刚重新发现了测试的重要性之一:它告诉你设计有问题时。换句话说,可测试性是衡量质量的一个很好的一级代理。考虑以下内容:
class DB(object):
    def __init__(self, **credentials):
        self._connect = partial(pymysql.connect, **credentials)

    def query(self, q_str, params):
        with self._connect as conn:
            with conn.cursor() as cur:
                cur.execute(q_str, params)
                return cur.fetchall()

# now for usage

test_credentials = {
    # use credentials to a fake database
}

test_db = DB(**test_credentials)
test_db.query(write_query, list_of_fake_params)
results = test_db.query(read_query)
assert results = what_the_results_should_be

如果您需要使用多个数据库,您可以使用多态性或根据API相似性将特定的数据库作为构造函数参数传递给对象。

网页内容由stack overflow 提供, 点击上面的
可以查看英文原文,
原文链接