"""Unit tests for Amazon DataPulse flow."""

from unittest import mock
from unittest.mock import MagicMock

import pytest

from feed_ingestion.flows.amazon_datapulse.flow import Flow


@pytest.mark.parametrize(
    'context,expected',
    [
        (
            {'report_name': 'fraud_report', 'context_date': '2024-01-15'},
            'amazon_datapulse_fraud_report-2024-01-15',
        ),
        (
            {'report_name': 'fraud_report', 'context_date': '2024-12-31'},
            'amazon_datapulse_fraud_report-2024-12-31',
        ),
        (
            {
                'report_name': 'fraud_report',
                'context_date': '2024-06-01',
                'partition': {
                    'cadence': 'daily',
                    'region': 'US',
                    'report_date': '2024-06-01',
                    'entity_name': 'foo',
                    'service': 'music',
                    'music_territory': 'global',
                },
            },
            'amazon_datapulse_fraud_report_daily'
            '_us_2024_06_01_foo_music_global'
            '-2024-06-01',
        ),
    ],
)
def test_workflow_id(context, expected):
    """Test workflow_id combines feed_name, report_name, and date."""
    assert Flow().workflow_id(context) == expected


def test_flow_decider_schedules_all_activities():
    """Test decider schedules all expected activities in order."""
    flow = Flow()
    schedule = MagicMock()
    schedule_result = MagicMock()
    schedule_result.result = {}
    schedule.return_value = schedule_result

    flow.decider(schedule)

    schedule.assert_has_calls(
        [
            mock.call('bootstrap', mock.ANY),
            mock.call('fetch_from_athena', mock.ANY, requires=[mock.ANY]),
            mock.call('load_staging_raw_table', mock.ANY, requires=[mock.ANY]),
            mock.call('set_status_to_ingested', mock.ANY, requires=[mock.ANY]),
        ]
    )
    assert schedule.call_count == 4


def test_flow_decider_stops_on_bootstrap_stop():
    """Test decider returns early when bootstrap sets stop=True."""
    flow = Flow()
    schedule = MagicMock()
    schedule_result = MagicMock()
    schedule_result.result = {'bootstrap.stop': True}
    schedule.return_value = schedule_result

    flow.decider(schedule)

    schedule.assert_called_once_with('bootstrap', mock.ANY)


@pytest.mark.parametrize(
    'context,expected',
    [
        (
            {'report_name': 'fraud_report'},
            'amazon_datapulse_fraud_report',
        ),
        (
            {
                'report_name': 'fraud_report',
                'partition': {
                    'cadence': 'daily',
                    'region': 'US',
                    'report_date': '2024-01-15',
                    'entity_name': 'foo',
                    'service': 'music',
                    'music_territory': 'global',
                },
            },
            'amazon_datapulse_fraud_report_daily'
            '_us_2024_01_15_foo_music_global',
        ),
        (
            {
                'report_name': 'fraud_report',
                'partition': {
                    'cadence': 'bi-weekly',
                    'region': 'US',
                    'report_date': '2024-01-15',
                    'entity_name': 'foo',
                    'service': 'music',
                    'music_territory': 'global',
                },
            },
            'amazon_datapulse_fraud_report_bi_weekly_us'
            '_2024_01_15_foo_music_global',
        ),
    ],
)
def test_contextified_feed_name(context, expected):
    """Test contextified_feed_name with and without partition."""
    assert Flow().contextified_feed_name(context) == expected