Reputation: 3
Trying to set unit test for this function that performs a Database connection and a query execution.
clean_database.py
def clean_database(self, list_files_names:list) -> None:
"""_summary_ : Clean the database by deleting rows of defined files
Args:
list_files_names (list): list of path .tif files to delete
"""
# connect to database
engine = create_engine(self.db_url)
# query to delete rows
for f_name in list_files_names:
query = f'''DELETE FROM {self.db_schema}."{self.pollutant.upper()}"
WHERE "location" LIKE '{f_name}';'''
# execute query
with engine.connect() as connection:
connection.execute(text(query))
engine.dispose()
return None
Below the test functions:
test_clean_database.py
# Initialize the class `DeleteMaps` that contains the method `clean_database()`:
delete_maps = DeleteMaps(
db_url,
db_schema,
host,
username,
password,
pollutant,
directory_data)
class TestCleanDatabase():
I add the fixture for create the engine and mock the connection response:
@pytest.fixture
def mock_engine(self, mocker):
mock_engine = mocker.patch('sqlalchemy.create_engine')
mock_connection = mocker.Mock()
mock_engine.return_value.connect.return_value.__enter__.return_value = mock_connection
return mock_engine, mock_connection
I mock then the configuration database :
@pytest.fixture
def mock_config(self):
return {
"db_url": "postresql:///:memory:",
"db_schema": "test_schema",
"db_table_pollutant": "NO2"
}
I write the test function, using the mock engine and the mock connection functions as arguments:
def test_clean_database_with2files(self, mock_engine: tuple, mock_config: dict):
# Create a DatabaseCleaner instance
db_url = mock_config['db_url']
db_schema = mock_config['db_schema']
pollutant = mock_config['db_table_pollutant']
mock_engine, mock_connection = mock_engine
# Mock list_files_names
list_files_names = [f'{pollutant}_20240724T070000.tif', f'{pollutant}_20240724T080000.tif']
# Mock the result of the query execution
mock_result = MagicMock()
mock_result.rowcount = 2
mock_connection.execute.return_value = mock_result
# Call the method
delete_maps.clean_database(list_files_names)
# Execute queries
for file_name in list_files_names:
expected_query = f'''DELETE FROM {db_schema}."{pollutant}"
WHERE "location" LIKE '{file_name}';'''
# Check if the query was executed
mock_connection.execute.assert_any_call(text(expected_query))
# Check if the engine was disposed of
mock_engine.return_value.dispose.assert_called_once()
When debugging my code it breaks at the line:
# Check if the query was executed
mock_connection.execute.assert_any_call(text(expected_query))
Error :
AssertionError: execute(<sqlalchemy.sql.elements.TextClause object at 0x7f13343b3290>) call not found
horair_poetry/tests/test_delete_maps.py:290: AssertionError
Upvotes: 0
Views: 40