MigueL
MigueL

Reputation: 3

Pytest Mock Create Engine and execute query: Assertion Error

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

Answers (0)

Related Questions