在 Python 中设置模拟数据库进行单元测试

Question

我想设置一个模拟数据库（如果可能的话，而不是创建一个测试数据库）来检查数据是否被正确查询而不是被转换为 Pandas 数据帧。我有一些模拟和单元测试的经验，并且成功地设置了以前的测试。但是，我在应用如何模拟现实生活中的对象（如数据库）进行测试时遇到了困难。

目前，我在运行测试时无法生成结果。我相信我没有正确地模拟数据库对象，我错过了一个涉及的步骤或者我的思考过程不正确。我把我的测试和我的代码放在同一个脚本中进行测试以简化事情。

• 我已经彻底阅读了 Python 单元测试和模拟文档，所以我知道它做了什么以及它是如何工作的（在大多数情况下）。
• 我已经阅读了无数关于 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()

我希望预期的数据帧和使用模拟数据库对象运行测试后的结果是一回事。不是这种情况。

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

空数据帧列：[] 索引：[]

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

断言错误：DataFrame 不同；
DataFrame 形状不匹配 [左]: (0, 0) [右]: (1, 3)

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.

*我无法对此进行测试，因此可能需要修复一些错误