首页 > 编程问答 >关闭 contextlib 模块时如何模拟 psycopg2.connect?

关闭 contextlib 模块时如何模拟 psycopg2.connect?

时间:2024-07-21 03:33:27浏览次数:15  
标签:python psycopg2 python-unittest python-3.8

我有以下带有 get_data 方法的数据库类。它使用“contextlib”模块的“关闭”来关闭连接。

class Database:

    def __init__(self) -> None:
        self.db_details = {<connection details>}

    def get_data(self, query, parameters):

        with closing(psycopg2.connect(**self.db_details)) as conn:
            with conn.cursor() as cursor:
                cursor.execute(query, parameters)
                data = cursor.fetchall()

        return data

通过引用 这个 解决方案,我想出了以下单元测试

class TestDatabase(unittest.TestCase):

    @patch("psycopg2.connect")
    @patch("Database.__init__")
    def test_get_data_pos(self, mock_init, mock_connect):

        expected_result = [(1, "John", "New York City")]
        query = "SELECT * FROM user_details;"
        parameters = {}

        mock_init.return_value = None

        db_inst = Database()
        db_inst.db_details = {
            <some_dummy_values>
        }

        mock_con_cm = mock_connect.return_value
        mock_conn = mock_con_cm.__enter__.return_value
        mock_cur = mock_conn.cursor.return_value.__enter__.return_value
        mock_cur.fetchall.return_value = expected_result

        actual_result = db_inst.get_data(query, parameters)

        self.assertEqual(expected_result, actual_result)


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

但我收到以下断言错误:

AssertionError: [(1, 'John', 'New York City')] != <MagicMock name='connect().cursor().__ent[35 chars]384'>

注意 :我无法修改数据库类的实现。


你的问题出在 mock 对象的嵌套方式上。在 get_data 方法中,你使用了两个上下文管理器:一个是 closing(psycopg2.connect(**self.db_details)) ,另一个是 conn.cursor() 。这意味着你需要对你的 mock 对象进行相应的嵌套,以便正确地模拟这些调用。

下面是修正后的测试代码:

class TestDatabase(unittest.TestCase):

    @patch("psycopg2.connect")
    @patch("Database.__init__")
    def test_get_data_pos(self, mock_init, mock_connect):

        expected_result = [(1, "John", "New York City")]
        query = "SELECT * FROM user_details;"
        parameters = {}

        mock_init.return_value = None

        db_inst = Database()
        db_inst.db_details = {
            # some_dummy_values
        }

        # 正确嵌套 mock 对象
        mock_conn = mock_connect.return_value
        mock_cursor = mock_conn.cursor.return_value
        mock_cursor.__enter__.return_value.fetchall.return_value = expected_result

        actual_result = db_inst.get_data(query, parameters)

        self.assertEqual(expected_result, actual_result)

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

在这个修正后的代码中:

  1. 我们首先创建了 mock_conn 对象,它是 mock_connect.return_value ,表示模拟的连接对象。
  2. 然后我们创建了 mock_cursor 对象,它是 mock_conn.cursor.return_value ,表示模拟的游标对象。
  3. 最后,我们将 expected_result 设置为 mock_cursor.__enter__.return_value.fetchall.return_value 。这是因为 conn.cursor() 返回一个上下文管理器,我们需要调用 __enter__ 方法获取游标对象,然后才能调用 fetchall 方法。

通过这种方式正确地嵌套 mock 对象,你就可以成功地模拟 psycopg2.connect conn.cursor() 的调用,并验证你的 get_data 方法的行为。

标签:python,psycopg2,python-unittest,python-3.8
From: 78570759

相关文章

  • Python应用程序跨子包共享的配置文件
    我正在构建一个应用程序来控制一些硬件。我在包中实现了不同类型的硬件:电机和测量设备。我的文件结构如下:name_of_my_app/__init__.pymain.pyconfig.iniCONFIG.pymotors/__init__.pyone_kind_of_motor.pymeasurement_devices/......
  • python中时间序列数据的梯度计算
    我正在尝试编写一个函数,它可以从最适合下面的线返回梯度dataframe在浏览了谷歌的几个资源之后,我仍然不确定这是如何完成的。我明白最佳拟合线的计算公式为:y=mx+b将因变量(y)设置为foos,将自变量(x)设置为DateTimeDatafram......
  • 调试用 C 编写的 Python 扩展
    我非常熟悉编写C代码,并且很擅长编写Python代码。我正在尝试学习如何用C编写可以从OSX10.15.7上的Python-3.9.X调用的模块。我已经得到了几个“helloworld”类型的示例,但是对于复杂的示例,我正在努力弄清楚如何调试我编写的C扩展。MWE:src/add.c//......
  • 具有块大小选项的 Python pandas read_sas 因索引不匹配而失败并出现值错误
    我有一个非常大的SAS文件,无法容纳在我的服务器内存中。我只需要转换为镶木地板格式的文件。为此,我使用pandas中chunksize方法的read_sas选项分块读取它。它主要是在工作/做它的工作。除此之外,一段时间后它会失败并出现以下错误。此特定SAS文件有794......
  • 使用 requests 包 python 时打开文件太多
    我正在使用Pythonrequests包向API发出大量请求。然而,在某些时候,我的程序由于“打开的文件太多”而崩溃。当我明确关闭我的会话时,我真的不知道这是怎么回事。我使用以下代码:importrequestsimportmultiprocessingimportnumpyasnps=requests.session()s.keep......
  • Python 是一种选择性解释语言吗?为什么下面的代码不起作用?
    由于程序是从上到下运行的,为什么下面的代码不执行块中的第一行就直接抛出错误?if5>2:print("TwoislessthanFive!")print("Fiveisgreaterthantwo!")错误:文件“/Users/____/Desktop/Pythonpractise/practise.py”,第3行print("五比二大!")Indentati......
  • 裁剪时间变量 Python Matplotlib Xarray
    我不确定这是否是一个愚蠢的问题,但我想按时间变量剪辑.nc文件。我在xarray中打开了数据集,但以下ds.sel行(之前已运行)仅返回错误。ds=xr.open_dataset('/Users/mia/Desktop/RMP/data/tracking/mcs_tracks_2015_11.nc')selected_days=ds.sel(time=slice('2015-11-22',......
  • 用于匹配两个数据列表中的项目的高效数据结构 - python
    我有两个列表,其中一个列表填充ID,另一个列表填充进程名称。多个进程名称可以共享一个ID。我希望能够创建一个可以使用特定ID的数据结构,然后返回与该ID关联的进程列表。我还希望能够使用特定的进程名称并返回与其连接的ID列表。我知道我可以为此创建一个字典,但是I......
  • 有人可以解决我的代码中的问题吗?而且我无法在我的电脑上安装 nsetools。如何在 python
    从nsetools导入Nseimportpandasaspdnse=Nse()all_stock_codes=nse.get_stock_codes()companies_with_low_pe=[]对于all_stock_codes中的代码:如果代码=='符号':继续尝试:stock_quote=nse.get_quote(代码)pe_ratio=stock_quote.get('priceT......
  • 将 python 脚本的 stdin 重定向到 fifo 会导致 RuntimeError: input():lost sys.stdin
    我有这个python脚本,它的作用是充当服务器,它从重定向到fifo的stdin读取命令:test.py:whileTrue:try:line=input()exceptEOFError:breakprint(f'Received:{line}')在bash中运行命令:mkfifotestfifotest.py<testfifo......