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.housekeeper import Housekeeper
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_follow_members_monthly_payment_frequency(tmp_path):
    repository, member = _repository(tmp_path)
    member.accepted_at = "2025-01-01"
    member.membership_started_at = "2025-01-01"
    member.payment_frequency = "monthly"
    repository.save_member(member)
    repository.save_contributions(member.member_id, ContributionData())

    Housekeeper(repository).run(today=date(2026, 3, 31))
    debits, warnings = pending_direct_debits(
        repository,
        due_from=date(2026, 1, 1),
        due_until=date(2026, 3, 31),
    )

    assert warnings == []
    assert len(debits) == 1
    assert debits[0].amount == Decimal("37.50")
    assert len(debits[0].claim_ids) == 3


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/")