Python单元测试：搭建模拟数据库

Question

我想建立一个模拟数据库（与创建测试数据库相反，如果可能的话）来检查数据是否被正确地查询并转换为Pandas数据框架。我有一些关于模拟和单元测试的经验，并且以前成功地设置了先前的测试。然而，我在应用如何模拟像数据库这样的真实对象进行测试方面遇到了困难。

目前，当我的测试运行时，我遇到了生成结果的困难。我认为我没有正确地模拟数据库对象，我错过了其中涉及的某个步骤或者我的思维过程是错误的。我将我的测试和要测试的代码放在同一个脚本中以简化事情。

• 我已经彻底阅读了Python unittest和mock文档，所以我知道它是做什么和如何工作的（在大多数情况下）。
• 我已经阅读了无数关于mocking的帖子，包括Stack内部和外部的帖子。它们有助于理解一般概念以及可以在所述特定情况下完成哪些工作，但我无法在我的情况下使其正常工作。
• 我尝试模拟函数的各个方面，包括数据库连接、查询和使用“pd_read_sql(query,con)”函数，但都没有成功。我认为这是我最接近目标的方式。

我的最新测试代码

import pandas as pd
import pyodbc
import unittest
import pandas.util.testing as tm

from unittest import mock

# Function that I want to test
def p2ctt_data_frame():
    conn = pyodbc.connect(
        r'Driver={Microsoft Access Driver (*.mdb, *.accdb)};'
        r'DBQ=My\Path\To\Actual\Database\Access Database.accdb;'
    )

    query = 'select * from P2CTT_2016_Plus0HHs'

    # I want to make sure this dataframe object is created as intended
    df = pd.read_sql(query, conn) 

    return df


class TestMockDatabase(unittest.TestCase):

    @mock.patch('directory1.script1.pyodbc.connect')  # Mocking connection
    def test_mock_database(self, mock_access_database):

        # The dataframe I expect as the output after query is run on the 'mock database'
        expected_result = pd.DataFrame({
            'POSTAL_CODE':[
                'A0A0A1'
            ],
            'DA_ID':[
                1001001
            ],
            'GHHDS_DA':[
                100
            ]
        })

        # This is the line that I believe is wrong. I want to create a return value that mocks an Access table
        mock_access_database.connect().return_value = [('POSTAL_CODE', 'DA_ID', 'GHHDS_DA'), ('A0A0A1', 1001001, 100)]

        result = p2ctt_data_frame()  # Run original function on the mock database

        tm.assert_frame_equal(result, expected_result) 


if __name__ == "__main__":
    unittest.main()

我期望使用模拟数据库对象运行测试后的预期数据框与结果相同，但实际情况并非如此。

目前，当我尝试模拟数据库时，如果打印出结果，我会得到：

空的数据框 列: [] 索引: []

此外，在测试运行后，我会收到以下错误：

断言错误：数据框不同；
数据框形状不匹配
【左侧】：（0, 0）
【右侧】：（1, 3）

- ShockDoctor

---

为什么不查询数据库，而只使用前三个值？在进行准确性测试时，请确保列名正确，以及您正在查找的数据框的一些其他方面，例如邮政编码长度为6个字符。 - Daniel Butler

嗨，感谢您的回复！您认为这是测试数据库的最佳方法吗？我对单元测试和集成测试仍然有些疑虑，不确定它们是否应该进行类似的测试。 - ShockDoctor

我发现最好两者兼顾。这本书教会了我很多关于测试以及如何正确进行测试的知识：https://www.obeythetestinggoat.com/book/praise.harry.html - Daniel Butler

会仔细阅读，再次感谢！ - ShockDoctor

Answer 1

我建议将它分成几个单独的测试。一个功能测试，以确保能够生成所需结果；一个测试以确保可以访问数据库并获得预期结果；最后是如何实现它的单元测试。我会按照这个顺序编写每个测试，在实际函数之前先完成测试。如果发现无法解决某些问题，我会在单独的REPL上尝试或创建一个git分支来处理它，然后返回到主分支。有关更多信息，请参见https://obeythetestinggoat.com/book/praise.harry.html。每个测试都有注释和原因。

import pandas as pd
import pyodbc

def p2ctt_data_frame(query='SELECT * FROM P2CTT_2016_Plus0HHs;'): # set query as default
    with  pyodbc.connect(
        r'Driver={Microsoft Access Driver (*.mdb, *.accdb)};'
        r'DBQ=My\Path\To\Actual\Database\Access Database.accdb;'
    ) as conn:  # use with so the connection is closed once completed

        df = pd.read_sql(query, conn)

    return df

单独的测试文件：

import pandas as pd
import pyodbc
import unittest
from unittest import mock

class TestMockDatabase(unittest.TestCase):

    def test_p2ctt_data_frame_functional_test(self):  # Functional test on data I know will not change
        actual_df = p2ctt_data_frame(query='SELECT * FROM P2CTT_2016_Plus0HHs WHERE DA_ID = 1001001;')

        expected_df = pd.DataFrame({
            'POSTAL_CODE':[
                'A0A0A1'
            ],
            'DA_ID':[
                1001001
            ],
            'GHHDS_DA':[
                100
            ]
        })

        self.assertTrue(actual_df == expected_df)

    def test_access_database_returns_values(self):  # integration test with the database to make sure it works
        with pyodbc.connect(
            r'Driver={Microsoft Access Driver (*.mdb, *.accdb)};'
            r'DBQ=My\Path\To\Actual\Database\Access Database.accdb;'
        ) as conn:
            with conn.cursor() as cursor:
                cursor.execute("SELECT TOP 1 * FROM P2CTT_2016_Plus0HHs WHERE DA_ID = 1001001;")
                result = cursor.fetchone()

        self.assertTrue(len(result) == 3)  # should be 3 columns by 1 row

        # Look for accuracy in the database
        info_from_db = []
        for data in result:  # add to the list all data in the database
            info_from_db.append(data)

        self.assertListEqual(   # All the information matches in the database
            ['A0A0A1', 1001001, 100], info_from_db
        )


    @mock.patch('directory1.script1.pd')  # testing pandas
    @mock.patch('directory1.script1.pyodbc.connect')  # Mocking connection so nothing sent to the outside
    def test_pandas_read_sql_called(self, mock_access_database, mock_pd):  # unittest for the implentation of the function
        p2ctt_data_frame()
        self.assert_True(mock_pd.called)  # Make sure that pandas has been called
        self.assertIn(
            mock.call('select * from P2CTT_2016_Plus0HHs'), mock_pd.mock_calls
        )  # This is to make sure the proper value is sent to pandas. We don't need to unittest that pandas handles the
        # information correctly.

*我没有测试过这个，所以可能有一些错误需要我修复

- Daniel Butler

---

哇，谢谢，那非常清晰和有用！不过我有一个问题。如果测试1通过了，那么测试3是不是就不必要了？我这么说的唯一原因是，如果预期的数据帧和实际的数据帧相等，那么不就意味着Pandas必须使用该特定查询吗？因此测试调用Pandas是多余的吗？ - ShockDoctor

在这种情况下，它有点冗余，因为我们是在考虑这些函数的独立性。但是假设您需要验证查询，因为它正在使用用户输入。我们将想要在p2ctt_data_frame函数中创建一个单独的validate_query来处理它。在测试p2ctt_data_frame时，我将在unittest中模拟validate_query，然后确保它被调用。完成后，创建一个单独的unittest，用于验证validate_query是否按预期运行。 - Daniel Butler