"""Unit tests for amazon_music specific Snowflake SQL executor."""

from unittest.mock import MagicMock
from unittest.mock import patch

import pytest
from snowflake import connector

from feed_ingestion.flows.amazon_music import config
from feed_ingestion.flows.amazon_music import snowflake_executor
from tests.conftest import SubstringMatcher


@pytest.fixture
def mock_executor(sf_config_mock, monkeypatch):
    """Yield executor context."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)
    executor = snowflake_executor.AmazonMusicSFExecutor(sf_config_mock)
    aws_params = {
        'aws_key_id': 'test_id',
        'aws_secret_key': 'test_secret',
        'aws_token': 'test_token'}
    with patch.object(executor, 'execute', wraps=executor.execute) as \
            executor.ex_mock, patch.object(executor, 'get_aws_params',
                                           return_value=aws_params):
        executor.fetchall = MagicMock()
        yield executor


@pytest.mark.parametrize(
    'report_type, date, sql_text',
    [
            ('activity', '2021-02-17', 'transctiontype'),
            # new field producttypekey for activity since 2021-02-18
            ('activity', '2021-02-18', 'producttypekey'),
            # new field voiceactivatedstream for activity since 2021-02-19
            ('activity', '2021-02-19', 'voiceactivatedstream'),
            ('user', '2021-02-18', 'numberofsmetracksincollection'),
            ('playlist', '2021-02-18', 'playlistfollowers'),
            # new fields totalplaycountforallvendors and countrycode
            # for activity since 2021-06-30
            ('activity', '2021-06-30', 'totalplaycountforallvendors'),
            ('activity', '2021-06-30', 'countrycode'),
    ]
)
def test_create_temp_staging_raw_table(mock_executor, report_type, date,
                                       sql_text):
    """Test create_temp_staging_raw_table method."""
    mock_executor.create_temp_staging_raw_table(
        temp_staging_raw_table='temp_table', report_type=report_type,
        date_for_sqlloader=date)

    mock_executor.ex_mock.assert_any_call(
        SubstringMatcher(
            containing=[
                'CREATE OR REPLACE TRANSIENT',
                'test_db.test_schema.temp_table',
                sql_text]),
        params={})

    assert mock_executor.ex_mock.call_count == 1


def test_load_temp_staging_raw_table(mock_executor):
    """Test load_temp_staging_raw_table method."""
    mock_executor.load_temp_staging_raw_table(
        temp_staging_raw_table='load_temp_staging_raw_table',
        aws={}, key_dir='s3://somepath', error_limit=1)

    assert mock_executor.fetchall.call_count == 1
    mock_executor.fetchall.assert_any_call(
        SubstringMatcher(
            containing=['COPY INTO', 'test_db.test_schema']),
        params={
            'aws_key_id': 'test_id',
            'aws_secret_key': 'test_secret',
            'aws_token': 'test_token',
            's3_path': 's3://somepath'})


@pytest.mark.parametrize(
    'value_in_kwagrs, value_in_sql',
    [
            ('true', 'TRUE'),
            ('false', 'FALSE'),
            ('FALSE', 'FALSE'),
            ('False', 'FALSE'),
            ('', 'TRUE'),
            ('anythingelse', 'TRUE'),
    ]
)
def test_load_temp_staging_raw_table_on_column_count_mismatch(
        mock_executor, aws_config_mock, value_in_kwagrs, value_in_sql):
    """Test load_temp_staging_raw_table method."""
    mock_executor.load_temp_staging_raw_table(
        temp_staging_raw_table='load_temp_staging_raw_table',
        aws=aws_config_mock, key_dir='s3://somepath', error_limit=1,
        error_on_column_count_mismatch=value_in_kwagrs)

    assert mock_executor.fetchall.call_count == 1
    mock_executor.fetchall.assert_any_call(
        SubstringMatcher(
            containing=['COPY INTO', 'test_db.test_schema',
                        f'ERROR_ON_COLUMN_COUNT_MISMATCH={value_in_sql}']),
        params={
            'aws_key_id': 'test_id',
            'aws_secret_key': 'test_secret',
            'aws_token': 'test_token',
            's3_path': 's3://somepath'})


@pytest.mark.parametrize(
    'date, sql_text',
    [
            ('2019-09-14', 'activity.royaltybearingplay'),
            # new field producttypekey for activity since 2021-02-18
            ('2021-02-18', 'activity.producttypekey'),
            # new field voiceactivatedstream for activity since 2021-02-19
            ('2021-02-19', 'activity.voiceactivatedstream'),
            # new fields totalplaycountforallvendors and countrycode
            # for activity since 2021-06-30
            ('2021-06-30', 'activity.totalplaycountforallvendors'),
            ('2021-06-30', 'activity.countrycode'),
    ]
)
def test_load_staging_raw_table(mock_executor, date, sql_text):
    """Test load_staging_raw_table method."""
    kwargs = {
        'licensor': 'theorchard',
        'temp_staging_raw_user_table':
            'amazon_unlimited_daily_user_report_ORED_AT_20190914',
        'temp_staging_raw_playlist_table':
            'amazon_unlimited_daily_playlist_report_ORED_AT_20190914',
        'temp_staging_raw_activity_table':
            'amazon_unlimited_daily_activity_report_ORED_AT_20190914',
        'activity_report_filename':
            'ORED_AT_20190914_Daily_Activity_Report.txt.gz',
        'user_report_filename':
            'ORED_AT_20190914_Daily_User_Report.txt.gz',
        'playlist_report_filename':
            'ORED_AT_20190914_Daily_Playlist_Report.txt.gz',
        'date_for_sqlloader': date}

    mock_executor.load_staging_raw_table(
        temp_staging_raw_table='temp_table',
        staging_raw_table='staging_raw_table',
        date=date,
        **kwargs)

    mock_executor.ex_mock.assert_any_call(
        SubstringMatcher(
            containing=[
                'INSERT INTO',
                'test_db.test_schema.staging_raw_table',
                'from test_db.test_schema.'
                'amazon_unlimited_daily_user_report_ORED_AT_20190914',
                'amazon_unlimited_daily_playlist_report_ORED_AT_20190914',
                'amazon_unlimited_daily_activity_report_ORED_AT_20190914',
                sql_text
            ]),
        params={
            'download_date': date,
            'date_for_sqlloader': date,
            'licensor': 'theorchard',
            'activity_report_filename':
                'ORED_AT_20190914_Daily_Activity_Report.txt.gz',
            'user_report_filename':
                'ORED_AT_20190914_Daily_User_Report.txt.gz',
            'playlist_report_filename':
                'ORED_AT_20190914_Daily_Playlist_Report.txt.gz'})


def _expected_bind_params(licensor, feedid, entity_name, service):
    """Build the exact JinjaSql bind-param dict for a datapulse load.

    The numeric suffixes match the order in which the {{ ... }} bind
    expressions appear in queries/load_staging_raw_datapulse.sql.
    """
    return {
        'download_date_1': '2026-05-22',
        'licensor_2': licensor,
        'feedid_3': feedid,
        'download_date_4': '2026-05-22',
        'entity_name_5': entity_name,
        'download_date_6': '2026-05-22',
        'entity_name_7': entity_name,
        'download_date_8': '2026-05-22',
        'service_9': service,
        'entity_name_10': entity_name,
    }


# Every report x licensor loads into the single unified staging_raw table.
DATAPULSE_TARGET = 'consumer_reporting.amazon.staging_raw_amazon_music'


@pytest.mark.parametrize(
    'report_name, licensor, expected_bind_params',
    [
        ('unlimited', 'theorchard',
         _expected_bind_params('theorchard', 37, 'ZQOR', 'Unlimited')),
        ('unlimited', 'sme',
         _expected_bind_params('sme', 37, 'Sony', 'Unlimited')),
        ('prime', 'theorchard',
         _expected_bind_params('theorchard', 35, 'ZQOR', 'Prime')),
        ('prime', 'sme',
         _expected_bind_params('sme', 35, 'Sony', 'Prime')),
        ('adsupported', 'theorchard',
         _expected_bind_params('theorchard', 36, 'ZQOR', 'FreeTier')),
        ('adsupported', 'sme',
         _expected_bind_params('sme', 36, 'Sony', 'FreeTier')),
    ]
)
def test_load_staging_raw_from_datapulse(
        mock_executor, monkeypatch, report_name, licensor,
        expected_bind_params):
    """SQL targets the unified table and binds the slice filters.

    Every report x licensor loads into the single unified
    consumer_reporting.amazon.staging_raw_amazon_music table; the run is
    scoped to its own slice purely via the service / entity_name / licensor /
    feedid bind values.
    """
    from feed_ingestion.flows.amazon_datapulse import (
        config as datapulse_config,
    )

    monkeypatch.setattr(
        datapulse_config, 'get_staging_raw_database_schema',
        lambda report: (
            'consumer_reporting', 'amazon',
            datapulse_config.reports[report]['staging_raw_table'],
        ))

    mock_executor.load_staging_raw_from_datapulse(
        date='2026-05-22',
        report_name=report_name,
        licensor=licensor,
    )

    mock_executor.ex_mock.assert_called_once()
    sql_call = mock_executor.ex_mock.call_args
    sql_text = sql_call.kwargs.get(
        'sql_template', sql_call.args[0] if sql_call.args else '')
    bind_params = sql_call.kwargs.get('params') or {}

    # The rendered SQL targets the unified staging_raw table and reads from
    # the datapulse source tables.
    assert SubstringMatcher(containing=[
        'INSERT INTO',
        DATAPULSE_TARGET,
        'staging_raw_amazon_daily_play_events play_events',
        'staging_raw_amazon_daily_customer_metadata',
        'staging_raw_amazon_daily_playlist_metadata',
        'play_events.entityname',
        'play_events.service',
    ]) == sql_text

    # The slice filters and download_date are passed exactly as bind values.
    assert bind_params == expected_bind_params


@pytest.fixture
def orchard_unlimited_executor(sf_config_mock, monkeypatch):
    """Concrete unlimited/theorchard executor with a wrapped execute()."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)
    executor = snowflake_executor.AmazonOrchardUnlimitedSFExecutor(
        sf_config_mock)
    with patch.object(executor, 'execute', wraps=executor.execute) as \
            executor.ex_mock:
        yield executor


def test_clean_staging_raw_table_datapulse(
        orchard_unlimited_executor, monkeypatch):
    """Datapulse clean targets the unified table in the datapulse schema.

    The delete is scoped by feedid so other reports' rows for the same date
    and licensor are preserved.
    """
    from feed_ingestion.flows.amazon_datapulse import (
        config as datapulse_config,
    )
    from feed_ingestion.flows.amazon_music import config

    monkeypatch.setattr(
        datapulse_config, 'get_staging_raw_database_schema',
        lambda report: (
            'consumer_reporting', 'amazon',
            datapulse_config.reports[report]['staging_raw_table']))

    orchard_unlimited_executor.clean_staging_raw_table(
        staging_raw_table=config.DATAPULSE_STAGING_RAW_TABLE,
        date='2026-05-22')

    orchard_unlimited_executor.ex_mock.assert_called_once()
    call = orchard_unlimited_executor.ex_mock.call_args
    sql_text = call.kwargs.get(
        'sql_template', call.args[0] if call.args else '')
    bind_params = call.kwargs.get('params') or {}

    assert SubstringMatcher(containing=[
        'DELETE FROM',
        'consumer_reporting.amazon.staging_raw_amazon_music',
        'download_date',
        'licensor',
        'feedid',
    ]) == sql_text
    assert bind_params == {
        'date': '2026-05-22',
        'licensor': 'theorchard',
        'feedid': 37,
    }


def test_clean_staging_raw_table_sftp_uses_per_report_table(
        orchard_unlimited_executor):
    """Non-datapulse clean still targets the per-report staging_raw table."""
    orchard_unlimited_executor.clean_staging_raw_table(
        staging_raw_table='staging_raw_amazon_unlimited', date='2026-05-22')

    orchard_unlimited_executor.ex_mock.assert_called_once()
    call = orchard_unlimited_executor.ex_mock.call_args
    sql_text = call.kwargs.get(
        'sql_template', call.args[0] if call.args else '')

    assert SubstringMatcher(containing=[
        'DELETE FROM',
        'test_db.test_schema.staging_raw_amazon_unlimited',
    ]) == sql_text


def test_update_dimension_table(monkeypatch, sf_config_mock):
    """Test update_dimension_table method."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)

    date = '2017-10-01'
    snowflake_executor.AmazonUnlimitedSFExecutor(
        sf_config_mock).update_dimension_table(date, 'dim_user')

    assert connect_mock.return_value.cursor.return_value.execute. \
        call_count == 1


def test_load_staging_fact_table_unlimited(monkeypatch, sf_config_mock):
    """Load staging fact table for unlimited report."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)

    date = '2017-10-01'
    snowflake_executor.AmazonOrchardUnlimitedSFExecutor(
        sf_config_mock).load_staging_fact_table(date)

    assert connect_mock.return_value.cursor.return_value.execute. \
        call_count == 1


def test_load_staging_fact_table_datapulse(
        orchard_unlimited_executor, monkeypatch):
    """Datapulse staging fact reads the unified datapulse staging table.

    The staging_fact / dim tables stay in the analytics db/schema, but the
    unified staging_raw_amazon_music is read from the datapulse schema, scoped
    by feedid + licensor. Extra common_kwargs (licensor/use_s3/country_list)
    are absorbed by **kwargs.
    """
    from feed_ingestion.flows.amazon_datapulse import (
        config as datapulse_config,
    )

    monkeypatch.setattr(
        datapulse_config, 'get_staging_raw_database_schema',
        lambda report: (
            'consumer_reporting', 'amazon',
            datapulse_config.reports[report]['staging_raw_table']))

    orchard_unlimited_executor.load_staging_fact_table(
        '2026-05-22', source=config.SOURCE_DATAPULSE,
        licensor='theorchard', use_s3=False, country_list='US')

    orchard_unlimited_executor.ex_mock.assert_called_once()
    call = orchard_unlimited_executor.ex_mock.call_args
    sql_text = call.kwargs.get(
        'sql_template', call.args[0] if call.args else '')
    bind_params = call.kwargs.get('params') or {}

    assert SubstringMatcher(containing=[
        'INSERT INTO',
        'consumer_reporting.amazon.staging_raw_amazon_music',
        'test_db.test_schema.dim_licensor',
        'srs.entity_name',
        'srs.feedid',
        'srs.licensor',
    ]) == sql_text
    assert bind_params == {
        'reportdate': '2026-05-22',
        'storeid': 187,
        'licensor': 'theorchard',
        'feedid': 37,
    }


def test_load_staging_fact_table_sftp_uses_per_licensor_query(
        orchard_unlimited_executor):
    """Non-datapulse staging fact still reads the per-licensor staging_raw."""
    orchard_unlimited_executor.load_staging_fact_table('2026-05-22')

    orchard_unlimited_executor.ex_mock.assert_called_once()
    call = orchard_unlimited_executor.ex_mock.call_args
    sql_text = call.kwargs.get(
        'sql_template', call.args[0] if call.args else '')

    assert SubstringMatcher(containing=[
        'INSERT INTO',
        'test_db.test_schema.staging_raw_amazon_unlimited',
    ]) == sql_text


def test_load_fact_error_table_unlimited(monkeypatch, sf_config_mock):
    """Load fact error table for unlimited report."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)

    date = '2017-10-01'

    snowflake_executor.AmazonOrchardUnlimitedSFExecutor(
        sf_config_mock).load_fact_error_data(date)

    assert connect_mock.return_value.cursor.return_value.execute. \
        call_count == 1


def test_load_fact_data_datapulse_drops_staging_raw_join(
        orchard_unlimited_executor):
    """Datapulse matched fact load selects from the staging fact table only.

    The legacy staging_raw join (only a download_date filter) is dropped, so
    the query references neither staging_raw nor a datapulse staging location.
    """
    orchard_unlimited_executor.load_fact_data(
        '2026-05-22', source=config.SOURCE_DATAPULSE)

    orchard_unlimited_executor.ex_mock.assert_called_once()
    call = orchard_unlimited_executor.ex_mock.call_args
    sql_text = call.kwargs.get(
        'sql_template', call.args[0] if call.args else '')
    bind_params = call.kwargs.get('params') or {}

    assert SubstringMatcher(containing=[
        'INSERT INTO',
        'test_db.test_schema.fact_analytics',
        'staging_fact_analytics_amazon_music_unlimited_theorchard_20260522',
    ]) == sql_text
    # No staging_raw table reference and no join at all in the query body.
    assert 'staging_raw_amazon' not in sql_text
    assert 'JOIN' not in sql_text
    assert bind_params == {}


def test_load_fact_data_sftp_keeps_staging_raw_join(
        orchard_unlimited_executor):
    """Non-datapulse matched fact load still joins the per-report table."""
    orchard_unlimited_executor.load_fact_data('2026-05-22')

    orchard_unlimited_executor.ex_mock.assert_called_once()
    call = orchard_unlimited_executor.ex_mock.call_args
    sql_text = call.kwargs.get(
        'sql_template', call.args[0] if call.args else '')

    assert SubstringMatcher(containing=[
        'INSERT INTO',
        'test_db.test_schema.staging_raw_amazon_unlimited',
    ]) == sql_text


def test_load_fact_error_data_datapulse(
        orchard_unlimited_executor, monkeypatch):
    """Datapulse fact error load reads the unified datapulse staging table.

    staging_raw is the driving table (unified staging_raw_amazon_music in the
    datapulse schema), scoped by feedid + licensor; the dim / staging fact
    tables stay in the analytics db/schema.
    """
    from feed_ingestion.flows.amazon_datapulse import (
        config as datapulse_config,
    )

    monkeypatch.setattr(
        datapulse_config, 'get_staging_raw_database_schema',
        lambda report: (
            'consumer_reporting', 'amazon',
            datapulse_config.reports[report]['staging_raw_table']))

    orchard_unlimited_executor.load_fact_error_data(
        '2026-05-22', source=config.SOURCE_DATAPULSE)

    orchard_unlimited_executor.ex_mock.assert_called_once()
    call = orchard_unlimited_executor.ex_mock.call_args
    sql_text = call.kwargs.get(
        'sql_template', call.args[0] if call.args else '')
    bind_params = call.kwargs.get('params') or {}

    assert SubstringMatcher(containing=[
        'INSERT INTO',
        'test_db.test_schema.fact_analytics_error',
        'consumer_reporting.amazon.staging_raw_amazon_music',
        'srs.entity_name',
        'srs.feedid',
        'stagingrawid IS NULL',
    ]) == sql_text
    assert bind_params == {
        'reportdate': '2026-05-22',
        'storeid': 187,
        'licensor': 'theorchard',
        'feedid': 37,
    }


def test_load_staging_fact_table_prime(monkeypatch, sf_config_mock):
    """Load staging fact table for prime report."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)

    date = '2017-10-01'
    snowflake_executor.AmazonOrchardPrimeSFExecutor(
        sf_config_mock).load_staging_fact_table(date)

    assert connect_mock.return_value.cursor.return_value.execute. \
        call_count == 1


def test_load_fact_error_table_prime(monkeypatch, sf_config_mock):
    """Load fact error table for prime report."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)

    date = '2017-10-01'

    snowflake_executor.AmazonOrchardPrimeSFExecutor(
        sf_config_mock).load_fact_error_data(date)

    assert connect_mock.return_value.cursor.return_value.execute. \
        call_count == 1


def test_load_staging_fact_table_adsupported(monkeypatch, sf_config_mock):
    """Load staging fact table for adsupported report."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)

    date = '2017-10-01'
    snowflake_executor.AmazonOrchardAdSupportedSFExecutor(
        sf_config_mock).load_staging_fact_table(date)

    assert connect_mock.return_value.cursor.return_value.execute. \
        call_count == 1


def test_load_fact_error_table_adsupported(monkeypatch, sf_config_mock):
    """Load fact error table for adsupported report."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)

    date = '2017-10-01'

    snowflake_executor.AmazonOrchardAdSupportedSFExecutor(
        sf_config_mock).load_fact_error_data(date)

    assert connect_mock.return_value.cursor.return_value.execute. \
        call_count == 1


def test_clean_aggregated_staging_raw_table_prime(monkeypatch, sf_config_mock):
    """Test clean_aggregated_staging_raw_table for prime report."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)

    date = '2017-10-01'
    snowflake_executor.AmazonOrchardAdSupportedSFExecutor(
        sf_config_mock).clean_aggregated_staging_raw_table(date, ['ORCA'])

    assert connect_mock.return_value.cursor.return_value.execute. \
        call_count == 1


def test_clean_aggregated_staging_raw_table_adsupported(
        monkeypatch, sf_config_mock):
    """Test clean_aggregated_staging_raw_table for adsupported report."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)

    date = '2017-10-01'

    snowflake_executor.AmazonOrchardPrimeSFExecutor(
        sf_config_mock).clean_aggregated_staging_raw_table(date, ['ORCA'])

    assert connect_mock.return_value.cursor.return_value.execute. \
        call_count == 1


def test_clean_aggregated_staging_raw_table_unlimited(
        monkeypatch, sf_config_mock):
    """Test clean_aggregated_staging_raw_table for unlimited report."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)

    date = '2017-10-01'

    snowflake_executor.AmazonOrchardUnlimitedSFExecutor(
        sf_config_mock).clean_aggregated_staging_raw_table(date, ['ORCA'])

    assert connect_mock.return_value.cursor.return_value.execute. \
        call_count == 1


def test_load_aggregated_staging_raw_table_prime(
        monkeypatch, sf_config_mock):
    """Test clean_aggregated_staging_raw_table for prime report."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)

    date = '2017-10-01'
    snowflake_executor.AmazonAdSupportedSFExecutor(
        sf_config_mock).load_aggregated_staging_raw_table(date)

    assert connect_mock.return_value.cursor.return_value.execute. \
        call_count == 1


def test_load_aggregated_staging_raw_table_adsupported(
        monkeypatch, sf_config_mock):
    """Test clean_aggregated_staging_raw_table for adsupported report."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)

    date = '2017-10-01'

    snowflake_executor.AmazonPrimeSFExecutor(
        sf_config_mock).load_aggregated_staging_raw_table(date)

    assert connect_mock.return_value.cursor.return_value.execute. \
        call_count == 1


def test_load_aggregated_staging_raw_table_unlimited(
        monkeypatch, sf_config_mock):
    """Test clean_aggregated_staging_raw_table for unlimited report."""
    connect_mock = MagicMock()
    monkeypatch.setattr(connector, 'connect', connect_mock)

    date = '2017-10-01'

    snowflake_executor.AmazonUnlimitedSFExecutor(
        sf_config_mock).load_aggregated_staging_raw_table(date)

    assert connect_mock.return_value.cursor.return_value.execute. \
        call_count == 1


def test_load_aggregated_staging_raw_table_datapulse(
        orchard_unlimited_executor, monkeypatch):
    """Datapulse aggregated load reads the unified datapulse staging table.

    It targets the analytics aggregated table but reads the unified
    staging_raw_amazon_music from the datapulse schema, scoped by
    feedid + licensor.
    """
    from feed_ingestion.flows.amazon_datapulse import (
        config as datapulse_config,
    )

    monkeypatch.setattr(
        datapulse_config, 'get_staging_raw_database_schema',
        lambda report: (
            'consumer_reporting', 'amazon',
            datapulse_config.reports[report]['staging_raw_table']))

    orchard_unlimited_executor.load_aggregated_staging_raw_table(
        '2026-05-22', source=config.SOURCE_DATAPULSE)

    orchard_unlimited_executor.ex_mock.assert_called_once()
    call = orchard_unlimited_executor.ex_mock.call_args
    sql_text = call.kwargs.get(
        'sql_template', call.args[0] if call.args else '')
    bind_params = call.kwargs.get('params') or {}

    assert SubstringMatcher(containing=[
        'INSERT INTO',
        'test_db.test_schema.staging_raw_amazon_aggregated_streams',
        'consumer_reporting.amazon.staging_raw_amazon_music',
        'feedid',
        'licensor',
    ]) == sql_text
    assert bind_params == {
        'date': '2026-05-22',
        'licensor': 'theorchard',
        'feedid': 37,
    }


def test_clean_aggregated_staging_raw_table_datapulse(
        orchard_unlimited_executor):
    """Datapulse clean scopes by feedid + licensor, not store-code orgs."""
    orchard_unlimited_executor.clean_aggregated_staging_raw_table(
        '2026-05-22', orgs=None, source=config.SOURCE_DATAPULSE)

    orchard_unlimited_executor.ex_mock.assert_called_once()
    call = orchard_unlimited_executor.ex_mock.call_args
    sql_text = call.kwargs.get(
        'sql_template', call.args[0] if call.args else '')
    bind_params = call.kwargs.get('params') or {}

    assert SubstringMatcher(containing=[
        'DELETE FROM',
        'test_db.test_schema.staging_raw_amazon_aggregated_streams',
        'download_date',
        'feedid',
        'licensor',
    ]) == sql_text
    assert bind_params == {
        'date': '2026-05-22',
        'licensor': 'theorchard',
        'feedid': 37,
    }