from unittest.mock import MagicMock

import pandas as pd
from snowflake.snowpark import Row

from backend.dtos import FileUploadFilterOption
from backend.queries import (
    call_delete_file_upload_procedure,
    convert_country_to_iso2,
    insert_file_upload_rows,
    query_existing_campaign_ids,
    query_existing_file_upload_filenames,
    query_file_upload_campaign_artists,
    query_file_upload_campaign_stats,
    query_file_upload_filter_options,
    query_file_upload_status,
)


def _mock_session(*sql_return_sequences):
    session = MagicMock()
    session.sql.return_value.collect.side_effect = list(sql_return_sequences)
    return session


# --- query_file_upload_filter_options ---


def test_query_file_upload_filter_options_maps_rows():
    row = Row(
        VENDOR_ID="v1",
        LABEL_NAME="Label A",
        GLOBAL_PARTICIPANT_ID="g1",
        VIRTUAL_PARTICIPANT_ID=None,
        ARTIST_NAME="Artist A",
        MAILING_LIST_ID="m1",
        MAILING_LIST_NAME="List A",
        TYPE="artist",
    )
    session = _mock_session([row])

    result = query_file_upload_filter_options(session)

    assert result == [
        FileUploadFilterOption(
            vendor_id="v1",
            label_name="Label A",
            global_participant_id="g1",
            virtual_participant_id=None,
            artist_name="Artist A",
            mailing_list_id="m1",
            mailing_list_name="List A",
            type="artist",
        )
    ]


def test_query_file_upload_filter_options_maps_custom_list_rows():
    row = Row(
        VENDOR_ID="v1",
        LABEL_NAME="Label A",
        GLOBAL_PARTICIPANT_ID=None,
        VIRTUAL_PARTICIPANT_ID="vp1",
        ARTIST_NAME="Custom List A",
        MAILING_LIST_ID="m1",
        MAILING_LIST_NAME="List A",
        TYPE="custom list",
    )
    session = _mock_session([row])

    result = query_file_upload_filter_options(session)

    assert result == [
        FileUploadFilterOption(
            vendor_id="v1",
            label_name="Label A",
            global_participant_id=None,
            virtual_participant_id="vp1",
            artist_name="Custom List A",
            mailing_list_id="m1",
            mailing_list_name="List A",
            type="custom list",
        )
    ]


def test_query_file_upload_filter_options_empty():
    session = _mock_session([])
    assert query_file_upload_filter_options(session) == []


def test_query_file_upload_filter_options_uses_env_schema():
    session = _mock_session([])

    query_file_upload_filter_options(session, "PROD")

    sql = session.sql.call_args[0][0]
    assert "PREFERENCE_CENTER.PROD.FAN_MAILING_LIST" in sql
    assert "fansifter_app_reporting.PROD.artist_roster_main_rep" in sql
    assert "fansifter_app_reporting.PROD.artist_roster_local_rep" in sql
    assert "fansifter_app_reporting.PROD.artist_roster" in sql
    assert "facts.PROD.vendor" in sql
    assert "facts.PROD.global_participant" in sql
    assert "fansifter_app_reporting.PROD.virtual_participant" in sql
    assert "fansifter_app_reporting.PROD.custom_list" in sql


def test_query_file_upload_filter_options_coerces_numeric_ids():
    row = Row(
        VENDOR_ID=799210,
        LABEL_NAME="Label A",
        GLOBAL_PARTICIPANT_ID=12345,
        VIRTUAL_PARTICIPANT_ID=None,
        ARTIST_NAME="Artist A",
        MAILING_LIST_ID=67890,
        MAILING_LIST_NAME="List A",
        TYPE="artist",
    )
    session = _mock_session([row])

    result = query_file_upload_filter_options(session)

    assert result == [
        FileUploadFilterOption(
            vendor_id="799210",
            label_name="Label A",
            global_participant_id="12345",
            virtual_participant_id=None,
            artist_name="Artist A",
            mailing_list_id="67890",
            mailing_list_name="List A",
            type="artist",
        )
    ]


# --- query_existing_file_upload_filenames / query_existing_campaign_ids ---


def _mock_table_chain(session, return_value):
    session.table.return_value.select.return_value.distinct.return_value.to_pandas.return_value = return_value


def test_query_existing_file_upload_filenames_reads_correct_table():
    session = MagicMock()
    _mock_table_chain(session, pd.DataFrame({"FILE_NAME": ["a.csv", "b.csv"]}))

    result = query_existing_file_upload_filenames(session, "QA")

    session.table.assert_called_once_with(
        "FANSIFTER_APP_REPORTING.QA_CRM_FANS.EVENT_FILE_UPLOAD"
    )
    assert result == ["a.csv", "b.csv"]


def test_query_existing_file_upload_filenames_returns_empty_on_error():
    session = MagicMock()
    session.table.side_effect = ValueError("boom")

    assert query_existing_file_upload_filenames(session, "QA") == []


def test_query_existing_campaign_ids_reads_correct_table_prod():
    session = MagicMock()
    _mock_table_chain(session, pd.DataFrame({"CAMPAIGN_ID": ["abc"]}))

    result = query_existing_campaign_ids(session, "PROD")

    session.table.assert_called_once_with(
        "FANSIFTER_APP_REPORTING.PROD_CRM_FANS.EVENT_FILE_UPLOAD"
    )
    assert result == ["abc"]


# --- insert_file_upload_rows ---


def test_insert_file_upload_rows_writes_expected_table():
    session = MagicMock()
    dataframe = pd.DataFrame({"EMAIL_REQUIRED": ["a@example.com"]})

    rows_uploaded = insert_file_upload_rows(session, dataframe, "QA")

    session.create_dataframe.assert_called_once_with(dataframe)
    save_call = session.create_dataframe.return_value.write.save_as_table
    save_call.assert_called_once_with(
        "FANSIFTER_APP_REPORTING.QA_CRM_FANS.EVENT_FILE_UPLOAD",
        mode="append",
        column_order="name",
    )
    assert rows_uploaded == 1


# --- call_delete_file_upload_procedure ---


def test_call_delete_file_upload_procedure_calls_expected_proc():
    session = MagicMock()
    session.call.return_value = "Deleted 5 rows"

    result = call_delete_file_upload_procedure(
        session, "upload.csv", "fixing bad data", "TEST_USER", "QA"
    )

    session.call.assert_called_once_with(
        "FANSIFTER_APP_REPORTING.QA_CRM_FANS.DELETE_BACKOFFICE_FILE_UPLOAD",
        "upload.csv",
        "fixing bad data",
        "TEST_USER",
    )
    assert result == "Deleted 5 rows"


# --- query_file_upload_status ---


def test_query_file_upload_status_returns_counts_and_failures():
    session = MagicMock()
    session.table.return_value.filter.return_value.select.return_value.distinct.return_value.count.return_value = 42
    failure_row = Row(NOT_VALID_REASON="invalid email", FANS=3)
    session.table.return_value.filter.return_value.groupBy.return_value.agg.return_value.collect.return_value = [
        failure_row
    ]

    result = query_file_upload_status(session, "my-upload", "QA")

    assert result.processed_count == 42
    assert len(result.failures) == 1
    assert result.failures[0].reason == "invalid email"
    assert result.failures[0].count == 3
    assert session.table.call_args_list[0].args[0] == (
        "FANSIFTER_APP_REPORTING.QA_CRM_FANS.EVENT_FANRESPONSE_VALIDATED"
    )
    assert session.table.call_args_list[1].args[0] == (
        "FANSIFTER_APP_REPORTING.QA_CRM_FANS.EVENT_FANRESPONSE_ERROR"
    )


def test_query_file_upload_status_returns_empty_on_error():
    session = MagicMock()
    session.table.side_effect = ValueError("boom")

    result = query_file_upload_status(session, "my-upload", "QA")

    assert result.processed_count == 0
    assert result.failures == []


# --- query_file_upload_campaign_artists ---


def test_query_file_upload_campaign_artists_reads_correct_table():
    session = MagicMock()
    _mock_table_chain(
        session, pd.DataFrame({"ARTIST_NAME": ["Beta Artist", "Alpha Artist"]})
    )

    result = query_file_upload_campaign_artists(session, "PROD")

    session.table.assert_called_once_with(
        "FANSIFTER_APP_REPORTING.PROD_CRM_FANS.FILE_UPLOAD_CAMPAIGNS"
    )
    assert result == ["Alpha Artist", "Beta Artist"]


def test_query_file_upload_campaign_artists_returns_empty_on_error():
    session = MagicMock()
    session.table.side_effect = ValueError("boom")

    assert query_file_upload_campaign_artists(session, "QA") == []


# --- query_file_upload_campaign_stats ---


def test_query_file_upload_campaign_stats_maps_rows():
    session = MagicMock()
    row = Row(
        ARTIST_NAME="Henry Moodie",
        FILE_SOURCE_DESCRIPTION="Some source",
        FILE_NAME="upload.csv",
        FAN_COUNT=7,
    )
    session.sql.return_value.collect.return_value = [row]

    result = query_file_upload_campaign_stats(session, "Henry Moodie", "PROD")

    assert len(result) == 1
    assert result[0].artist_name == "Henry Moodie"
    assert result[0].file_source_description == "Some source"
    assert result[0].file_name == "upload.csv"
    assert result[0].fan_count == 7
    sql = session.sql.call_args.args[0]
    assert "FANSIFTER_APP_REPORTING.PROD_CRM_FANS.FILE_UPLOAD_CAMPAIGNS" in sql
    assert "WHERE uc.ARTIST_NAME = 'Henry Moodie'" in sql


def test_query_file_upload_campaign_stats_escapes_single_quotes():
    session = MagicMock()
    session.sql.return_value.collect.return_value = []

    query_file_upload_campaign_stats(session, "O'Brien", "QA")

    sql = session.sql.call_args.args[0]
    assert "WHERE uc.ARTIST_NAME = 'O''Brien'" in sql


def test_query_file_upload_campaign_stats_returns_empty_on_error():
    session = MagicMock()
    session.sql.side_effect = ValueError("boom")

    assert query_file_upload_campaign_stats(session, "Henry Moodie", "QA") == []


# --- convert_country_to_iso2 ---


def test_convert_country_to_iso2_calls_udf_with_env():
    session = MagicMock()
    converted_df = pd.DataFrame({"COUNTRY_REGION": ["US"]})
    session.create_dataframe.return_value.withColumn.return_value.to_pandas.return_value = converted_df
    dataframe = pd.DataFrame({"COUNTRY_REGION": ["United States"]})

    result = convert_country_to_iso2(session, dataframe, "PROD")

    session.create_dataframe.assert_called_once_with(dataframe)
    assert result is converted_df