from contextlib import contextmanager from email.parser import BytesParser from email.policy import default import pytest from ccma.domain.models import Member from ccma.services.member_data_mail import ( data_review_recipients, generate_data_review_mails, mask_iban, member_data_fields, payment_reference, ) from ccma.storage.repository import MemberRepository, RepositoryError def _store(tmp_path): repository = MemberRepository(tmp_path) repository.initialize() organization = repository.get_configuration()["organization"] organization.update( { "name": "Chaos Computer Club Mannheim e.V.", "email": "verwaltung@example.org", "iban": "DE98670505050038907751", "bic": "MANSDE66XXX", } ) repository.save_organization(organization) return repository def _member(repository, first_name, last_name, **values): member = repository.create_member( first_name=first_name, last_name=last_name, birth_date="1990-01-01" ) for key, value in values.items(): setattr(member, key, value) repository.save_member(member) return repository.get_member(member.member_id) def test_the_mail_lists_the_stored_record_of_that_member(tmp_path): repository = _store(tmp_path) member = _member( repository, "Ada", "Lovelace", email="ada@example.org", phone="0621 123456", street="Hauptstraße 1", postal_code="68159", city="Mannheim", status="active", membership_started_at="2026-01-01", ) generated, warnings = generate_data_review_mails( repository, [member.member_id], delivery_mode="local", output_directory=tmp_path / "out", sender_name="Verwaltung C3MA", sender_email="verwaltung@example.org", signature="Der Vorstand", ) assert warnings == [] message = BytesParser(policy=default).parsebytes(generated[0].export_path.read_bytes()) content = message.get_content() assert message["To"] == "ada@example.org" assert "Liebes Mitglied," in content assert "folgende Daten haben wir über dich gespeichert" in content assert "Name: Ada Lovelace" in content assert "E-Mail: ada@example.org" in content assert "Telefon: 0621 123456" in content assert "PLZ und Ort: 68159 Mannheim" in content assert "Mitglied seit: 01.01.2026" in content assert "Zahlweise: Halbjährlich" in content # Both halves identify a transfer on their own, so both are offered. assert "deine Mitgliedsnummer und/oder deinen vollen Namen" in content assert f"{member.member_number} – Ada Lovelace" in content # Nothing bank-related is stored, so the note about the shortened IBAN is left # out instead of pointing at a line that is not there. assert "verkürzt" not in content assert content.rstrip().endswith("Der Vorstand") def test_a_field_without_a_value_is_shown_as_missing_not_as_a_gap(tmp_path): repository = _store(tmp_path) member = _member(repository, "Ada", "Lovelace", email="ada@example.org") fields = dict(member_data_fields(member)) assert fields["Telefon"] == "(nicht hinterlegt)" assert fields["Mitglied seit"] == "(nicht hinterlegt)" # Nothing bank-related is stored, so the member is not asked to check a mandate # they never gave. assert "IBAN" not in fields assert "Mandatsreferenz" not in fields def test_bank_details_are_listed_with_a_masked_iban(tmp_path): repository = _store(tmp_path) member = _member( repository, "Ada", "Lovelace", email="ada@example.org", account_holder="Ada Lovelace", iban="DE98670505050038907751", mandate_reference="MANDAT-42", mandate_signed_at="2026-02-01", mandate_active=True, ) generated, _warnings = generate_data_review_mails( repository, [member.member_id], delivery_mode="local", output_directory=tmp_path / "out", sender_name="Verwaltung C3MA", sender_email="verwaltung@example.org", signature="Der Vorstand", ) content = BytesParser(policy=default).parsebytes( generated[0].export_path.read_bytes() ).get_content() assert "IBAN: DE****************7751" in content assert "DE98670505050038907751" not in content assert "Mandatsreferenz: MANDAT-42" in content assert "Lastschriftmandat: aktiv, erteilt am 01.02.2026" in content assert "(Die IBAN zeigen wir absichtlich nur verkürzt an" in content def test_the_transfer_reference_joins_number_and_name(): member = Member( member_id="1", member_number="CCMA-0007", first_name="Ada", last_name="Lovelace" ) assert payment_reference(member) == "CCMA-0007 – Ada Lovelace" # The store insists on a member number, but a record read from elsewhere may not # have one -- the name then carries the reference without a dangling dash. assert ( payment_reference( Member(member_id="1", member_number="", first_name="Ada", last_name="Lovelace") ) == "Ada Lovelace" ) def test_mask_iban_keeps_country_and_last_four_digits(): assert mask_iban("DE98 6705 0505 0038 9077 51") == "DE****************7751" # Too short to mask meaningfully -- a broken value is shown as it is stored, so # the member can tell the board it is wrong. assert mask_iban("DE98") == "DE98" def test_every_selected_member_gets_their_own_mail_and_archive_copy(tmp_path): repository = _store(tmp_path) ada = _member(repository, "Ada", "Lovelace", email="ada@example.org", status="active") grace = _member(repository, "Grace", "Hopper", email="grace@example.org", status="active") generated, warnings = generate_data_review_mails( repository, [ada.member_id, grace.member_id], delivery_mode="local", output_directory=tmp_path / "out", sender_name="Verwaltung C3MA", sender_email="verwaltung@example.org", signature="Der Vorstand", ) assert warnings == [] assert [item.recipient for item in generated] == ["ada@example.org", "grace@example.org"] for item in generated: assert item.archive_path.parent.name == "Datenpruefung" assert item.archive_path.read_bytes() == item.export_path.read_bytes() ada_content = BytesParser(policy=default).parsebytes( generated[0].archive_path.read_bytes() ).get_content() assert "Ada Lovelace" in ada_content assert "Grace Hopper" not in ada_content event = repository.get_events(grace.member_id)[-1] assert event.event_type == "data_review_email_sent" assert event.data["recipient"] == "grace@example.org" assert event.references["document"].endswith(".eml") def test_a_member_without_an_address_is_reported_and_the_run_continues(tmp_path): repository = _store(tmp_path) silent = _member(repository, "Anon", "Ymous", status="active") ada = _member(repository, "Ada", "Lovelace", email="ada@example.org", status="active") generated, warnings = generate_data_review_mails( repository, [silent.member_id, ada.member_id], delivery_mode="local", output_directory=tmp_path / "out", sender_name="Verwaltung C3MA", sender_email="verwaltung@example.org", signature="Der Vorstand", ) assert [item.member_id for item in generated] == [ada.member_id] assert warnings == [f"{silent.member_number}: E-Mail-Adresse fehlt."] assert repository.get_events(silent.member_id)[-1].event_type != "data_review_email_sent" def test_default_recipients_are_the_live_memberships_with_an_address(tmp_path): repository = _store(tmp_path) _member(repository, "Ada", "Lovelace", email="ada@example.org", status="active") _member(repository, "Grace", "Hopper", email="grace@example.org", status="honorary") _member(repository, "Alan", "Turing", email="alan@example.org", status="ended") _member(repository, "Anon", "Ymous", status="active") names = [member.display_name for member in data_review_recipients(repository)] assert names == ["Grace Hopper", "Ada Lovelace"] def test_send_mode_delivers_via_smtp_without_local_files(tmp_path, monkeypatch): import ccma.services.member_data_mail as module repository = _store(tmp_path) member = _member(repository, "Ada", "Lovelace", email="ada@example.org", status="active") repository.save_email_settings( delivery_mode="send", smtp_host="smtp.example.org", smtp_port=587, smtp_security="starttls", smtp_username="verwaltung", smtp_password="secret", imap_host="", imap_port=993, imap_security="ssl", imap_username="", imap_password="", imap_drafts_folder="INBOX.Entwürfe", ) sent = [] monkeypatch.setattr(module, "smtp_session", contextmanager(lambda settings: iter(["client"]))) monkeypatch.setattr(module, "send_via_smtp", lambda client, content: sent.append(content)) generated, warnings = generate_data_review_mails( repository, [member.member_id], delivery_mode="send", sender_name="Verwaltung C3MA", sender_email="verwaltung@example.org", signature="Der Vorstand", ) assert warnings == [] assert generated[0].export_path is None assert generated[0].archive_path.read_bytes() == sent[0] assert repository.get_events(member.member_id)[-1].data["delivery_mode"] == "send" def test_a_template_broken_outside_ccma_stops_that_member_without_archiving(tmp_path): repository = _store(tmp_path) member = _member(repository, "Ada", "Lovelace", email="ada@example.org", status="active") (repository.mail_templates_root / "datenpruefung.txt").write_text( "Betreff: Datenprüfung\n\nHallo,\n{{#data}}\n{{field.label}}\n", encoding="utf-8" ) generated, warnings = generate_data_review_mails( repository, [member.member_id], delivery_mode="local", output_directory=tmp_path / "out", sender_name="Verwaltung C3MA", sender_email="verwaltung@example.org", signature="Der Vorstand", ) assert generated == [] assert "wird nicht geschlossen" in warnings[0] archive = repository.members_root / member.member_id / "files" / "documents" / "Datenpruefung" assert not any(archive.glob("*.eml")) assert repository.get_events(member.member_id)[-1].event_type != "data_review_email_sent" def test_an_unknown_delivery_mode_is_refused(tmp_path): repository = _store(tmp_path) with pytest.raises(RepositoryError, match="Versandmodus"): generate_data_review_mails( repository, [], delivery_mode="postal", sender_name="Verwaltung C3MA", sender_email="verwaltung@example.org", signature="Der Vorstand", )