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