spiele-redaktion/tests/test_export.py

"""Tests: Plugin „export“ — CSV-Inhalt, PDF-Erzeugung, Rollen-Zugriff, UI.

CSV-Tests pruefen Semikolon-Trennzeichen, UTF-8-BOM und Umlaute; PDF-Tests
pruefen den gueltigen Datei-Kopf („%PDF-“) und laufen nur, wenn WeasyPrint
auf dem System rendern kann (Fixture `pdf_faehig` skippt sonst).
Datenzugriffe in den Tests nutzen dieselben Module wie die App.
"""
from __future__ import annotations

import csv
import io
import itertools
import sys

import pytest

from fastapi.testclient import TestClient
from sqlalchemy import select, text

from redaktionskern.app import create_app
from tests.conftest import lege_benutzer_an, melde_an

_bgg_zaehler = itertools.count(100000)


# ---------------- Hilfen ----------------

def plugin_modul(app, name: str):
    plugin_instanz = app.state.registry.get(name) or sys.modules.get(
        f"spiele_redaktion_plugins.{name}"
    )
    return sys.modules[type(plugin_instanz).__module__]


def neuheit_anlegen(app, titel, verlag=None, autor=None, jahr=None, status="neuheit"):
    Neuheit = plugin_modul(app, "neuheiten").Neuheit
    with app.state.session_factory() as db:
        eintrag = Neuheit(
            titel=titel, verlag=verlag, autor=autor,
            erscheinungsjahr=jahr, bgg_id=next(_bgg_zaehler),
            status=status,
        )
        db.add(eintrag)
        db.commit()
        return eintrag.id


def planung_anlegen(app, titel, rezensent_id, **kwargs):
    Planungseintrag = plugin_modul(app, "planung").Planungseintrag
    with app.state.session_factory() as db:
        eintrag = Planungseintrag(titel=titel, rezensent_id=rezensent_id, **kwargs)
        db.add(eintrag)
        db.commit()
        return eintrag.id


def benutzer_objekt(app, username: str):
    from redaktionskern.auth.models import User

    with app.state.session_factory() as db:
        return db.scalar(select(User).where(User.username == username))


def csv_zeilen(antwort) -> list[list[str]]:
    """Antwort-Bytes als geparste CSV-Zeilen (Semikolon, BOM-abhaengig)."""
    text_dokument = antwort.content.decode("utf-8-sig")
    return list(csv.reader(io.StringIO(text_dokument), delimiter=";"))


@pytest.fixture
def daten_bestand(app):
    """Kleiner Redaktions-Bestand: Neuheiten (inkl. archiviert), Planung."""
    lege_benutzer_an(app, "rez1", "rezensent")
    rez = benutzer_objekt(app, "rez1")
    neuheit_anlegen(
        app, "Füchse im Wunderland", verlag="Hans im Glück Verlag",
        autor="Uwe Rosenberg", jahr=2025,
    )
    # Titel mit Trennzeichen/Sonderfällen für die CSV-Quoting-Prüfung.
    neuheit_anlegen(
        app, "Schätze; der Azteken", verlag='Anfänger "und" Profis',
        autor="Ärger Ödipus ß", jahr=2024,
    )
    neuheit_anlegen(app, "Alte Schachtel", verlag="Museum Verlag", jahr=1999,
                    status="archiv")
    planung_anlegen(
        app, "Füchse im Wunderland", rez.id, ausgabe="3/2025",
        notizen="Rezensionsexemplar anfordern – Müller",
    )
    planung_anlegen(
        app, "Schätze; der Azteken", rez.id, ausgabe="4/2025",
        status="in_bearbeitung",
    )


# ---------------- Seite & Zugriff ----------------

def test_export_seite_erreichbar(client):
    melde_an(client)
    antwort = client.get("/export")
    assert antwort.status_code == 200
    assert "Neuheitenliste" in antwort.text
    assert "Planungsliste" in antwort.text
    assert "Archiv" in antwort.text
    assert "CSV herunterladen" in antwort.text
    assert "PDF herunterladen" in antwort.text


def test_anonym_wird_zum_login_umgeleitet(client):
    assert client.get("/export", follow_redirects=False).status_code == 303
    assert client.get("/export/planung/csv", follow_redirects=False).status_code == 303
    assert client.get("/export/neuheiten/pdf", follow_redirects=False).status_code == 303
    weiterleitung = client.get("/export/planung/csv", follow_redirects=False)
    assert weiterleitung.headers["location"] == "/login"


def test_unbekannte_liste_404(client):
    melde_an(client)
    assert client.get("/export/geheim/csv").status_code == 404
    assert client.get("/export/geheim/pdf").status_code == 404


# ---------------- Rollen ----------------

@pytest.mark.parametrize("rolle", ["admin", "redakteur"])
@pytest.mark.parametrize("route", ["csv", "pdf"], ids=["csv", "pdf"])
def test_redaktion_exportiert_alles(app, client, daten_bestand, rolle, route, pdf_faehig):
    lege_benutzer_an(app, f"{rolle}1", rolle)
    melde_an(client, f"{rolle}1", "test-12345678")
    for liste in ("neuheiten", "planung", "archiv"):
        antwort = client.get(f"/export/{liste}/{route}")
        assert antwort.status_code == 200, liste
        if route == "pdf":
            assert antwort.content.startswith(b"%PDF-")
        else:
            assert antwort.headers["content-type"].startswith("text/csv")


def test_rezensent_nur_planung_csv(app, client, daten_bestand):
    melde_an(client, "rez1", "test-12345678")
    assert client.get("/export/planung/csv").status_code == 200
    assert client.get("/export/neuheiten/csv").status_code == 403
    assert client.get("/export/archiv/csv").status_code == 403


def test_rezensent_nur_planung_pdf(app, client, daten_bestand, pdf_faehig):
    melde_an(client, "rez1", "test-12345678")
    assert client.get("/export/planung/pdf").status_code == 200
    assert client.get("/export/neuheiten/pdf").status_code == 403
    assert client.get("/export/archiv/pdf").status_code == 403


def test_rezensent_sieht_gesperrte_karten(app, client):
    """UI: Rezensent sieht alle Karten, aber Neuheiten/Archiv ohne Buttons."""
    lege_benutzer_an(app, "rez1", "rezensent")
    melde_an(client, "rez1", "test-12345678")
    antwort = client.get("/export")
    assert antwort.status_code == 200
    assert "Nur für Admins und Redakteure" in antwort.text
    assert "/export/neuheiten/csv" not in antwort.text
    assert "/export/planung/csv" in antwort.text


# ---------------- CSV-Inhalt ----------------

def test_csv_bom_trennzeichen_umlaut(app, client, daten_bestand):
    melde_an(client)
    antwort = client.get("/export/planung/csv")
    assert antwort.status_code == 200
    assert antwort.headers["content-type"].startswith("text/csv")

    roh = antwort.content
    # 1) UTF-8 mit Byte-Order-Mark
    assert roh.startswith(b"\xef\xbb\xbf")
    # 2) Umlaute unzerstoert als UTF-8
    assert "Füchse im Wunderland".encode("utf-8") in roh
    assert "–".encode("utf-8") in roh  # Gedankenstrich aus der Notiz

    zeilen = csv_zeilen(antwort)
    kopf = zeilen[0]
    assert kopf[:2] == ["Titel", "Verlag"]
    assert "Ausgabe" in kopf and "Rezensent" in kopf and "Status" in kopf
    daten = [zeile for zeile in zeilen[1:] if zeile]
    assert len(daten) == 2
    fuechse = next(zeile for zeile in daten if zeile[0] == "Füchse im Wunderland")
    assert fuechse[kopf.index("Ausgabe")] == "3/2025"
    assert fuechse[kopf.index("Status")] == "Offen"
    # Rezensent ist aufgeloest (Anzeigename), nicht nur die ID:
    assert fuechse[kopf.index("Rezensent")] == "Rez1"


def test_csv_semikolon_getrennt_und_quoting(app, client, daten_bestand):
    """Trennzeichen ist wirklich ';'; Felder mit ';' werden korrekt gequotet."""
    melde_an(client)
    antwort = client.get("/export/neuheiten/csv")
    zeilen = csv_zeilen(antwort)
    kopf = zeilen[0]
    assert kopf[:6] == ["Titel", "Verlag", "Autor", "Erscheinungsjahr", "Status", "Aktualisiert am"]
    schaetze = next(z for z in zeilen[1:] if z and z[0].startswith("Schätze"))
    # Der Titel enthält selbst ein Semikolon — nur EIN Feld, sauber gequotet.
    assert schaetze[0] == "Schätze; der Azteken"
    assert schaetze[1] == 'Anfänger "und" Profis'
    assert schaetze[3] == "2024"


def test_csv_archiv_enthaelt_nur_archiviertes(app, client, daten_bestand):
    melde_an(client)
    zeilen = csv_zeilen(client.get("/export/archiv/csv"))
    titel = {z[0] for z in zeilen[1:] if z}
    assert titel == {"Alte Schachtel"}
    status_spalte = zeilen[0].index("Status")
    assert all(z[status_spalte] == "Archiviert" for z in zeilen[1:] if z)


def test_csv_leere_liste_nur_kopfzeile(app, client):
    melde_an(client)
    antwort = client.get("/export/planung/csv")
    assert antwort.status_code == 200
    assert antwort.content.startswith(b"\xef\xbb\xbf")
    zeilen = csv_zeilen(antwort)
    assert len(zeilen) == 1  # nur Kopfzeile
    assert zeilen[0][0] == "Titel"


def test_download_header_streaming(app, client, daten_bestand):
    """StreamingResponse: Attachment-Disposition mit Dateiname."""
    melde_an(client)
    antwort = client.get("/export/planung/csv")
    disposition = antwort.headers["content-disposition"]
    assert "attachment" in disposition
    assert "filename=\"spiele-planung-export-" in disposition


# ---------------- PDF ----------------

def test_pdf_gueltige_datei(app, client, daten_bestand, pdf_faehig):
    melde_an(client)
    antwort = client.get("/export/planung/pdf")
    assert antwort.status_code == 200
    assert antwort.headers["content-type"].startswith("application/pdf")
    # gültiger PDF-Kopf und -Abschluss:
    assert antwort.content.startswith(b"%PDF-")
    assert antwort.content.rstrip().endswith(b"%%EOF")
    assert len(antwort.content) > 500  # kein leeres Dokument


def test_pdf_leere_liste_bleibt_gueltig(app, client, pdf_faehig):
    melde_an(client)
    antwort = client.get("/export/archiv/pdf")
    assert antwort.status_code == 200
    assert antwort.content.startswith(b"%PDF-")
    assert antwort.content.rstrip().endswith(b"%%EOF")


def test_pdf_enthaelt_gerenderte_inhaltsstroeme(app, client, daten_bestand, pdf_faehig):
    """WeasyPrint komprimiert Inhaltsströme (Flate) — ohne Inhalt gäbe es keine."""
    melde_an(client)
    antwort = client.get("/export/neuheiten/pdf")
    assert antwort.content.startswith(b"%PDF-")
    assert b"FlateDecode" in antwort.content  # komprimierte Seiten-Inhalte vorhanden
    # Struktur: mindestens ein Seitenobjekt im Objektstrom des PDFs
    import zlib

    strome = []
    daten = antwort.content
    start = 0
    while True:
        start = daten.find(b"stream\r?\n", start)
        if start == -1:
            break
    # robuster: alle Streams zwischen "stream\n" und "endstream" dekomprimieren
    stuecke = daten.split(b"stream\n")
    for stueck in stuecke[1:]:
        roh = stueck.split(b"endstream")[0].rstrip(b"\r\n")
        try:
            strome.append(zlib.decompress(roh))
        except Exception:
            continue
    assert strome, "Mindestens ein dekomprimierbarer Inhaltsstrom erwartet"
    assert any(b"BT" in strom for strom in strome)  # Text-Zeichenoperationen


# ---------------- Audit & Robustheit ----------------

def test_export_wird_im_audit_log_protokolliert(app, client, daten_bestand):
    melde_an(client)
    client.get("/export/planung/csv")
    with app.state.engine.connect() as conn:
        eintraege = conn.execute(
            text(
                "SELECT action, objekt_typ FROM audit_eintraege "
                "WHERE action='exportiert' ORDER BY id DESC"
            )
        ).fetchall()
    assert len(eintraege) >= 1
    assert eintraege[0] == ("exportiert", "liste-planung")


def test_fehlendes_partner_plugin_degradiert_sauber(settings):
    """Ohne planungs-/neuheiten-Tabelle zeigt die Seite „nicht verfügbar“ statt Fehler."""
    from unittest.mock import patch

    from redaktionskern.db import Base

    class FehlendeTabellen:
        def get(self, _name):  # Base.metadata.tables-Ersatz
            return None

    app = create_app(settings)
    with patch.object(Base.metadata, "tables", FehlendeTabellen()):
        lege_benutzer_an(app, "admin2", "admin", "test-admin-123")
        client = TestClient(app)
        melde_an(client, "admin2", "test-admin-123")
        seite = client.get("/export")
        assert seite.status_code == 200
        assert "Nicht verfügbar" in seite.text
        ziel = client.get("/export/planung/csv", follow_redirects=False)
        assert ziel.status_code == 303
        assert "fehler" in ziel.headers["location"]