business-card-scanner/tests/test_api.py

"""Durchstich durch die HTTP-Schnittstelle.

Ohne ANTHROPIC_API_KEY laeuft die Extraktion in den Leerlauf - Freistellen,
Speichern, Suchen, Bearbeiten und Export werden trotzdem vollstaendig gepruft.
"""
import asyncio
import time
from datetime import datetime, timedelta, timezone

import pytest
from fastapi.testclient import TestClient

from app import db, extract, mailer, reminders
from app.main import app
from tests.helpers import photo_bytes


def warte_bis_gelesen(client, card_ids, timeout=15.0):
    """Das Auslesen laeuft im Hintergrund - hier auf das Ende warten."""
    frist = time.monotonic() + timeout
    while time.monotonic() < frist:
        stati = [
            client.get(f"/api/cards/{i}").json()["extract_status"] for i in card_ids
        ]
        if all(status != "offen" for status in stati):
            return stati
        time.sleep(0.1)
    raise AssertionError("Karten wurden nicht fertig gelesen")


@pytest.fixture(scope="module")
def client():
    with TestClient(app) as test_client:
        yield test_client


@pytest.fixture(scope="module")
def scanned(client):
    response = client.post(
        "/api/scan", files={"file": ("tisch.jpg", photo_bytes(4), "image/jpeg")}
    )
    assert response.status_code == 200
    data = response.json()
    warte_bis_gelesen(client, [c["id"] for c in data["cards"]])
    return data


def test_healthz(client):
    assert client.get("/healthz").json() == {"status": "ok"}


def test_scan_legt_je_karte_einen_datensatz_an(scanned):
    assert scanned["count"] == 4
    assert scanned["fallback"] is False
    assert len({card["id"] for card in scanned["cards"]}) == 4


def test_jede_karte_hat_ein_gespeichertes_bild(client, scanned):
    for card in scanned["cards"]:
        image = client.get(card["image_url"])
        assert image.status_code == 200
        assert image.headers["content-type"] == "image/jpeg"
        assert len(image.content) > 1000


def test_liste_zaehlt_und_liefert_die_karten(client, scanned):
    data = client.get("/api/cards").json()

    assert data["stats"]["total"] >= 4
    assert data["stats"]["offen"] >= 4


def test_bearbeiten_und_suchen(client, scanned):
    card_id = scanned["cards"][0]["id"]

    patched = client.patch(
        f"/api/cards/{card_id}",
        json={"first_name": "Anna", "last_name": "Berg", "company": "Berg & Partner", "notes": "Messe"},
    )
    assert patched.status_code == 200
    assert patched.json()["last_name"] == "Berg"

    hits = client.get("/api/cards", params={"q": "berg"}).json()["cards"]
    assert [c["id"] for c in hits] == [card_id]

    by_note = client.get("/api/cards", params={"q": "Messe"}).json()["cards"]
    assert card_id in [c["id"] for c in by_note]


def test_unbekannte_felder_werden_abgewiesen(client, scanned):
    response = client.patch(
        f"/api/cards/{scanned['cards'][0]['id']}", json={"extract_status": "ok"}
    )

    assert response.status_code == 400


def test_vcard_download_markiert_die_karte_als_exportiert(client, scanned):
    card_id = scanned["cards"][1]["id"]
    client.patch(f"/api/cards/{card_id}", json={"first_name": "Bea", "last_name": "Kurz"})

    response = client.get(f"/vcf/{card_id}")

    assert response.status_code == 200
    assert response.headers["content-type"].startswith("text/vcard")
    assert "Bea-Kurz.vcf" in response.headers["content-disposition"]
    assert "FN:Bea Kurz" in response.text
    assert client.get(f"/api/cards/{card_id}").json()["exported_at"] is not None


def test_filter_offen_blendet_exportierte_karten_aus(client, scanned):
    offen = client.get("/api/cards", params={"filter": "offen"}).json()["cards"]

    assert all(card["exported_at"] is None for card in offen)


def test_erinnerung_anlegen_und_filtern(client, scanned):
    card_id = scanned["cards"][2]["id"]
    due = datetime.now(timezone.utc) + timedelta(days=3)

    created = client.post(
        f"/api/cards/{card_id}/reminders",
        json={"text": "Angebot nachfassen", "due_at": due.isoformat()},
    )
    assert created.status_code == 200

    detail = client.get(f"/api/cards/{card_id}").json()
    assert len(detail["reminders"]) == 1

    mit_erinnerung = client.get("/api/cards", params={"filter": "erinnerung"}).json()
    assert card_id in [c["id"] for c in mit_erinnerung["cards"]]
    assert mit_erinnerung["stats"]["erinnerungen"] >= 1


def test_erinnerung_ohne_datum_wird_abgewiesen(client, scanned):
    response = client.post(
        f"/api/cards/{scanned['cards'][2]['id']}/reminders",
        json={"text": "Ohne Termin", "due_at": "morgen"},
    )

    assert response.status_code == 400


def test_faellige_erinnerung_wird_verschickt(client, scanned, monkeypatch):
    card_id = scanned["cards"][3]["id"]
    client.patch(f"/api/cards/{card_id}", json={"first_name": "Carl", "last_name": "Spät"})
    past = datetime.now(timezone.utc) - timedelta(minutes=5)
    client.post(
        f"/api/cards/{card_id}/reminders",
        json={"text": "Rueckruf", "due_at": past.isoformat()},
    )

    verschickt = []
    monkeypatch.setattr(mailer, "configured", lambda: True)
    monkeypatch.setattr(mailer, "send", lambda subject, body: verschickt.append((subject, body)))

    assert reminders.process_due() == 1
    assert "Carl Spät" in verschickt[0][0]
    assert "Rueckruf" in verschickt[0][1]

    # Zweiter Durchlauf darf nicht erneut verschicken.
    assert reminders.process_due() == 0


def test_loeschen_entfernt_karte_und_bild(client):
    created = client.post(
        "/api/scan", files={"file": ("einzel.jpg", photo_bytes(1), "image/jpeg")}
    ).json()
    card_id = created["cards"][0]["id"]
    image_file = db.query_one("SELECT image_file FROM cards WHERE id = ?", (card_id,))["image_file"]

    from app import config

    assert (config.IMAGE_DIR / image_file).exists()
    assert client.delete(f"/api/cards/{card_id}").status_code == 200
    assert client.get(f"/api/cards/{card_id}").status_code == 404
    assert not (config.IMAGE_DIR / image_file).exists()


def test_kaputter_upload_gibt_400(client):
    response = client.post(
        "/api/scan", files={"file": ("kaputt.jpg", b"kein bild", "image/jpeg")}
    )

    assert response.status_code == 400


def test_pwa_dateien_werden_ausgeliefert(client):
    assert client.get("/").status_code == 200
    assert client.get("/sw.js").status_code == 200
    assert client.get("/manifest.webmanifest").status_code == 200
    assert client.get("/static/app.js").status_code == 200


def test_scan_antwortet_bevor_das_auslesen_fertig_ist(client, monkeypatch):
    """Der Kern der Hintergrundverarbeitung: die Antwort wartet nicht auf das
    Modell. Ohne das haengt ein Zwanzigerstapel eine Minute im Upload."""
    original = extract.extract_all

    async def langsam(images):
        await asyncio.sleep(1.5)
        return await original(images)

    monkeypatch.setattr(extract, "extract_all", langsam)

    start = time.monotonic()
    response = client.post(
        "/api/scan", files={"file": ("einzel.jpg", photo_bytes(1), "image/jpeg")}
    )
    dauer = time.monotonic() - start
    card_id = response.json()["cards"][0]["id"]

    assert dauer < 1.2, f"Antwort brauchte {dauer:.2f}s, hat also gewartet"
    assert client.get(f"/api/cards/{card_id}").json()["extract_status"] == "offen"

    monkeypatch.undo()
    warte_bis_gelesen(client, [card_id])


def test_offene_karten_zaehlen_nicht_als_unvollstaendig(client):
    """Waehrend des Lesens sind Felder naturgemaess leer - das ist kein Mangel."""
    created = client.post(
        "/api/scan", files={"file": ("einzel.jpg", photo_bytes(1), "image/jpeg")}
    ).json()
    card_id = created["cards"][0]["id"]
    warte_bis_gelesen(client, [card_id])  # sonst ueberschreibt der Hintergrundlauf
    db.execute("UPDATE cards SET extract_status = 'offen' WHERE id = ?", (card_id,))

    unvollstaendig = client.get(
        "/api/cards", params={"filter": "unvollstaendig"}
    ).json()["cards"]

    assert card_id not in [c["id"] for c in unvollstaendig]
    assert client.get("/api/cards").json()["stats"]["laeuft"] >= 1

    db.execute("UPDATE cards SET extract_status = 'ok' WHERE id = ?", (card_id,))


def test_sammelexport_ist_entfernt(client):
    assert client.get("/vcf", params={"ids": "egal"}).status_code == 404


def test_handeintrag_ueberlebt_den_hintergrundlauf(client, monkeypatch):
    """Die Karte ist sichtbar, bevor sie gelesen ist. Wer in der Zwischenzeit
    etwas eintraegt, darf es nicht wieder verlieren."""
    original = extract.extract_all

    async def langsam(images):
        await asyncio.sleep(1.0)
        return await original(images)

    monkeypatch.setattr(extract, "extract_all", langsam)
    card_id = client.post(
        "/api/scan", files={"file": ("einzel.jpg", photo_bytes(1), "image/jpeg")}
    ).json()["cards"][0]["id"]

    client.patch(f"/api/cards/{card_id}", json={"first_name": "Von", "last_name": "Hand", "notes": "Messe"})

    monkeypatch.undo()
    warte_bis_gelesen(client, [card_id])

    card = client.get(f"/api/cards/{card_id}").json()
    assert card["first_name"] == "Von"
    assert card["last_name"] == "Hand"
    assert card["notes"] == "Messe"


def test_anlass_wird_gespeichert_und_ist_durchsuchbar(client):
    created = client.post(
        "/api/scan",
        files={"file": ("tisch.jpg", photo_bytes(2), "image/jpeg")},
        data={"occasion": "Hannover Messe 2026"},
    ).json()
    ids = [c["id"] for c in created["cards"]]
    warte_bis_gelesen(client, ids)

    assert all(c["occasion"] == "Hannover Messe 2026" for c in created["cards"])

    gefunden = client.get("/api/cards", params={"q": "hannover messe"}).json()["cards"]
    assert set(ids) <= {c["id"] for c in gefunden}


def test_anlass_ist_nachtraeglich_aenderbar(client):
    card_id = client.post(
        "/api/scan", files={"file": ("einzel.jpg", photo_bytes(1), "image/jpeg")}
    ).json()["cards"][0]["id"]
    warte_bis_gelesen(client, [card_id])

    patched = client.patch(f"/api/cards/{card_id}", json={"occasion": "Kundentermin"})

    assert patched.json()["occasion"] == "Kundentermin"


def test_dubletten_stehen_in_der_detailansicht(client):
    erste = client.post(
        "/api/scan", files={"file": ("a.jpg", photo_bytes(1), "image/jpeg")}
    ).json()["cards"][0]["id"]
    zweite = client.post(
        "/api/scan", files={"file": ("b.jpg", photo_bytes(1), "image/jpeg")}
    ).json()["cards"][0]["id"]
    warte_bis_gelesen(client, [erste, zweite])
    for card_id in (erste, zweite):
        client.patch(f"/api/cards/{card_id}", json={"email": "doppelt@firma.de"})

    detail = client.get(f"/api/cards/{zweite}").json()

    assert erste in [d["id"] for d in detail["duplicates"]]
    assert detail["duplicates"][0]["reason"] == "gleiche E-Mail"


def test_rueckseite_nachtragen(client):
    card_id = client.post(
        "/api/scan", files={"file": ("vorne.jpg", photo_bytes(1), "image/jpeg")}
    ).json()["cards"][0]["id"]
    warte_bis_gelesen(client, [card_id])

    response = client.post(
        f"/api/cards/{card_id}/back",
        files={"file": ("hinten.jpg", photo_bytes(1), "image/jpeg")},
    )
    assert response.status_code == 200
    assert response.json()["back_image_url"].endswith("side=back")
    warte_bis_gelesen(client, [card_id])

    bild = client.get(f"/api/cards/{card_id}/image", params={"side": "back"})
    assert bild.status_code == 200
    assert bild.headers["content-type"] == "image/jpeg"


def test_ohne_rueckseite_gibt_es_kein_rueckseitenbild(client):
    card_id = client.post(
        "/api/scan", files={"file": ("vorne.jpg", photo_bytes(1), "image/jpeg")}
    ).json()["cards"][0]["id"]
    warte_bis_gelesen(client, [card_id])

    assert client.get(
        f"/api/cards/{card_id}/image", params={"side": "back"}
    ).status_code == 404


def test_loeschen_entfernt_auch_die_rueckseite(client):
    from app import config

    card_id = client.post(
        "/api/scan", files={"file": ("vorne.jpg", photo_bytes(1), "image/jpeg")}
    ).json()["cards"][0]["id"]
    warte_bis_gelesen(client, [card_id])
    client.post(
        f"/api/cards/{card_id}/back",
        files={"file": ("hinten.jpg", photo_bytes(1), "image/jpeg")},
    )
    warte_bis_gelesen(client, [card_id])
    row = db.query_one("SELECT * FROM cards WHERE id = ?", (card_id,))

    client.delete(f"/api/cards/{card_id}")

    assert not (config.IMAGE_DIR / row["image_file"]).exists()
    assert not (config.IMAGE_DIR / row["back_image_file"]).exists()