banyaro/tests/test_partner_qr.py

"""Smoke-Tests fuer Partner-QR-Kontingente (Bestellung, Scan, Registrierungs-Rueckverfolgung)."""

import secrets


def _create_code(client, admin, code=None):
    code = code or f"QRTEST{secrets.token_hex(3).upper()}"
    r = client.post("/api/admin/partner/codes", headers=admin["headers"], json={
        "code": code, "label": f"Testpartner {code}", "grants_founder": 0,
    })
    assert r.status_code == 201, r.text
    return r.json()


def _create_batch(client, admin, code_id, quantity=5):
    r = client.post(f"/api/admin/partner/codes/{code_id}/qr-batches",
                    headers=admin["headers"],
                    json={"label": "Sticker Testlauf", "quantity": quantity})
    assert r.status_code == 201, r.text
    return r.json()


def _batch_tokens(batch_id):
    from database import db
    with db() as conn:
        return [r["token"] for r in conn.execute(
            "SELECT token FROM partner_qr_codes WHERE batch_id=? ORDER BY seq", (batch_id,)
        ).fetchall()]


def test_batch_create_and_pdf(client, admin):
    """Kontingent anlegen -> N eindeutige Tokens + druckfertiges PDF."""
    code = _create_code(client, admin)
    batch = _create_batch(client, admin, code["id"], quantity=7)
    assert batch["quantity"] == 7 and batch["codes"] == 7
    tokens = _batch_tokens(batch["id"])
    assert len(set(tokens)) == 7

    r = client.get(f"/api/admin/partner/qr-batches/{batch['id']}/pdf", headers=admin["headers"])
    assert r.status_code == 200
    assert r.headers["content-type"] == "application/pdf"
    assert r.content[:4] == b"%PDF"


def test_scan_redirects_and_counts(client, admin):
    """/q/{token} -> 302 mit ref+qr, Scan-Zaehler steigt; unbekannter Token -> /."""
    code = _create_code(client, admin)
    batch = _create_batch(client, admin, code["id"], quantity=1)
    token = _batch_tokens(batch["id"])[0]

    r = client.get(f"/q/{token}", follow_redirects=False)
    assert r.status_code == 302
    # Bewusst KEIN Klartext-Code in der URL — sonst liest jeder Scanner den Code ab
    assert r.headers["location"] == f"/?qr={token}"
    assert code["code"] not in r.headers["location"]
    client.get(f"/q/{token}", follow_redirects=False)

    r = client.get("/api/admin/partner/qr-batches", headers=admin["headers"])
    mine = [b for b in r.json() if b["id"] == batch["id"]][0]
    assert mine["scans"] == 2

    r = client.get("/q/gibtsnich", follow_redirects=False)
    assert r.status_code == 302
    assert r.headers["location"] == "/"


def test_registration_attributed_to_qr(client, admin):
    """Registrierung mit ref+qr -> referred_qr gesetzt; unbestaetigt=Versuch, bestaetigt=Registrierung."""
    code = _create_code(client, admin)
    batch = _create_batch(client, admin, code["id"], quantity=2)
    token = _batch_tokens(batch["id"])[0]

    email = f"qr-{secrets.token_hex(4)}@example.com"
    r = client.post("/api/auth/register", json={
        "email": email, "password": "QrTest1234!", "name": f"qru{secrets.token_hex(3)}",
        "ref_code": code["code"], "qr_token": token,
    })
    assert r.status_code == 200, r.text

    from database import db
    with db() as conn:
        row = conn.execute("SELECT referred_by, referred_qr FROM users WHERE email=?", (email,)).fetchone()
    assert row["referred_by"] == -code["id"]
    assert row["referred_qr"] == token

    # Frisch registriert = E-Mail unbestaetigt -> zaehlt als Versuch
    def _batch():
        r = client.get("/api/admin/partner/qr-batches", headers=admin["headers"])
        return [b for b in r.json() if b["id"] == batch["id"]][0]
    assert _batch()["attempts"] == 1 and _batch()["registrations"] == 0

    # Nach E-Mail-Bestaetigung -> echte Registrierung
    with db() as conn:
        conn.execute("UPDATE users SET email_verified=1 WHERE email=?", (email,))
    assert _batch()["registrations"] == 1 and _batch()["attempts"] == 0

    # Admin-Detail-Liste: Account mit Datum, Status und Sticker-Nr
    r = client.get(f"/api/admin/partner/qr-batches/{batch['id']}/registrations",
                   headers=admin["headers"])
    assert r.status_code == 200
    regs = r.json()
    assert len(regs) == 1
    assert regs[0]["email"] == email
    assert regs[0]["email_verified"] == 1
    assert regs[0]["seq"] == 1
    assert regs[0]["created_at"]


def test_registration_with_qr_only(client, admin):
    """Registrierung NUR mit qr_token (ohne ref_code) -> Code wird server-seitig aufgeloest."""
    code = _create_code(client, admin)
    batch = _create_batch(client, admin, code["id"], quantity=1)
    token = _batch_tokens(batch["id"])[0]

    email = f"qro-{secrets.token_hex(4)}@example.com"
    r = client.post("/api/auth/register", json={
        "email": email, "password": "QrTest1234!", "name": f"qro{secrets.token_hex(3)}",
        "qr_token": token,
    })
    assert r.status_code == 200, r.text
    from database import db
    with db() as conn:
        row = conn.execute("SELECT referred_by, referred_qr FROM users WHERE email=?", (email,)).fetchone()
    assert row["referred_by"] == -code["id"]
    assert row["referred_qr"] == token


def test_code_registrations_with_channel(client, admin):
    """Admin-Liste aller Code-Einloesungen unterscheidet QR-Sticker und Link/manuell."""
    code = _create_code(client, admin)
    batch = _create_batch(client, admin, code["id"], quantity=1)
    token = _batch_tokens(batch["id"])[0]

    # 1x via QR, 1x via Code direkt
    client.post("/api/auth/register", json={
        "email": f"ch1-{secrets.token_hex(4)}@example.com", "password": "QrTest1234!",
        "name": f"ch1{secrets.token_hex(3)}", "ref_code": code["code"], "qr_token": token,
    })
    client.post("/api/auth/register", json={
        "email": f"ch2-{secrets.token_hex(4)}@example.com", "password": "QrTest1234!",
        "name": f"ch2{secrets.token_hex(3)}", "ref_code": code["code"],
    })

    r = client.get(f"/api/admin/partner/codes/{code['id']}/registrations", headers=admin["headers"])
    assert r.status_code == 200
    regs = r.json()
    assert len(regs) == 2
    channels = {(x["qr_seq"] or 0) for x in regs}
    assert channels == {0, 1}  # einer ohne QR (None), einer über Sticker #1


def test_paused_code_not_redeemable(client, admin):
    """Pausierter Code (Notbremse) -> keine Einloesung, Info-Endpoint 404; reaktivierbar."""
    code = _create_code(client, admin)
    r = client.post(f"/api/admin/partner/codes/{code['id']}/toggle", headers=admin["headers"])
    assert r.status_code == 200 and r.json()["active"] == 0

    # Info-Endpoint: wie nicht existent
    assert client.get(f"/api/partner/codes/{code['code']}/info").status_code == 404

    # Registrierung mit pausiertem Code -> keine Zuordnung
    email = f"qrp-{secrets.token_hex(4)}@example.com"
    r = client.post("/api/auth/register", json={
        "email": email, "password": "QrTest1234!", "name": f"qrp{secrets.token_hex(3)}",
        "ref_code": code["code"],
    })
    assert r.status_code == 200, r.text
    from database import db
    with db() as conn:
        row = conn.execute("SELECT referred_by FROM users WHERE email=?", (email,)).fetchone()
    assert row["referred_by"] is None

    # Reaktivieren funktioniert
    r = client.post(f"/api/admin/partner/codes/{code['id']}/toggle", headers=admin["headers"])
    assert r.json()["active"] == 1
    assert client.get(f"/api/partner/codes/{code['code']}/info").status_code == 200


def test_qr_token_must_match_code(client, admin):
    """QR-Token eines FREMDEN Codes wird nicht zugeordnet (Manipulationsschutz)."""
    code_a = _create_code(client, admin)
    code_b = _create_code(client, admin)
    batch_b = _create_batch(client, admin, code_b["id"], quantity=1)
    token_b = _batch_tokens(batch_b["id"])[0]

    email = f"qrx-{secrets.token_hex(4)}@example.com"
    r = client.post("/api/auth/register", json={
        "email": email, "password": "QrTest1234!", "name": f"qrx{secrets.token_hex(3)}",
        "ref_code": code_a["code"], "qr_token": token_b,
    })
    assert r.status_code == 200, r.text
    from database import db
    with db() as conn:
        row = conn.execute("SELECT referred_qr FROM users WHERE email=?", (email,)).fetchone()
    assert row["referred_qr"] is None


def test_partner_thank_you_mail(client, admin, user, monkeypatch):
    """E-Mail-Bestaetigung eines Geworbenen -> Dank-Mail mit Statistik an den Code-Besitzer."""
    from database import db
    with db() as conn:
        conn.execute("UPDATE users SET is_partner=1 WHERE email=?", (user["email"],))
        uid = conn.execute("SELECT id FROM users WHERE email=?", (user["email"],)).fetchone()["id"]

    code = _create_code(client, admin)
    batch = _create_batch(client, admin, code["id"], quantity=1)
    token = _batch_tokens(batch["id"])[0]
    client.post(f"/api/admin/partner/codes/{code['id']}/owner",
                headers=admin["headers"], json={"user_id": uid})

    sent = []
    import routes.outreach as outreach
    monkeypatch.setattr(outreach, "_send_smtp",
                        lambda to, subject, body, account="partner", html=None:
                        sent.append({"to": to, "subject": subject, "body": body}))

    email = f"qrm-{secrets.token_hex(4)}@example.com"
    r = client.post("/api/auth/register", json={
        "email": email, "password": "QrTest1234!", "name": f"qrm{secrets.token_hex(3)}",
        "ref_code": code["code"], "qr_token": token,
    })
    assert r.status_code == 200, r.text
    with db() as conn:
        vtoken = conn.execute(
            "SELECT verification_token FROM users WHERE email=?", (email,)
        ).fetchone()["verification_token"]

    sent.clear()  # Verifikations-Mail an den Neuen ignorieren
    r = client.get(f"/api/auth/verify-email/{vtoken}", follow_redirects=False)
    assert r.status_code == 302

    thank = [m for m in sent if m["to"] == user["email"]]
    assert len(thank) == 1, f"Dank-Mail fehlt: {sent}"
    assert "Danke" in thank[0]["subject"]
    assert "1 bestätigte Registrierung" in thank[0]["body"]   # Statistik
    assert "#1" in thank[0]["body"]                            # QR-Sticker-Herkunft

    # Doppelt verifizieren -> keine zweite Mail
    sent.clear()
    client.get(f"/api/auth/verify-email/{vtoken}", follow_redirects=False)
    assert not [m for m in sent if m["to"] == user["email"]]


def test_partner_self_service_qr(client, admin, user):
    """Code-Besitzer sieht eigene Kontingente + kann PDF laden; Fremde nicht."""
    from database import db
    with db() as conn:
        conn.execute("UPDATE users SET is_partner=1 WHERE email=?", (user["email"],))
        uid = conn.execute("SELECT id FROM users WHERE email=?", (user["email"],)).fetchone()["id"]

    code = _create_code(client, admin)
    batch = _create_batch(client, admin, code["id"], quantity=3)

    # Ohne Besitzer: leere Liste
    r = client.get("/api/partner/my-qr", headers=user["headers"])
    assert r.status_code == 200 and r.json() == []

    # Besitzer zuordnen -> sichtbar + PDF
    r = client.post(f"/api/admin/partner/codes/{code['id']}/owner",
                    headers=admin["headers"], json={"user_id": uid})
    assert r.status_code == 200
    r = client.get("/api/partner/my-qr", headers=user["headers"])
    assert [b["id"] for b in r.json()] == [batch["id"]]
    assert r.json()[0]["codes_used"] == 0
    r = client.get(f"/api/partner/my-qr/{batch['id']}/pdf", headers=user["headers"])
    assert r.status_code == 200 and r.content[:4] == b"%PDF"

    # Einzel-Code-Status: alle frei, dann einer verbraucht
    r = client.get(f"/api/partner/my-qr/{batch['id']}/codes", headers=user["headers"])
    codes_list = r.json()
    assert len(codes_list) == 3
    assert all(c["registrations"] == 0 and c["scans"] == 0 for c in codes_list)

    token = codes_list[0]["token"]
    client.get(f"/q/{token}", follow_redirects=False)
    email = f"qrc-{secrets.token_hex(4)}@example.com"
    client.post("/api/auth/register", json={
        "email": email, "password": "QrTest1234!", "name": f"qrc{secrets.token_hex(3)}",
        "ref_code": code["code"], "qr_token": token,
    })
    with db() as conn:
        conn.execute("UPDATE users SET email_verified=1 WHERE email=?", (email,))

    r = client.get(f"/api/partner/my-qr/{batch['id']}/codes", headers=user["headers"])
    first = [c for c in r.json() if c["seq"] == 1][0]
    assert first["scans"] == 1 and first["registrations"] == 1
    assert first["first_registration_at"]
    r = client.get("/api/partner/my-qr", headers=user["headers"])
    assert r.json()[0]["codes_used"] == 1

    # Dashboard-Stats: eigener Code mit Zahlen + Profil-Status
    r = client.get("/api/partner/my-stats", headers=user["headers"])
    assert r.status_code == 200
    d = r.json()
    mycode = [c for c in d["codes"] if c["id"] == code["id"]][0]
    assert mycode["registrations"] == 1
    assert mycode["registrations_month"] == 1
    assert "approved" in d["profile"]