import io
from unittest.mock import MagicMock


from backend.actions.bulk_delete import (
    MAX_EMAILS,
    bulk_delete_emails_from_csv,
    parse_emails_from_csv,
)
from backend.queries import insert_deleted_profiles


def _csv(content: str) -> io.BytesIO:
    return io.BytesIO(content.encode("utf-8"))


# --- parse_emails_from_csv ---


def test_parse_emails_basic():
    f = _csv("email\nfoo@example.com\nbar@example.com\n")
    emails, errors = parse_emails_from_csv(f)
    assert emails == ["foo@example.com", "bar@example.com"]
    assert errors == []


def test_parse_emails_header_name_doesnt_matter():
    f = _csv("address\nfoo@example.com\n")
    emails, errors = parse_emails_from_csv(f)
    assert emails == ["foo@example.com"]
    assert errors == []


def test_parse_emails_lowercased():
    f = _csv("email\nFOO@EXAMPLE.COM\n")
    emails, errors = parse_emails_from_csv(f)
    assert emails == ["foo@example.com"]
    assert errors == []


def test_parse_emails_deduplicates():
    f = _csv("email\nfoo@example.com\nFOO@EXAMPLE.COM\nfoo@example.com\n")
    emails, errors = parse_emails_from_csv(f)
    assert emails == ["foo@example.com"]
    assert errors == []


def test_parse_emails_invalid_email_reports_error():
    f = _csv("email\nnot-an-email\n")
    emails, errors = parse_emails_from_csv(f)
    assert emails == []
    assert len(errors) == 1
    assert "not-an-email" in errors[0]


def test_parse_emails_empty_file():
    f = _csv("")
    emails, errors = parse_emails_from_csv(f)
    assert emails == []
    assert len(errors) == 1


def test_parse_emails_header_only():
    f = _csv("email\n")
    emails, errors = parse_emails_from_csv(f)
    assert emails == []
    assert len(errors) == 1


def test_parse_emails_skips_blank_rows():
    f = _csv("email\nfoo@example.com\n\n  \nbar@example.com\n")
    emails, errors = parse_emails_from_csv(f)
    assert emails == ["foo@example.com", "bar@example.com"]
    assert errors == []


def test_parse_emails_empty_email_cell_reports_error():
    # A row with a comma-separated empty first cell is an empty email, not a blank row
    f = _csv("email,other\n,extra\nfoo@example.com,extra\n")
    emails, errors = parse_emails_from_csv(f)
    assert "foo@example.com" in emails
    assert any("empty" in e for e in errors)


def test_parse_emails_strips_bom():
    content = "email\nfoo@example.com\n"
    f = io.BytesIO(content.encode("utf-8-sig"))
    emails, errors = parse_emails_from_csv(f)
    assert emails == ["foo@example.com"]
    assert errors == []


def test_parse_emails_rejects_over_limit():
    lines = ["email"] + [f"user{i}@example.com" for i in range(MAX_EMAILS + 1)]
    f = _csv("\n".join(lines))
    emails, errors = parse_emails_from_csv(f)
    assert emails == []
    assert len(errors) == 1
    assert str(MAX_EMAILS) in errors[0] or "100,000" in errors[0]


def test_parse_emails_accepts_exactly_at_limit():
    lines = ["email"] + [f"user{i}@example.com" for i in range(MAX_EMAILS)]
    f = _csv("\n".join(lines))
    emails, errors = parse_emails_from_csv(f)
    assert len(emails) == MAX_EMAILS
    assert errors == []


# --- insert_deleted_profiles ---


def _mock_session_for_insert(count: int = 2):
    session = MagicMock()
    session.get_current_warehouse.return_value = '"MY_WH"'
    session.sql.return_value.collect.return_value = []
    matched_df = MagicMock()
    matched_df.count.return_value = count
    session.table.return_value.select.return_value.join.return_value.select.return_value = matched_df
    return session, matched_df


def test_insert_deleted_profiles_returns_zero_for_empty_list():
    session = MagicMock()
    count = insert_deleted_profiles(session, [], "testuser", "QA")
    assert count == 0
    session.create_dataframe.assert_not_called()


def test_insert_deleted_profiles_returns_matched_count():
    session, matched_df = _mock_session_for_insert(count=2)
    count = insert_deleted_profiles(session, ["a@b.com", "c@d.com"], "testuser", "QA")
    assert count == 2


def test_insert_deleted_profiles_reads_correct_tables_qa():
    session, _ = _mock_session_for_insert()
    insert_deleted_profiles(session, ["a@b.com"], "testuser", "QA")
    session.table.assert_called_once_with("PREFERENCE_CENTER.QA.FAN_PROFILE")
    save_args = session.table.return_value.select.return_value.join.return_value.select.return_value.write.save_as_table.call_args
    assert save_args.args[0] == "PREFERENCE_CENTER.QA.DELETED_PROFILE"


def test_insert_deleted_profiles_reads_correct_tables_prod():
    session, _ = _mock_session_for_insert()
    insert_deleted_profiles(session, ["a@b.com"], "testuser", "PROD")
    session.table.assert_called_once_with("PREFERENCE_CENTER.PROD.FAN_PROFILE")
    save_args = session.table.return_value.select.return_value.join.return_value.select.return_value.write.save_as_table.call_args
    assert save_args.args[0] == "PREFERENCE_CENTER.PROD.DELETED_PROFILE"


def test_insert_deleted_profiles_appends():
    session, _ = _mock_session_for_insert()
    insert_deleted_profiles(session, ["a@b.com"], "testuser", "QA")
    save_kwargs = session.table.return_value.select.return_value.join.return_value.select.return_value.write.save_as_table.call_args.kwargs
    assert save_kwargs.get("mode") == "append"


def test_insert_deleted_profiles_emails_passed_to_create_dataframe():
    session, _ = _mock_session_for_insert()
    insert_deleted_profiles(session, ["a@b.com", "c@d.com"], "testuser", "QA")
    data_arg = session.create_dataframe.call_args.args[0]
    assert data_arg == [["a@b.com"], ["c@d.com"]]


# --- bulk_delete_emails_from_csv ---


def test_bulk_delete_returns_errors_on_invalid_csv():
    session = MagicMock()
    f = _csv("email\nnot-valid\n")
    result = bulk_delete_emails_from_csv(session, f, "RBOMBERG", "QA")
    assert result.inserted_count == 0
    assert len(result.errors) == 1
    session.create_dataframe.assert_not_called()


def test_bulk_delete_header_only_csv_returns_error():
    session = MagicMock()
    f = _csv("email\n")
    result = bulk_delete_emails_from_csv(session, f, "RBOMBERG", "QA")
    assert result.inserted_count == 0
    assert len(result.errors) == 1
    session.create_dataframe.assert_not_called()


def test_bulk_delete_calls_insert_and_returns_count():
    session, _ = _mock_session_for_insert(count=1)
    f = _csv("email\nfoo@example.com\n")
    result = bulk_delete_emails_from_csv(session, f, "RBOMBERG", "QA")
    assert result.inserted_count == 1
    assert result.errors == []
    session.create_dataframe.assert_called_once()


def test_bulk_delete_rejects_over_limit():
    session = MagicMock()
    lines = ["email"] + [f"user{i}@example.com" for i in range(MAX_EMAILS + 1)]
    f = _csv("\n".join(lines))
    result = bulk_delete_emails_from_csv(session, f, "RBOMBERG", "QA")
    assert result.inserted_count == 0
    assert len(result.errors) == 1
    session.create_dataframe.assert_not_called()