模拟 psycopg2 连接以测试是否插入了数据

问题描述

class Test_Postgres(unittest.TestCase):
    def setUp(self):
        #mock connection
        conx = Mock()
        #mock cursor
        cursor      = Magicmock()
        cursor.__enter__.return_value = Magicmock()
        cursor.__exit__ = Magicmock()
        #define cursor as output of connection cursor
        conx.cursor.return_value = cursor
        #add conx to set-up for use across tests
        self.conx = conx


    @patch('__main__.psycopg2')
    def testWriteStats(self,mock_postgres_sql):
        #Test that the mocked library matches the library itself
        self.assertIs(psycopg2,mock_postgres_sql)
        #set mock library connection as mock connection
        mock_postgres_sql.connect.return_value = self.conx

        print(mock_postgres_sql.connect.return_value)
        

        db = Statistics()
        db.insert(statistics_data)

        # self.assertTrue(mock_result.execute.called)
        self.assertTrue(mock_postgres_sql.connect.cursor().called)

上面的代码试图对 psycopg2 连接/游标进行​​建模，以便可以独立于连接依赖性测试另一个函数（其中插入到数据库方法）。

我真的很挣扎，因为我是修补/模拟的新手。

目前我的代码失败：

Traceback (most recent call last):
  File "c:\program files (x86)\python38-32\lib\unittest\mock.py",line 1342,in patched
    return func(*newargs,**newkeywargs)
  File "test.py",line 44,in testWriteStats
    self.assertTrue(mock_postgres_sql.connect.cursor().called)
AssertionError: False is not true

我正在调用的附加函数可以在行的第一个代码片段中看到：

...
        db = Statistics()
        db.insert(statistics_data)
...

本节调用以下内容：

    def insert(self,row):
        #magic variables
        field_names = row.keys() 
        number_of_fields = len(
            field_names
        )

        # Produces string in the form: `(Field_1,Field_2,Field_3 ...etc)` 
        specified_fields_string = self.list_to_query_string(
            field_names
        )

        #Produces string in the form: `(%s,%s,%s ... etc)` 
        values_placeholder_format = self.list_to_query_string(
            ["%s"] * number_of_fields
        )

        query_string = f"INSERT INTO {self.table_name} {specified_fields_string} VALUES {values_placeholder_format}"

        #Define the values to be used within the query string
        params = list(row.values()) 

        # Execute query
        connection = psycopg2.connect(**database_settings)
        cursor = connection.cursor()
        cursor.execute(query_string,params)

我真的需要一些指导来让 psycopg2 的测试取得一些有用的进展，无论是通过对我的设置的反馈，还是对可以对上述代码片段进行更改的想法。

解决方法

暂无找到可以解决该程序问题的有效方法，小编努力寻找整理中！

如果你已经找到好的解决方法，欢迎将解决方案带上本链接一起发送给小编。

小编邮箱:dio#foxmail.com (将#修改为@）