certain/backend/tests/test_mail_import.py

"""Postfächer, Verschlüsselung und der Abruf gegen einen IMAP-Doppelgänger.

Ein echter Mailserver wäre im Test weder verfügbar noch wünschenswert. Der
Doppelgänger unten spricht nur die Handvoll IMAP-Befehle, die ``mail_import``
benutzt, und liefert die Muster-Mails aus ``fixtures/mail`` aus.
"""

import pathlib

import pytest

from app.services import geheimnis, mail_import

MUSTER = pathlib.Path(__file__).parent / "fixtures" / "mail"


class FakeImap:
    """Minimaler IMAP-Server: nur was ``mail_import`` tatsächlich aufruft."""

    def __init__(self, mails: list[bytes]):
        self.mails = mails
        self.gesehen: list[bytes] = []
        self.kopiert: list[tuple[bytes, str]] = []
        self.abgemeldet = False

    def login(self, user, passwort):
        return "OK", [b""]

    def list(self):
        return "OK", [b'(\\HasNoChildren) "/" "INBOX"', b'(\\HasNoChildren) "/" "Belege"']

    def select(self, ordner):
        return ("OK", [b"1"]) if "INBOX" in ordner or "Belege" in ordner else ("NO", [b""])

    def search(self, charset, kriterium):
        return "OK", [b" ".join(str(i + 1).encode() for i in range(len(self.mails)))]

    def fetch(self, uid, teile):
        return "OK", [(b"1 (RFC822 {1})", self.mails[int(uid) - 1])]

    def store(self, uid, befehl, flags):
        if "Seen" in flags:
            self.gesehen.append(uid)
        return "OK", [b""]

    def copy(self, uid, ordner):
        self.kopiert.append((uid, ordner))
        return "OK", [b""]

    def logout(self):
        self.abgemeldet = True
        return "BYE", [b""]


@pytest.fixture()
def fake_imap(monkeypatch):
    """Ersetzt die Verbindung; gibt eine Funktion zum Bestücken zurück."""
    gehalten = {}

    def bestuecken(*dateinamen):
        mails = [(MUSTER / n).read_bytes() for n in dateinamen]
        server = FakeImap(mails)
        gehalten["server"] = server
        monkeypatch.setattr(mail_import, "_verbinden", lambda quelle: server)
        return server

    return bestuecken


def _konto(admin, **kw):
    body = {"name": "Raiffeisen", "type": "girokonto", "currency": "CHF",
            "opening_balance_cents": 0}
    body.update(kw)
    return admin.post("/api/accounts", json=body).json()


def _postfach(admin, **kw):
    body = {"name": "Belege", "host": "imap.example.com", "port": 993,
            "username": "kundin@example.com", "password": "geheim123", "folder": "INBOX"}
    body.update(kw)
    r = admin.post("/api/mail/sources", json=body)
    assert r.status_code == 201, r.text
    return r.json()


def test_passwort_wird_verschluesselt_abgelegt_und_nie_ausgeliefert(admin):
    from app.database import SessionLocal
    from app.models import MailSource

    quelle = _postfach(admin)
    assert "password" not in quelle
    assert quelle["password_set"] is True

    db = SessionLocal()
    try:
        zeile = db.get(MailSource, quelle["id"])
        assert zeile.password_enc != "geheim123", "Klartext in der Datenbank"
        assert geheimnis.entschluesseln(zeile.password_enc) == "geheim123"
    finally:
        db.close()


def test_unlesbares_passwort_meldet_sich_verstaendlich(admin, monkeypatch):
    """Getauschtes Server-Geheimnis darf nicht in einem Stacktrace enden."""
    from app.database import SessionLocal
    from app.models import MailSource

    quelle = _postfach(admin)
    db = SessionLocal()
    try:
        zeile = db.get(MailSource, quelle["id"])
        zeile.password_enc = "kaputt"
        db.commit()
        with pytest.raises(mail_import.MailFehler) as fehler:
            mail_import._verbinden(zeile)
        assert "entschlüsseln" in str(fehler.value)
    finally:
        db.close()


def test_passwort_bleibt_wenn_es_beim_aendern_weggelassen_wird(admin):
    from app.database import SessionLocal
    from app.models import MailSource

    quelle = _postfach(admin)
    r = admin.put(f"/api/mail/sources/{quelle['id']}", json={
        "host": "imap.example.com", "username": "kundin@example.com", "folder": "Belege"})
    assert r.status_code == 200, r.text
    assert r.json()["folder"] == "Belege"

    db = SessionLocal()
    try:
        assert geheimnis.entschluesseln(db.get(MailSource, quelle["id"]).password_enc) == "geheim123"
    finally:
        db.close()


def test_abruf_legt_erkannte_zahlung_in_den_eingang(admin, fake_imap):
    konto = _konto(admin)
    quelle = _postfach(admin)
    admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
        "name": "Raiffeisen", "from_contains": "raiffeisen.ch", "account_id": konto["id"]})
    server = fake_imap("raiffeisen_zahlungsstatus.eml")

    r = admin.post(f"/api/mail/sources/{quelle['id']}/fetch")
    assert r.status_code == 200, r.text
    assert r.json()["neu"] == 1

    treffer = [e for e in admin.get("/api/inbox").json()
               if e["payee"] == "Sunrise GmbH / Yallo"]
    assert len(treffer) == 1
    assert treffer[0]["amount_cents"] == -1490
    assert treffer[0]["currency"] == "CHF"
    assert treffer[0]["date"] == "2026-07-31"
    assert treffer[0]["account_id"] == konto["id"]
    # Nichts wird stillschweigend gebucht.
    assert treffer[0]["status"] == "pending"
    assert treffer[0]["transaction_id"] is None
    assert server.gesehen, "Mail muss als gelesen markiert werden"


def test_dieselbe_mail_zweimal_ergibt_einen_eintrag(admin, fake_imap):
    """Die Message-ID zählt, nicht die UID – die vergibt der Server neu."""
    konto = _konto(admin)
    quelle = _postfach(admin)
    admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
        "from_contains": "raiffeisen.ch", "account_id": konto["id"]})
    fake_imap("raiffeisen_zahlungsstatus.eml")

    admin.post(f"/api/mail/sources/{quelle['id']}/fetch")
    vorher = len(admin.get("/api/inbox").json())
    zweiter = admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()
    assert zweiter["neu"] == 0
    assert zweiter["uebersprungen"] == 1
    assert len(admin.get("/api/inbox").json()) == vorher


def test_mail_ohne_passende_zuordnung_wird_uebergangen(admin, fake_imap):
    quelle = _postfach(admin)
    admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
        "from_contains": "sparkasse.de"})
    fake_imap("raiffeisen_zahlungsstatus.eml")

    ergebnis = admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()
    assert ergebnis["neu"] == 0
    assert ergebnis["uebersprungen"] == 1


def test_zuordnung_ohne_konto_landet_als_unklar(admin, fake_imap):
    """Ohne Zielkonto lässt sich nicht buchen – das muss sichtbar sein."""
    quelle = _postfach(admin)
    admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
        "from_contains": "revolut.com"})
    fake_imap("revolut_ueberweisung.eml")

    assert admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()["unklar"] == 1


def test_verbindungstest_liefert_die_ordner(admin, fake_imap):
    quelle = _postfach(admin)
    fake_imap()
    antwort = admin.post(f"/api/mail/sources/{quelle['id']}/test")
    assert antwort.status_code == 200, antwort.text
    assert "Belege" in antwort.json()["folders"]


def test_verschieben_statt_nur_markieren(admin, fake_imap):
    konto = _konto(admin)
    quelle = _postfach(admin, done_folder="Belege")
    admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
        "from_contains": "raiffeisen.ch", "account_id": konto["id"]})
    server = fake_imap("raiffeisen_zahlungsstatus.eml")

    admin.post(f"/api/mail/sources/{quelle['id']}/fetch")
    assert server.kopiert and "Belege" in server.kopiert[0][1]