Reputation: 21
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