I've been unable to mock the psycopg2 database connection and cursor since I've refactored it to use a context manager for the cursor. I'm aware that with a context manager, there are additional magic methods that are called for resource setup and cleanup (__enter__, __exit__), but even adding this into the mix has not narrowed down the problem.
Here is the code:
import os
import psycopg2
import psycopg2.extras
DB_HOST = os.getenv('DB_HOST')
DB_PORT = os.getenv('DB_PORT')
DB_NAME = os.getenv('DB_NAME')
DB_USER = os.getenv('DB_USER')
DB_PASSWORD = os.getenv('DB_PASSWORD')
CONN = psycopg2.connect(f'dbname={DB_NAME} user={DB_USER} host={DB_HOST} port={DB_PORT} password={DB_PASSWORD}')
def my_func():
message = None
print(CONN) # Added to debug - this never prints out a magic mock reference
select_sql = 'SELECT * FROM app.users WHERE name = %s LIMIT 1;'
with CONN.cursor(cursor_factory = psycopg2.extras.DictCursor) as cursor:
print(cursor) # Debug - no magic mock reference
cursor.execute(select_sql, ("Bob"))
row = cursor.fetchone()
if row is not None:
message = "name found"
else:
message = "name not found"
return message
Here's the test code and my attempt to mock the connection and cursor
import pytest
from src import my_class
from unittest import mock
class TestFunction:
@mock.patch('psycopg2.connect')
def test_my_func(self, mock_connect):
mock_cursor = mock.MagicMock()
mock_cursor.__enter__.return_value.fetchone.return_value = {
"id": 1,
"name": "Bob",
"age": 25
}
mock_connect.return_value.cursor.return_value = mock_cursor
result = my_class.my_func()
assert result == "found"
I'm not sure how to mock the connection or the cursor if I'm calling cursor on the global connection, CONN. Currently running pytest shows the test fails and the print statements show that the underlying object is not a magic mock. How does one mock a global db connection and a db cursor using a context manager? Any help is appreciated.