Plugin export: Listen als CSV (Excel-DE) und PDF (WeasyPrint)
- CSV über stdlib csv: Semikolon-getrennt, UTF-8 mit BOM, korrektes
Quoting; PDF über WeasyPrint mit deutscher Kopfzeile (je Seite
wiederholt), Erstell-Datum und Seitenzahlen (Seite X von Y)
- Drei Listen: Neuheiten, Planung, Archiv (Neuheiten gefiltert auf
Status „archiviert“); Partner-Tabellen über gemeinsame Metadata,
ohne Plugin-Importe — fehlende Plugins degradieren sauber
- Rollen: Admin/Redakteur alles, Rezensent nur Planungsliste;
serverseitige Prüfung an den Download-Endpunkten (StreamingResponse),
UI-Karten für nicht erlaubte Listen ausgegraut
- Jeder Download best effort im Audit-Log (Aktion „exportiert“)
- WeasyPrint lazy importiert: ohne System-Bibliotheken bleiben App und
CSV nutzbar; Tests skippen PDF-Fälle nach Installationsversuch
- 20 neue Tests (CSV-Inhalt/BOM/Umlaute/Quoting, PDF-Gültigkeit,
Rollen, Audit, Degradierung); 161 gesamt grün
- README: Plugin-Dokumentation, Fortschritt export auf ✅
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
"""Gemeinsame Fixtures: App mit temporaerer SQLite-DB und TestClient."""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from dataclasses import replace
|
||||
from pathlib import Path
|
||||
|
||||
@@ -66,6 +67,58 @@ def melde_an(
|
||||
return antwort
|
||||
|
||||
|
||||
def _installiere_weasyprint():
|
||||
"""Best effort: WeasyPrint nachinstallieren (z. B. frisches Checkout ohne uv sync).
|
||||
|
||||
Zuerst uv im Projektverzeichnis, als Rueckfallebene pip. Scheitert beides
|
||||
(kein Netz, keine Tools), skippen die PDF-Tests spaeter sauber.
|
||||
"""
|
||||
import shutil
|
||||
import subprocess
|
||||
|
||||
befehle = [
|
||||
["uv", "pip", "install", "weasyprint"],
|
||||
[sys.executable, "-m", "pip", "install", "--quiet", "weasyprint"],
|
||||
]
|
||||
for befehl in befehle:
|
||||
if shutil.which(befehl[0]) is None:
|
||||
continue
|
||||
try:
|
||||
subprocess.run(
|
||||
befehl,
|
||||
cwd=str(PROJEKT_WURZEL),
|
||||
check=False,
|
||||
timeout=180,
|
||||
capture_output=True,
|
||||
)
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def pdf_faehig():
|
||||
"""WeasyPrint bereitstellen; PDF-Tests ueberspringen, wenn das System es nicht hergibt.
|
||||
|
||||
- fehlt das Paket: einmaliger Installationsversuch (uv, sonst pip)
|
||||
- fehlen die System-Bibliotheken (Pango/fontconfig & Co.): Probe-Renderung
|
||||
schlaegt fehl -> pytest.skip statt rotem Test
|
||||
"""
|
||||
try:
|
||||
import weasyprint # noqa: F401
|
||||
except ImportError:
|
||||
_installiere_weasyprint()
|
||||
try:
|
||||
import weasyprint # noqa: F401
|
||||
except ImportError:
|
||||
pytest.skip("WeasyPrint ist nicht installiert und konnte nicht nachinstalliert werden.")
|
||||
from weasyprint import HTML
|
||||
|
||||
try:
|
||||
HTML(string="<p>Probe äöü ß</p>").write_pdf()
|
||||
except Exception as exc:
|
||||
pytest.skip(f"WeasyPrint kann auf diesem System nicht rendern (System-Bibliotheken fehlen?): {exc}")
|
||||
|
||||
|
||||
def lege_benutzer_an(app, username: str, role: str, password: str = "test-12345678"):
|
||||
"""Legt direkt per DB einen Benutzer an (fuer Rollen-Tests)."""
|
||||
from redaktionskern.auth.models import User
|
||||
|
||||
319
tests/test_export.py
Normal file
319
tests/test_export.py
Normal file
@@ -0,0 +1,319 @@
|
||||
"""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"]
|
||||
@@ -41,7 +41,9 @@ def test_plugins_haben_navigation(app):
|
||||
|
||||
# Voll implementierte Plugins tragen keinen Platzhalter-Text mehr auf ihrer
|
||||
# Seite; die Prüfung läuft weiter für alle übrigen Platzhalter.
|
||||
VOLL_IMPLEMENTIERT = {"audit-log", "neuheiten", "benachrichtigung", "dedup", "planung"}
|
||||
VOLL_IMPLEMENTIERT = {
|
||||
"audit-log", "neuheiten", "benachrichtigung", "dedup", "planung", "export",
|
||||
}
|
||||
PLATZHALTER_STUBS = ERWARTETE_STUBS - VOLL_IMPLEMENTIERT
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user