Тестирование cursor.execute с вызовом sql-скрипта с помощью pytest

Функция для тестирования

def get_adgroups_not_taked_share(
    campaign_ids: List[str], src_table: str, spend_src_table: str
) -> List[Tuple[str, str]]:


    loses_adgroups: List[Tuple[str, str]] = []

    with RedshiftCursor() as cursor:
        cursor.execute(
            """
            SELET some_data from some_table WHERE some_condition
            """
        )
        for row in cursor.fetchall():
            loses_adgroups.append((row[0], str(row[1])))

    return loses_adgroups

Существует тест для этой функции

import pytest

from my_ap import get_adgroups_not_taked_share


@pytest.fixture
def campaigns_redshift_cursor_mock(mocker):
    cursor_mock = mocker.MagicMock()
    cursor_mock.fetchall.return_value = [
        ('hs_video544', '123123123', 100),
        ('hs_video547', '123123123', 50),
    ]

    rs_cursor_creator = mocker.patch('google_panel.logic.clean_creative.RedshiftCursor')
    rs_cursor_creator.return_value.__enter__.return_value = cursor_mock
    return rs_cursor_creator


@pytest.mark.django_db
def test_get_adgroups_not_taked_share(
        campaigns_redshift_cursor_mock,
        ):
    campaign_ids = ['1111', '2222', '3333']
    result = get_adgroups_not_taked_share(campaign_ids, 'test_table', 'spend_src_table')
    assert result == [('hs_video544', '123123123'), ('hs_video547', '123123123')]

Теперь я хочу добавить новую функцию для проверки sql скрипта. Проверяя, что вызывается правильный sql-запрос. Что-то вроде

def test_get_adgroups_not_taked_share(
            campaigns_redshift_cursor_mock,
            ):
    ......
    query = """SELET some_data from some_table WHERE some_condition"""
    campaigns_redshift_cursor_mock.execute.assert_called_with(query)

Но получил

E       AssertionError: Expected call: execute('query')
E       Not called

Короткий ответ заключается в том, что вам нужно утверждать следующее: campaigns_redshift_cursor_mock.return_value.__enter__.return_value.execute.assert_called_once_with(query). Причина в том, что вы используете RedshiftCursor в качестве менеджера контекста (отсюда часть return_value.__enter__.return_value) и только потом вызываете метод execute.

Немного более длинный ответ - как я получил это утверждение.

Я написал эту библиотеку , которая добавляет приспособление mg pytest. Чтобы использовать ее, установите pip, а затем добавьте приспособление mg в ваш тест и выполните его generate_asserts метод следующим образом:

@pytest.mark.django_db
def test_get_adgroups_not_taked_share(
        campaigns_redshift_cursor_mock,
        mg
        ):
    campaign_ids = ['1111', '2222', '3333']
    result = get_adgroups_not_taked_share(campaign_ids, 'test_table', 'spend_src_table')
    mg.generate_asserts(campaigns_redshift_cursor_mock)
    assert result == [('hs_video544', '123123123'), ('hs_video547', '123123123')]

Затем запустите тест как обычно, и вы получите следующий вывод на консоль:

assert 1 == campaigns_redshift_cursor_mock.call_count
campaigns_redshift_cursor_mock.assert_called_once_with()
campaigns_redshift_cursor_mock.return_value.__enter__.assert_called_once_with()
campaigns_redshift_cursor_mock.return_value.__enter__.return_value.execute.assert_called_once_with('\n            SELET some_data from some_table WHERE some_condition\n            ')
campaigns_redshift_cursor_mock.return_value.__enter__.return_value.fetchall.assert_called_once_with()
campaigns_redshift_cursor_mock.return_value.__exit__.assert_called_once_with(None, None, None)

Вот все вызовы, сделанные к макету, вы можете отфильтровать те, которые имеют для вас значение.

Вернуться на верх