Jake Andrews
Jake Andrews

Reputation: 21

Mocking psycopg2 connection to test if data was inserted

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)

The code above is attempting to model the psycopg2 connection/cursor so that another function (in which the insert-to-database method) can be tested, independently of the connection dependencies.

I am really struggling since I am new to patching/mocking.

Currently my code fails with:

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

The additional function I am calling is seen in the first code snippet in the lines:

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

This section calls the following:

    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)

I really need some guidance on what to do to make a test of psycopg2 progress to something useful, either through feedback on my set-up or with ideas on changes that could be made to the above code snippets.

Upvotes: 0

Views: 810

Answers (0)

Related Questions