Как протестировать динамически изменяющиеся параметры в sql-запросе к базе данных с помощью pytest

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

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

    start_date = (
        date.today() - timedelta(days=get_redshift_query_param_value('start_date'))
    ).strftime('%Y-%m-%d')
    end_date = (date.today() - timedelta(days=1)).strftime('%Y-%m-%d')

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

    with RedshiftCursor() as cursor:
        cursor.execute(
            """
            SELECT ad_group, spends.campaign_id, ad_spend
                FROM (
                    SELECT
                        ...
                        SUM(spend_eur) AS ad_spend
                    FROM %(src_table)s
                    WHERE
                        ......
                    ) as spends
                    LEFT JOIN
                    (SELECT ... AS total_spend
                    FROM %(spend_src_table)s
                     GROUP BY campaign_id, spend_eur
                    ) AS sg
                    ON ...
            WHERE ad_spend * 100 / %(cutoff_percent)s < total_spend
            """,
            {
                'campaign_ids': tuple(campaign_ids),
                'start_date': start_date,
                'end_date': end_date,
                'cutoff_percent': get_redshift_query_param_value('cutoff_percent'),
                'src_table': AsIs(src_table),
                'spend_src_table': AsIs(spend_src_table),
            },
        )
        for row in cursor.fetchall():
            loses_adgroups.append((row[0], str(row[1])))

    return loses_adgroups

В нем вызывается функция get_redshift_query_param_value

def get_redshift_query_param_value(param: str) -> int:
    params = {
        'cutoff_percent': RedshiftQuery.CUTOFF_PERCENT,
        'min_place': RedshiftQuery.MIN_PLACE,
        'start_date': RedshiftQuery.START_DATE,
    }
    return int(RedshiftQueryParam.objects.get(name=params[param]).value)

, который извлекает числовое значение из баз данных на основе переданного ключа. Эти значения изменяются в панели django-admin и записываются в базу.

Написал тест, который проверяет доступ к базе данных и возвращаемое значение

test.py

import pytest

from google_panel.logic.clean_creative 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,
        get_redshift_query_param_value_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 скрипт тестирование с различными start_date и cutoff_percent значениями

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