96d6af8163
Teil 1 — Coverbilder: - Migration 0004_bild_url (neuheiten), 0002_bild_url (planung, archiv mit beiden Spiegel-Tabellen); nullable VARCHAR(500), SQLite↔Postgres portabel, bestehende Zeilen bleiben erhalten - BGG-Client extrahiert Cover aus der Thing-Antwort (bevorzugt <image>, sonst <thumbnail>; protokoll-relative URLs → https) - Sync-Upsert pflegt bild_url beim Update; fehlendes Bild löscht keines - Web-Quellen übernehmen img-URLs aus den Listenelementen (src/data-src, relative Pfade gegen die Seiten-URL aufgelöst); SPIEL-Essen nutzt das BILDER-Feld des JSON-Endpunkts - Verschiebung Neuheiten→Planung und Archivierung/Wiederherstellen schreiben bild_url mit - Anzeige: kleine Vorschaubilder (40x56) in den Listen, groß (max 200px) in Planung-Bearbeiten/Prüf-Dialog; Platzhalter-Div ohne Bild; Hotlinking mit loading=lazy + decoding=async Teil 2 — Responsives Layout: - Basislayout: Burger-Menü (Alpine.js) auf Mobil, Desktop-Leiste ab md: - Listen kollabieren <768px zu Karten-Ansichten (Cover links, Titel + Metadaten rechts, Status-Badge oben); Desktop behält Tabellenform - Touch-Ziele >= 44px auf Mobil, Formulare vollbreit mit Labels oben - Sync-Übersicht/Audit-Log bleiben horizontal scrollbare Tabellen Tests: 15 neue Fälle (tests/test_covers.py) — Migration inkl. Daten-Erhalt, BGG-Parsing gemockt, Quellen-Übernahme, Template-Render; 316 Tests grün. README aktualisiert.
577 lines
23 KiB
Python
577 lines
23 KiB
Python
"""Tests: Plugin „planung“ — Eintragswege, dedup-Verzahnung, Rollen, UI.
|
|
|
|
Abhängigkeiten werden gemockt: das dedup-Plugin durch einen Stub mit
|
|
steuerbaren Prüfergebnissen, benachrichtigung und audit-log durch
|
|
aufzeichnende Stubs. Ein Test beweist zusätzlich die Integration mit den
|
|
echten Plugins (In-App-Nachricht + Audit-Eintrag in der Datenbank).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import html
|
|
import sys
|
|
from dataclasses import dataclass, field
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import func, select, text
|
|
|
|
from redaktionskern.app import create_app
|
|
from tests.conftest import ADMIN_PASSWORD, lege_benutzer_an, melde_an
|
|
|
|
|
|
# ---------------- 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, bgg_id=None):
|
|
Neuheit = plugin_modul(app, "neuheiten").Neuheit
|
|
with app.state.session_factory() as db:
|
|
eintrag = Neuheit(titel=titel, verlag=verlag, bgg_id=bgg_id)
|
|
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 planung_zaehlen(app) -> int:
|
|
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
|
with app.state.session_factory() as db:
|
|
return db.scalar(select(func.count()).select_from(Planungseintrag))
|
|
|
|
|
|
def sichtbar(antwort) -> str:
|
|
"""Response-Text mit aufgelösten HTML-Entities (für Umlaut-Assertions)."""
|
|
return html.unescape(antwort.text)
|
|
|
|
|
|
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))
|
|
|
|
|
|
# ---------------- Stubs (gemockte Abhängigkeiten) ----------------
|
|
|
|
@dataclass
|
|
class Aufgezeichnet:
|
|
"""Ein aufgezeichneter Aufruf von audit.log / send_notification."""
|
|
daten: dict = field(default_factory=dict)
|
|
|
|
|
|
class StubDedup:
|
|
"""Ersetzt das dedup-Plugin: liefert gesteuerte Prüfergebnisse."""
|
|
|
|
name = "dedup"
|
|
|
|
def __init__(self, modul):
|
|
self._modul = modul # echte Datenklassen für realistische Ergebnisse
|
|
self.aufrufe: list[dict] = []
|
|
self.warteschlange: list = []
|
|
self.standard_treffer = None
|
|
|
|
def treffer_einreihen(self, ergebnis) -> None:
|
|
self.warteschlange.append(ergebnis)
|
|
|
|
def konflikt(self, art, beschreibung, details=None, optionen=None, empfehlung=None,
|
|
titel="", verlag=None, bgg_id=None):
|
|
P, K = self._modul.PruefErgebnis, self._modul.TitelKonflikt
|
|
return P(
|
|
titel=titel, verlag=verlag, bgg_id=bgg_id,
|
|
konflikte=[K(art=art, beschreibung=beschreibung, details=details or {})],
|
|
verlags_optionen=list(optionen or []),
|
|
titel_empfehlung=empfehlung,
|
|
)
|
|
|
|
async def check_titel(self, titel, verlag=None, bgg_id=None, *, user=None):
|
|
self.aufrufe.append(
|
|
{"titel": titel, "verlag": verlag, "bgg_id": bgg_id,
|
|
"user_id": user.id if user is not None else None}
|
|
)
|
|
if self.warteschlange:
|
|
ergebnis = self.warteschlange.pop(0)
|
|
ergebnis.titel = titel
|
|
ergebnis.verlag = verlag
|
|
ergebnis.bgg_id = bgg_id
|
|
return ergebnis
|
|
P = self._modul.PruefErgebnis
|
|
return P(titel=titel, verlag=verlag, bgg_id=bgg_id)
|
|
|
|
|
|
class StubBenachrichtigung:
|
|
name = "benachrichtigung"
|
|
|
|
def __init__(self):
|
|
self.gesendet: list[Aufgezeichnet] = []
|
|
|
|
async def send_notification(self, user, titel, text, kategorie="allgemein"):
|
|
self.gesendet.append(Aufgezeichnet({
|
|
"user_id": user.id, "titel": titel, "text": text, "kategorie": kategorie,
|
|
}))
|
|
return SimpleNamespace(zugestellt=["inapp"], fehlgeschlagen=[])
|
|
|
|
|
|
class StubAudit:
|
|
name = "audit-log"
|
|
|
|
def __init__(self):
|
|
self.eintraege: list[Aufgezeichnet] = []
|
|
|
|
async def log(self, actor, action, objekt_typ="", objekt_id=None,
|
|
details=None, **kwargs):
|
|
self.eintraege.append(Aufgezeichnet({
|
|
"actor": actor, "action": action, "objekt_typ": objekt_typ,
|
|
"objekt_id": objekt_id, "details": details,
|
|
}))
|
|
|
|
def log_sync(self, *args, **kwargs): # Kompatibilität zur echten API
|
|
import asyncio
|
|
|
|
asyncio.run(self.log(*args, **kwargs))
|
|
|
|
|
|
@pytest.fixture
|
|
def stubs(app):
|
|
"""Ersetzt dedup/benachrichtigung/audit-log im Registry-Zustand der App."""
|
|
modul = sys.modules[plugin_modul(app, "dedup").__name__ + ".pruefung"]
|
|
dedup_stub = StubDedup(modul)
|
|
benachrichtigung = StubBenachrichtigung()
|
|
audit = StubAudit()
|
|
registry_inhalt = app.state.registry._plugins
|
|
registry_inhalt["dedup"] = dedup_stub
|
|
registry_inhalt["benachrichtigung"] = benachrichtigung
|
|
registry_inhalt["audit-log"] = audit
|
|
return SimpleNamespace(
|
|
dedup=dedup_stub, benachrichtigung=benachrichtigung, audit=audit,
|
|
)
|
|
|
|
|
|
# ---------------- Migration & Zugriff ----------------
|
|
|
|
def test_migration_legt_planungsliste_an(app):
|
|
with app.state.engine.connect() as conn:
|
|
tabellen = {
|
|
zeile[0] for zeile in conn.execute(
|
|
text("SELECT name FROM sqlite_master WHERE type='table'")
|
|
)
|
|
}
|
|
protokoll = conn.execute(
|
|
text("SELECT version FROM schema_migrations WHERE plugin='planung'")
|
|
).fetchall()
|
|
assert "planungsliste" in tabellen
|
|
# 0002_bild_url: Coverbild-Spalte (seit der Cover-Erweiterung).
|
|
assert [z[0] for z in protokoll] == ["0001_planungsliste", "0002_bild_url"]
|
|
|
|
|
|
def test_anonym_wird_zum_login_umgeleitet(client):
|
|
antwort = client.get("/planung", follow_redirects=False)
|
|
assert antwort.status_code == 303
|
|
assert antwort.headers["location"].startswith("/login")
|
|
|
|
|
|
def test_rezensent_sieht_planungsliste(app, client):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
melde_an(client, "rez1", "test-12345678")
|
|
antwort = client.get("/planung")
|
|
assert antwort.status_code == 200
|
|
text_html = sichtbar(antwort)
|
|
assert "Planungsliste" in text_html
|
|
assert "Titel händisch nachtragen" in text_html
|
|
|
|
|
|
# ---------------- Weg 1: händisches Nachtragen ----------------
|
|
|
|
def test_manuell_ohne_treffer_legt_eintrag_an(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
melde_an(client, "rez1", "test-12345678")
|
|
|
|
antwort = client.post("/planung/neu", data={
|
|
"titel": "Ein ganz neues Spiel", "verlag": "Neu Verlag",
|
|
"ausgabe": "3/2025", "bgg_id": "",
|
|
}, follow_redirects=True)
|
|
|
|
assert antwort.status_code == 200
|
|
assert "wurde eingetragen" in antwort.text
|
|
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
|
with app.state.session_factory() as db:
|
|
eintrag = db.scalar(select(Planungseintrag))
|
|
assert eintrag.titel == "Ein ganz neues Spiel"
|
|
assert eintrag.verlag == "Neu Verlag"
|
|
assert eintrag.ausgabe == "3/2025"
|
|
assert eintrag.quelle == "manuell"
|
|
assert eintrag.status == "offen"
|
|
assert eintrag.pruefung is None
|
|
|
|
# Dedup wurde befragt:
|
|
assert len(stubs.dedup.aufrufe) == 1
|
|
assert stubs.dedup.aufrufe[0]["titel"] == "Ein ganz neues Spiel"
|
|
# Kein Treffer → keine Benachrichtigung, aber Audit-Eintrag:
|
|
assert stubs.benachrichtigung.gesendet == []
|
|
assert [e.daten["action"] for e in stubs.audit.eintraege] == ["erstellt"]
|
|
assert stubs.audit.eintraege[0].daten["objekt_typ"] == "planungseintrag"
|
|
|
|
|
|
def test_leerer_titel_wird_abgewiesen(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
melde_an(client, "rez1", "test-12345678")
|
|
antwort = client.post("/planung/neu", data={"titel": ""}, follow_redirects=True)
|
|
assert "Bitte einen Spieltitel angeben" in antwort.text
|
|
assert planung_zaehlen(app) == 0
|
|
assert stubs.dedup.aufrufe == []
|
|
|
|
|
|
def test_manuell_mit_treffer_benachrichtigt_und_auditiert(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
rez1 = benutzer_objekt(app, "rez1")
|
|
stubs.dedup.standard_treffer = True
|
|
stubs.dedup.treffer_einreihen(stubs.dedup.konflikt(
|
|
"planung", "„Spiel X“ ist bereits in der Planungsliste von Anna.",
|
|
details={"rezensent": "Anna"},
|
|
))
|
|
melde_an(client, "rez1", "test-12345678")
|
|
|
|
antwort = client.post("/planung/neu", data={
|
|
"titel": "Spiel X", "verlag": "", "ausgabe": "", "bgg_id": "",
|
|
}, follow_redirects=True)
|
|
|
|
assert antwort.status_code == 200
|
|
assert "Benachrichtigung erhalten" in antwort.text
|
|
# Benachrichtigung an den eintragenden Rezensenten:
|
|
assert len(stubs.benachrichtigung.gesendet) == 1
|
|
nachricht = stubs.benachrichtigung.gesendet[0].daten
|
|
assert nachricht["user_id"] == rez1.id
|
|
assert nachricht["kategorie"] == "dedup"
|
|
assert "Dedup-Prüfung" in nachricht["text"]
|
|
# Audit-Eintrag enthält den Befund:
|
|
erstellt = next(e for e in stubs.audit.eintraege if e.daten["action"] == "erstellt")
|
|
assert erstellt.daten["details"]["dedup_konflikte"][0]["art"] == "planung"
|
|
# Der Befund wird am Eintrag gespeichert:
|
|
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
|
with app.state.session_factory() as db:
|
|
eintrag = db.scalar(select(Planungseintrag))
|
|
assert eintrag.pruefung[0]["art"] == "planung"
|
|
|
|
|
|
def test_verlagskonflikt_oeffnet_auswahl_dialog(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
stubs.dedup.treffer_einreihen(stubs.dedup.konflikt(
|
|
"verlag", "Gleiches Spiel ist bereits unter anderem Verlag geführt.",
|
|
optionen=["Hans im Glück", "Kosmos"],
|
|
))
|
|
melde_an(client, "rez1", "test-12345678")
|
|
|
|
antwort = client.post("/planung/neu", data={
|
|
"titel": "Umstrittenes Spiel", "verlag": "Kosmos", "ausgabe": "", "bgg_id": "",
|
|
})
|
|
|
|
# Kein Redirect: der Auswahl-Dialog wird gerendert.
|
|
assert antwort.status_code == 200
|
|
assert "Welcher Verlag" in antwort.text
|
|
assert "Hans im Glück" in antwort.text and "Kosmos" in antwort.text
|
|
assert 'action="/planung/neu"' in antwort.text
|
|
assert planung_zaehlen(app) == 0 # erst nach der Wahl wird angelegt
|
|
|
|
|
|
def test_verlagswahl_legt_eintrag_mit_gewaehltem_verlag_an(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
stubs.dedup.treffer_einreihen(stubs.dedup.konflikt(
|
|
"verlag", "Gleiches Spiel ist bereits unter anderem Verlag geführt.",
|
|
optionen=["Hans im Glück", "Kosmos"],
|
|
))
|
|
stubs.dedup.treffer_einreihen(stubs.dedup.konflikt(
|
|
"verlag", "Gleiches Spiel ist bereits unter anderem Verlag geführt.",
|
|
optionen=["Hans im Glück", "Kosmos"],
|
|
))
|
|
melde_an(client, "rez1", "test-12345678")
|
|
|
|
# Schritt 1: Dialog erscheint.
|
|
dialog = client.post("/planung/neu", data={
|
|
"titel": "Umstrittenes Spiel", "verlag": "Kosmos", "ausgabe": "", "bgg_id": "",
|
|
})
|
|
assert "Welcher Verlag" in dialog.text
|
|
|
|
# Schritt 2: Bestätigte Verlagswahl legt den Eintrag an.
|
|
antwort = client.post("/planung/neu", data={
|
|
"titel": "Umstrittenes Spiel", "verlag": "Kosmos",
|
|
"ausgabe": "", "bgg_id": "",
|
|
"verlag_wahl": "Hans im Glück", "verlag_bestaetigt": "1",
|
|
}, follow_redirects=True)
|
|
assert antwort.status_code == 200
|
|
|
|
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
|
with app.state.session_factory() as db:
|
|
eintrag = db.scalar(select(Planungseintrag))
|
|
assert eintrag is not None
|
|
assert eintrag.verlag == "Hans im Glück"
|
|
# Entscheidung und Befund sind auditiert:
|
|
erstellt = next(e for e in stubs.audit.eintraege if e.daten["action"] == "erstellt")
|
|
assert erstellt.daten["details"]["verlag"] == "Hans im Glück"
|
|
assert erstellt.daten["details"]["dedup_konflikte"][0]["art"] == "verlag"
|
|
assert len(stubs.benachrichtigung.gesendet) == 1
|
|
|
|
|
|
# ---------------- Weg 2: Verschiebung aus den Neuheiten ----------------
|
|
|
|
def test_uebernehmen_verschiebt_neuheit_in_planung(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
neuheit_id = neuheit_anlegen(app, "Azul Next Move", verlag="Next Move", bgg_id=500)
|
|
melde_an(client, "rez1", "test-12345678")
|
|
|
|
antwort = client.post(f"/planung/uebernehmen/{neuheit_id}", follow_redirects=True)
|
|
assert antwort.status_code == 200
|
|
assert "Azul Next Move" in antwort.text
|
|
assert "übernommen" in antwort.text
|
|
|
|
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
|
with app.state.session_factory() as db:
|
|
eintrag = db.scalar(select(Planungseintrag))
|
|
status = db.execute(
|
|
text("SELECT status FROM neuheiten WHERE id=:id"), {"id": neuheit_id}
|
|
).scalar()
|
|
assert eintrag.quelle == "neuheiten"
|
|
assert eintrag.bgg_id == 500
|
|
assert eintrag.rezensent_id == benutzer_objekt(app, "rez1").id
|
|
assert status == "planung" # Neuheit gilt als verplant
|
|
|
|
# dedup wurde mit den Neuheiten-Daten befragt; Audit als „verschoben“:
|
|
assert stubs.dedup.aufrufe == [
|
|
{"titel": "Azul Next Move", "verlag": "Next Move", "bgg_id": 500,
|
|
"user_id": benutzer_objekt(app, "rez1").id}
|
|
]
|
|
assert [e.daten["action"] for e in stubs.audit.eintraege] == ["verschoben"]
|
|
assert stubs.benachrichtigung.gesendet == []
|
|
|
|
|
|
def test_uebernehmen_mit_verlagskonflikt_zeigt_dialog(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
neuheit_id = neuheit_anlegen(app, "Umstritten", verlag="Verlag A", bgg_id=600)
|
|
stubs.dedup.treffer_einreihen(stubs.dedup.konflikt(
|
|
"verlag", "Gleiches Spiel ist bereits unter anderem Verlag geführt.",
|
|
optionen=["Verlag A", "Verlag B"],
|
|
))
|
|
melde_an(client, "rez1", "test-12345678")
|
|
|
|
dialog = client.post(f"/planung/uebernehmen/{neuheit_id}")
|
|
assert dialog.status_code == 200
|
|
assert "Welcher Verlag" in dialog.text
|
|
assert 'action="/planung/uebernehmen/' in dialog.text
|
|
assert planung_zaehlen(app) == 0
|
|
|
|
antwort = client.post(f"/planung/uebernehmen/{neuheit_id}", data={
|
|
"verlag_wahl": "Verlag B", "verlag_bestaetigt": "1",
|
|
}, follow_redirects=True)
|
|
assert antwort.status_code == 200
|
|
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
|
with app.state.session_factory() as db:
|
|
eintrag = db.scalar(select(Planungseintrag))
|
|
status = db.execute(
|
|
text("SELECT status FROM neuheiten WHERE id=:id"), {"id": neuheit_id}
|
|
).scalar()
|
|
assert eintrag.verlag == "Verlag B"
|
|
assert status == "planung"
|
|
|
|
|
|
def test_doppelte_uebernahme_wird_abgeblockt(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
neuheit_id = neuheit_anlegen(app, "Azul", verlag="Next Move", bgg_id=500)
|
|
melde_an(client, "rez1", "test-12345678")
|
|
erste = client.post(f"/planung/uebernehmen/{neuheit_id}", follow_redirects=False)
|
|
assert erste.status_code == 303
|
|
|
|
zweite = client.post(f"/planung/uebernehmen/{neuheit_id}", follow_redirects=True)
|
|
assert "nicht mehr als Neuheit verfügbar" in zweite.text
|
|
assert planung_zaehlen(app) == 1
|
|
|
|
|
|
def test_unbekannte_neuheit_fuehrt_zur_fehlermeldung(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
melde_an(client, "rez1", "test-12345678")
|
|
antwort = client.post("/planung/uebernehmen/9999", follow_redirects=True)
|
|
assert "existiert nicht mehr" in antwort.text
|
|
assert planung_zaehlen(app) == 0
|
|
|
|
|
|
# ---------------- Statuswechsel ----------------
|
|
|
|
def test_statuswechsel_eigener_eintrag_wird_auditiert(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
rez1 = benutzer_objekt(app, "rez1")
|
|
eintrag_id = planung_anlegen(app, "Ark Nova", rez1.id)
|
|
melde_an(client, "rez1", "test-12345678")
|
|
|
|
antwort = client.post(f"/planung/{eintrag_id}/status",
|
|
data={"status": "in_bearbeitung"}, follow_redirects=True)
|
|
assert antwort.status_code == 200
|
|
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
|
with app.state.session_factory() as db:
|
|
eintrag = db.get(Planungseintrag, eintrag_id)
|
|
assert eintrag.status == "in_bearbeitung"
|
|
geaendert = stubs.audit.eintraege[-1].daten
|
|
assert geaendert["action"] == "geaendert"
|
|
assert geaendert["details"]["alt"]["status"] == "offen"
|
|
assert geaendert["details"]["neu"]["status"] == "in_bearbeitung"
|
|
|
|
|
|
def test_statuswechsel_fremder_eintrag_403(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
lege_benutzer_an(app, "rez2", "rezensent")
|
|
eintrag_id = planung_anlegen(app, "Ark Nova", benutzer_objekt(app, "rez1").id)
|
|
melde_an(client, "rez2", "test-12345678")
|
|
antwort = client.post(f"/planung/{eintrag_id}/status", data={"status": "abgeschlossen"})
|
|
assert antwort.status_code == 403
|
|
|
|
|
|
def test_redakteur_darf_status_fremder_eintraege_wechseln(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
lege_benutzer_an(app, "red1", "redakteur")
|
|
eintrag_id = planung_anlegen(app, "Ark Nova", benutzer_objekt(app, "rez1").id)
|
|
melde_an(client, "red1", "test-12345678")
|
|
antwort = client.post(f"/planung/{eintrag_id}/status",
|
|
data={"status": "abgeschlossen"}, follow_redirects=True)
|
|
assert antwort.status_code == 200
|
|
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
|
with app.state.session_factory() as db:
|
|
assert db.get(Planungseintrag, eintrag_id).status == "abgeschlossen"
|
|
|
|
|
|
def test_unbekannter_status_wird_abgelehnt(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
eintrag_id = planung_anlegen(app, "Ark Nova", benutzer_objekt(app, "rez1").id)
|
|
melde_an(client, "rez1", "test-12345678")
|
|
antwort = client.post(f"/planung/{eintrag_id}/status",
|
|
data={"status": "quatsch"}, follow_redirects=True)
|
|
assert "Unbekannter Status" in antwort.text
|
|
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
|
with app.state.session_factory() as db:
|
|
assert db.get(Planungseintrag, eintrag_id).status == "offen"
|
|
|
|
|
|
# ---------------- Bearbeiten & Löschen ----------------
|
|
|
|
def test_eigener_eintrag_kann_bearbeitet_werden(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
eintrag_id = planung_anlegen(app, "Ark Nova", benutzer_objekt(app, "rez1").id,
|
|
ausgabe="")
|
|
melde_an(client, "rez1", "test-12345678")
|
|
|
|
formular = client.get(f"/planung/{eintrag_id}/bearbeiten")
|
|
assert formular.status_code == 200
|
|
assert "Ark Nova" in formular.text
|
|
|
|
antwort = client.post(f"/planung/{eintrag_id}/bearbeiten", data={
|
|
"titel": "Ark Nova", "verlag": "Feuerland Spiele",
|
|
"ausgabe": "4/2025", "notizen": "Schwerpunkt-Thema",
|
|
}, follow_redirects=True)
|
|
assert antwort.status_code == 200
|
|
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
|
with app.state.session_factory() as db:
|
|
eintrag = db.get(Planungseintrag, eintrag_id)
|
|
assert eintrag.ausgabe == "4/2025"
|
|
assert eintrag.notizen == "Schwerpunkt-Thema"
|
|
geaendert = next(
|
|
e for e in stubs.audit.eintraege if e.daten["action"] == "geaendert"
|
|
).daten
|
|
assert "alt" in geaendert["details"] and "neu" in geaendert["details"]
|
|
|
|
|
|
def test_fremder_eintrag_nicht_bearbeitbar_fuer_rezensenten(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
lege_benutzer_an(app, "rez2", "rezensent")
|
|
eintrag_id = planung_anlegen(app, "Ark Nova", benutzer_objekt(app, "rez1").id)
|
|
melde_an(client, "rez2", "test-12345678")
|
|
assert client.get(f"/planung/{eintrag_id}/bearbeiten").status_code == 403
|
|
assert client.post(f"/planung/{eintrag_id}/bearbeiten", data={
|
|
"titel": "Hack", "verlag": "", "ausgabe": "", "notizen": "",
|
|
}).status_code == 403
|
|
|
|
|
|
def test_eigener_eintrag_kann_geloescht_werden(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
eintrag_id = planung_anlegen(app, "Weg von hier", benutzer_objekt(app, "rez1").id)
|
|
melde_an(client, "rez1", "test-12345678")
|
|
|
|
antwort = client.post(f"/planung/{eintrag_id}/loeschen", follow_redirects=True)
|
|
assert antwort.status_code == 200
|
|
assert "wurde gelöscht" in antwort.text
|
|
assert planung_zaehlen(app) == 0
|
|
geloescht = next(
|
|
e for e in stubs.audit.eintraege if e.daten["action"] == "geloescht"
|
|
).daten
|
|
assert geloescht["details"]["titel"] == "Weg von hier"
|
|
|
|
|
|
def test_fremder_eintrag_nicht_loeschbar_fuer_rezensenten(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
lege_benutzer_an(app, "rez2", "rezensent")
|
|
eintrag_id = planung_anlegen(app, "Ark Nova", benutzer_objekt(app, "rez1").id)
|
|
melde_an(client, "rez2", "test-12345678")
|
|
antwort = client.post(f"/planung/{eintrag_id}/loeschen")
|
|
assert antwort.status_code == 403
|
|
assert planung_zaehlen(app) == 1
|
|
|
|
|
|
# ---------------- Liste / Zuordnung / Rechte im UI ----------------
|
|
|
|
def test_liste_zeigt_zuordnung_und_rechte(client, app, stubs):
|
|
lege_benutzer_an(app, "rez1", "rezensent", password="pw-rez1-123456")
|
|
lege_benutzer_an(app, "rez2", "rezensent", password="pw-rez2-123456")
|
|
own_id = planung_anlegen(app, "Mein Spiel", benutzer_objekt(app, "rez1").id)
|
|
fremd_id = planung_anlegen(app, "Fremdes Spiel", benutzer_objekt(app, "rez2").id)
|
|
|
|
melde_an(client, "rez1", "pw-rez1-123456")
|
|
antwort = client.get("/planung")
|
|
text_html = antwort.text
|
|
assert "Mein Spiel" in text_html and "Fremdes Spiel" in text_html
|
|
# Eigene Aktionen sichtbar, fremde nicht:
|
|
assert f"/planung/{own_id}/bearbeiten" in text_html
|
|
assert f"/planung/{fremd_id}/bearbeiten" not in text_html
|
|
|
|
|
|
# ---------------- Integration mit den echten Plugins ----------------
|
|
|
|
def test_integration_echte_plugins_benachrichtigung_und_audit(settings):
|
|
"""Ohne Stubs: Treffer löst echte In-App-Nachricht + Audit-Eintrag aus."""
|
|
app = create_app(settings)
|
|
lege_benutzer_an(app, "rez1", "rezensent")
|
|
lege_benutzer_an(app, "rez2", "rezensent")
|
|
planung_anlegen(app, "Azul", benutzer_objekt(app, "rez2").id) # fremd → Treffer
|
|
neuheit_id = neuheit_anlegen(app, "Azul", verlag="Next Move", bgg_id=777)
|
|
|
|
client = TestClient(app)
|
|
melde_an(client, "rez1", "test-12345678")
|
|
antwort = client.post(f"/planung/uebernehmen/{neuheit_id}", follow_redirects=True)
|
|
assert antwort.status_code == 200
|
|
|
|
# Echte In-App-Benachrichtigung für rez1:
|
|
benachrichtigungen = plugin_modul(app, "benachrichtigung").Benachrichtigung
|
|
with app.state.session_factory() as db:
|
|
nachrichten = db.scalars(select(benachrichtigungen)).all()
|
|
audit_zeilen = db.scalars(
|
|
select(plugin_modul(app, "audit-log").AuditEintrag)
|
|
).all()
|
|
assert any(n.user_id == benutzer_objekt(app, "rez1").id and "Azul" in n.titel
|
|
for n in nachrichten)
|
|
assert any(a.action == "verschoben" and a.objekt_typ == "planungseintrag"
|
|
for a in audit_zeilen)
|
|
|
|
# Und die echte dedup-Prüfung hat ihren Protokolleintrag bekommen:
|
|
pruefungen = plugin_modul(app, "dedup").Pruefung
|
|
with app.state.session_factory() as db:
|
|
protokoll = db.scalars(select(pruefungen)).all()
|
|
assert protokoll and protokoll[0].konflikte
|