CCMA/tests/test_sepa.py

import xml.etree.ElementTree as ET
from datetime import UTC, date, datetime
from decimal import Decimal
from email.parser import BytesParser
from email.policy import default

from ccma.domain.models import ContributionData
from ccma.services.sepa import PAIN_NAMESPACE, _safe, csv_text, pain008_bytes, pending_direct_debits
from ccma.services.sepa_mail import debit_mail_bytes, generate_debit_mails
from ccma.storage.repository import MemberRepository
from ccma.ui.sepa_dialog import _next_weekday


def _repository(tmp_path):
    repository = MemberRepository(tmp_path)
    repository.initialize()
    member = repository.create_member(first_name="Ada", last_name="Lovelace", member_number="C3-42")
    member.status = "active"
    member.email = "ada@example.org"
    member.account_holder = "Ada Lovelace"
    member.iban = "DE89370400440532013000"
    member.bic = "COBADEFFXXX"
    member.mandate_reference = "MANDAT-42"
    member.mandate_signed_at = "2025-01-10"
    member.mandate_active = True
    repository.save_member(member)
    repository.save_contributions(
        member.member_id,
        ContributionData(
            claims=[
                {
                    "claim_id": "due",
                    "title": "Mitgliedsbeitrag 2026",
                    "amount": "150.00",
                    "due_date": "2026-01-31",
                    "status": "open",
                },
                {
                    "claim_id": "future",
                    "title": "Mitgliedsbeitrag 2027",
                    "amount": "150.00",
                    "due_date": "2027-01-31",
                    "status": "open",
                },
            ],
            payments=[{"payment_id": "payment", "amount": "25.00"}],
            allocations=[{"payment_id": "payment", "claim_id": "due", "amount": "25.00"}],
        ),
    )
    return repository, member


def test_pending_debits_select_due_open_balance(tmp_path):
    repository, member = _repository(tmp_path)
    debits, warnings = pending_direct_debits(repository, due_until=date(2026, 12, 31))

    assert warnings == []
    assert len(debits) == 1
    assert debits[0].member_id == member.member_id
    assert debits[0].amount == Decimal("125.00")
    assert debits[0].claim_ids == ("due",)


def test_pending_debits_exclude_claims_before_lower_date(tmp_path):
    repository, _member = _repository(tmp_path)

    debits, warnings = pending_direct_debits(
        repository,
        due_from=date(2026, 2, 1),
        due_until=date(2026, 12, 31),
    )

    assert warnings == []
    assert debits == []


def test_next_weekday_moves_weekend_to_monday():
    assert _next_weekday(date(2026, 8, 1)) == date(2026, 8, 3)
    assert _next_weekday(date(2026, 8, 3)) == date(2026, 8, 3)


def test_sepa_text_normalizes_unsupported_characters():
    assert _safe("Sébastien O’Connor – Müller", 70) == "Sebastien O'Connor - Müller"


def test_csv_is_semicolon_separated_and_uses_decimal_comma(tmp_path):
    repository, _member = _repository(tmp_path)
    debits, _warnings = pending_direct_debits(repository, due_until=date(2026, 12, 31))

    text = csv_text(debits, collection_date=date(2026, 8, 3))

    assert "Einzugsdatum;Mitgliedsnummer" in text
    assert "125,00;EUR;MANDAT-42" in text


def test_pain008_contains_control_sum_mandate_and_creditor(tmp_path):
    repository, _member = _repository(tmp_path)
    debits, _warnings = pending_direct_debits(repository, due_until=date(2026, 12, 31))
    organization = {
        "name": "Chaos Computer Club Mannheim e.V.",
        "iban": "DE89370400440532013000",
        "bic": "COBADEFFXXX",
        "creditor_id": "DE98ZZZ09999999999",
    }

    content = pain008_bytes(
        debits,
        collection_date=date(2026, 8, 3),
        organization=organization,
        message_id="CCMA-TEST",
        created_at=datetime(2026, 7, 30, 12, 0, tzinfo=UTC),
    )
    root = ET.fromstring(content)
    ns = {"p": PAIN_NAMESPACE}

    assert root.findtext(".//p:GrpHdr/p:CtrlSum", namespaces=ns) == "125.00"
    assert root.findtext(".//p:MndtId", namespaces=ns) == "MANDAT-42"
    assert (
        root.findtext(".//p:DrctDbtTx/p:CdtrSchmeId//p:Othr/p:Id", namespaces=ns)
        == "DE98ZZZ09999999999"
    )
    assert root.find("p:CstmrDrctDbtInitn/p:PmtInf/p:CdtrSchmeId", ns) is None
    assert root.findtext(".//p:ReqdColltnDt", namespaces=ns) == "2026-08-03"


def test_pain008_contains_one_logical_batch_with_multiple_transactions(tmp_path):
    repository, _member = _repository(tmp_path)
    debit = pending_direct_debits(repository, due_until=date(2026, 12, 31))[0][0]
    organization = {
        "name": "Chaos Computer Club Mannheim e.V.",
        "iban": "DE89370400440532013000",
        "bic": "COBADEFFXXX",
        "creditor_id": "DE98ZZZ09999999999",
    }

    root = ET.fromstring(
        pain008_bytes(
            [debit, debit],
            collection_date=date(2026, 8, 3),
            organization=organization,
            message_id="CCMA-BATCH-TEST",
        )
    )
    ns = {"p": PAIN_NAMESPACE}

    assert len(root.findall("p:CstmrDrctDbtInitn", ns)) == 1
    assert len(root.findall(".//p:PmtInf", ns)) == 1
    assert len(root.findall(".//p:DrctDbtTxInf", ns)) == 2
    assert root.findtext(".//p:GrpHdr/p:NbOfTxs", namespaces=ns) == "2"
    assert root.findtext(".//p:GrpHdr/p:CtrlSum", namespaces=ns) == "250.00"


def test_debit_mail_is_thunderbird_draft(tmp_path):
    repository, member = _repository(tmp_path)
    debit = pending_direct_debits(repository, due_until=date(2026, 12, 31))[0][0]

    content = debit_mail_bytes(
        recipient=member.email,
        first_name=member.first_name,
        debit=debit,
        collection_date=date(2026, 8, 3),
        creditor_id="DE98ZZZ09999999999",
        sender_name="Verwaltung C3MA",
        sender_email="verwaltung@example.org",
        signature="Der Vorstand",
        created_at=datetime(2026, 7, 30, 12, 0, tzinfo=UTC),
    )
    message = BytesParser(policy=default).parsebytes(content)

    assert message["To"] == "ada@example.org"
    assert str(message["X-Mozilla-Draft-Info"]).strip().startswith("internal/draft")
    assert "125.00 Euro" in message.get_content()
    assert "MANDAT-42" in message.get_content()


def test_generated_mail_is_exported_archived_and_logged(tmp_path):
    repository, member = _repository(tmp_path)
    organization = repository.get_configuration()["organization"]
    organization.update(
        {
            "name": "Chaos Computer Club Mannheim e.V.",
            "email": "verwaltung@example.org",
            "iban": "DE89370400440532013000",
            "bic": "COBADEFFXXX",
            "creditor_id": "DE98ZZZ09999999999",
        }
    )
    repository.save_organization(organization)
    debits, _warnings = pending_direct_debits(repository, due_until=date(2026, 12, 31))

    generated, warnings = generate_debit_mails(
        repository,
        debits,
        collection_date=date(2026, 8, 3),
        output_directory=tmp_path / "mail-export",
        sender_name="Verwaltung C3MA",
        sender_email="verwaltung@example.org",
        signature="Der Vorstand",
    )

    assert warnings == []
    assert generated[0].export_path.is_file()
    assert generated[0].archive_path.is_file()
    assert generated[0].archive_path.read_bytes() == generated[0].export_path.read_bytes()
    event = repository.get_events(member.member_id)[-1]
    assert event.event_type == "sepa_notification_generated"
    assert event.references["document"].startswith("documents/SEPA/")