"""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"]