Plugins dedup + planung: Prüf-API, Verzahnung mit Benachrichtigung und Audit-Log
Plugin dedup: - Öffentliche Prüf-API check_titel(titel, verlag, bgg_id, user) mit strukturiertem PruefErgebnis (Konflikte, Verlagsauswahl, Titel-Empfehlung) - Vier Prüfungen: Verlags-Konflikt, Titel-Varianten (deutsche Version bevorzugt: Umlaut-/Wort-Heuristik + BGG Alternate-Names), Spiel oder Vorgänger bereits besprochen (BGG boardgameexpansion-Relation + rapidfuzz-Fuzzy-Match), Titel in fremder Planungsliste - BGG-Hilfsclient mit Rate-Limit, netzwerkfrei testbar, degradiert defensiv - Eigene Migration (Prüfprotokoll dedup_pruefungen) + Prüfseite /dedup - Fremde Plugin-Tabellen werden nur lesend über die gemeinsame Metadata gelesen — keine Import-Abhängigkeiten zwischen Plugins Plugin planung: - Migration planungsliste: Titel, Verlag, Ausgabe, Rezensent, Status (offen/in_bearbeitung/abgeschlossen), Quelle, Prüf-Befund (JSON) - Beide Eintragswege durch die dedup-Prüfung: Verschiebung aus den Neuheiten (Button „→ Zur Planung“, Neuheit wechselt auf Status planung) und händisches Nachtragen im Formular - Bei Treffern: send_notification an den eintragenden Rezensenten (benachrichtigung-Plugin) + Audit-Log-Eintrag (audit-log-Plugin) - Verlags-Konflikt → Auswahl-Dialog „Welcher Verlag wird geführt?“, Entscheidung wird auditiert - Planungsliste mit Zuordnung, Statuswechsel, Bearbeiten/Löschen; Rezensenten nur eigene Einträge, Admin/Redakteur alle Tests: 45 neue Tests (Heuristik, alle vier Prüfungen mit Fake-BGG-Client, Protokoll, beide Eintragswege mit Stub-gemockten Abhängigkeiten, Rollen, Integration mit echten Plugins) — 141 Tests grün.
This commit is contained in:
Flo Hartmann
@@ -22,6 +22,8 @@ def _hintergrundjobs_deaktiviert(monkeypatch):
|
||||
monkeypatch.setenv("SPIELE_BGG_SYNC_AKTIV", "1").
|
||||
"""
|
||||
monkeypatch.setenv("SPIELE_BGG_SYNC_AKTIV", "0")
|
||||
# Auch der BGG-Hilfsclient des dedup-Plugins bleibt in Tests offline.
|
||||
monkeypatch.setenv("SPIELE_DEDUP_BGG_AKTIV", "0")
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
391
tests/test_dedup.py
Normal file
391
tests/test_dedup.py
Normal file
@@ -0,0 +1,391 @@
|
||||
"""Tests: Plugin „dedup“ — Heuristik, vier Prüfungen, Protokoll, Prüfseite.
|
||||
|
||||
Alle Prüfungen laufen gegen die echte SQLite-Testdatenbank; der optionale
|
||||
BGG-Hilfsclient wird durch Fakes ersetzt (keine Netzwerk-Aufrufe).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import sys
|
||||
import time
|
||||
|
||||
import html
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import func, select, text
|
||||
|
||||
from tests.conftest import lege_benutzer_an, melde_an
|
||||
|
||||
|
||||
# ---------------- Hilfen ----------------
|
||||
|
||||
def plugin_modul(app, name: str):
|
||||
plugin_instanz = app.state.registry.get(name)
|
||||
return sys.modules[type(plugin_instanz).__module__]
|
||||
|
||||
|
||||
def pruef_modul(app):
|
||||
"""Das pruefung-Submodul (Heuristik-Funktionen) des dedup-Plugins."""
|
||||
return sys.modules[plugin_modul(app, "dedup").__name__ + ".pruefung"]
|
||||
|
||||
|
||||
def sichtbar(antwort) -> str:
|
||||
"""Response-Text mit aufgelösten HTML-Entities (für Umlaut-Assertions)."""
|
||||
return html.unescape(antwort.text)
|
||||
|
||||
|
||||
def dedup_plugin(app):
|
||||
return app.state.registry.get("dedup")
|
||||
|
||||
|
||||
def pruefe(app, titel, verlag=None, bgg_id=None, *, user=None, hilfsclient=None):
|
||||
"""Führt eine check_titel-Prüfung aus (optional mit Fake-BGG-Client)."""
|
||||
if hilfsclient is not None:
|
||||
dedup_plugin(app).bgg_client_fabrik = lambda: hilfsclient
|
||||
try:
|
||||
return asyncio.run(
|
||||
dedup_plugin(app).check_titel(titel, verlag, bgg_id, user=user)
|
||||
)
|
||||
finally:
|
||||
dedup_plugin(app).bgg_client_fabrik = None
|
||||
|
||||
|
||||
def neuheit_anlegen(app, titel, verlag=None, bgg_id=1, status="neuheit"):
|
||||
Neuheit = plugin_modul(app, "neuheiten").Neuheit
|
||||
with app.state.session_factory() as db:
|
||||
db.add(Neuheit(titel=titel, verlag=verlag, bgg_id=bgg_id, status=status))
|
||||
db.commit()
|
||||
|
||||
|
||||
def planung_anlegen(app, titel, rezensent_id, status="offen", verlag=None, bgg_id=None):
|
||||
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
||||
with app.state.session_factory() as db:
|
||||
eintrag = Planungseintrag(
|
||||
titel=titel, rezensent_id=rezensent_id, status=status,
|
||||
verlag=verlag, bgg_id=bgg_id,
|
||||
)
|
||||
db.add(eintrag)
|
||||
db.commit()
|
||||
return eintrag.id
|
||||
|
||||
|
||||
def benutzer_id(app, username: str) -> int:
|
||||
from redaktionskern.auth.models import User
|
||||
|
||||
with app.state.session_factory() as db:
|
||||
return db.scalar(select(User.id).where(User.username == username))
|
||||
|
||||
|
||||
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 konflikt_arten(ergebnis) -> list[str]:
|
||||
return [k.art for k in ergebnis.konflikte]
|
||||
|
||||
|
||||
class FakeBgg:
|
||||
"""Ersetzt den BGG-Hilfsclient: feste Antworten, keine Netzwerk-Aufrufe."""
|
||||
|
||||
def __init__(self, alternativen_map=None, vorgaenger_map=None):
|
||||
self._alternativen = alternativen_map or {}
|
||||
self._vorgaenger = vorgaenger_map or {}
|
||||
self.alternativen_aufrufe: list[int] = []
|
||||
self.vorgaenger_aufrufe: list[int] = []
|
||||
|
||||
def alternativen(self, bgg_id):
|
||||
self.alternativen_aufrufe.append(bgg_id)
|
||||
return list(self._alternativen.get(bgg_id, []))
|
||||
|
||||
def vorgaenger(self, bgg_id):
|
||||
self.vorgaenger_aufrufe.append(bgg_id)
|
||||
return list(self._vorgaenger.get(bgg_id, []))
|
||||
|
||||
def schliessen(self):
|
||||
pass
|
||||
|
||||
|
||||
# ---------------- Titel-Heuristik ----------------
|
||||
|
||||
def test_deutsche_titel_heuristik(app):
|
||||
m = pruef_modul(app)
|
||||
assert m.ist_deutscher_titel("Die Siedler von Catan")
|
||||
assert m.ist_deutscher_titel("Ein Fest für Odin")
|
||||
assert m.ist_deutscher_titel("Bärenpark")
|
||||
assert m.ist_deutscher_titel("Das verrückte Labyrinth")
|
||||
assert not m.ist_deutscher_titel("The Crew")
|
||||
assert not m.ist_deutscher_titel("Terra Mystica") # neutral → kein Deutsch-Beleg
|
||||
|
||||
|
||||
def test_fuzzy_match_erkennt_titel_varianten(app):
|
||||
m = pruef_modul(app)
|
||||
assert m.titel_aehnlich("Die Siedler von Catan", "Siedler von Catan, Die")
|
||||
assert m.titel_aehnlich("Cascadia", "cascadia")
|
||||
assert not m.titel_aehnlich("Cascadia", "Everdell")
|
||||
assert not m.titel_aehnlich("", "Cascadia")
|
||||
|
||||
|
||||
def test_bevorzuge_deutschen_titel_waehlt_hoechsten_score(app):
|
||||
m = pruef_modul(app)
|
||||
kandidaten = ["The Settlers of Catan", "Catan", "Die Siedler von Catan"]
|
||||
assert m.bevorzuge_deutschen_titel(kandidaten) == "Die Siedler von Catan"
|
||||
# Ein deutscher Kandidat schlägt einen neutralen Eingangstitel:
|
||||
assert m.bevorzuge_deutschen_titel(["Terra Mystica", "Bärenpark"]) == "Bärenpark"
|
||||
|
||||
|
||||
# ---------------- Migration & Grundlagen ----------------
|
||||
|
||||
def test_migration_legt_protokolltabelle_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='dedup'")
|
||||
).fetchall()
|
||||
assert "dedup_pruefungen" in tabellen
|
||||
assert [z[0] for z in protokoll] == ["0001_pruefprotokoll"]
|
||||
|
||||
|
||||
def test_leere_tabellen_liefern_keine_konflikte(app):
|
||||
ergebnis = pruefe(app, "Ganz neues Spiel", "Irgendein Verlag", 4711)
|
||||
assert not ergebnis.hat_konflikte
|
||||
assert ergebnis.titel == "Ganz neues Spiel"
|
||||
assert ergebnis.bgg_id == 4711
|
||||
|
||||
|
||||
# ---------------- Prüfung a) Verlags-Konflikt ----------------
|
||||
|
||||
def test_verlagskonflikt_gegen_neuheiten_eintrag(app):
|
||||
neuheit_anlegen(app, "Stone Age", verlag="Hans im Glück", bgg_id=21)
|
||||
ergebnis = pruefe(app, "Stone Age", "Kosmos", 21)
|
||||
assert ergebnis.hat_verlagskonflikt
|
||||
assert "Hans im Glück" in ergebnis.verlags_optionen
|
||||
assert "Kosmos" in ergebnis.verlags_optionen
|
||||
assert any(k.art == "verlag" for k in ergebnis.konflikte)
|
||||
|
||||
|
||||
def test_kein_verlagskonflikt_bei_identischem_verlag(app):
|
||||
neuheit_anlegen(app, "Stone Age", verlag="Kosmos", bgg_id=21)
|
||||
ergebnis = pruefe(app, "Stone Age", "kosmos ", 21) # Groß-/Kleinschreibung egal
|
||||
assert not ergebnis.hat_konflikte
|
||||
|
||||
|
||||
def test_verlagskonflikt_ohne_bgg_id_ueber_fuzzy_match(app):
|
||||
neuheit_anlegen(app, "Cascadia", verlag="Lookout Spiele")
|
||||
ergebnis = pruefe(app, "Cascadia", "Feuerland Spiele", None)
|
||||
assert ergebnis.hat_verlagskonflikt
|
||||
assert "Lookout Spiele" in ergebnis.verlags_optionen
|
||||
|
||||
|
||||
# ---------------- Prüfung b) Titel-Varianten / deutsche Version ----------------
|
||||
|
||||
def test_deutsche_version_wird_empfohlen(app):
|
||||
neuheit_anlegen(app, "Die Siedler von Catan", verlag="Kosmos", bgg_id=13)
|
||||
ergebnis = pruefe(app, "Settlers of Catan", "Kosmos", 13)
|
||||
assert ergebnis.titel_empfehlung == "Die Siedler von Catan"
|
||||
assert any(
|
||||
k.art == "titel" and "empfohlen" in k.beschreibung.lower()
|
||||
for k in ergebnis.konflikte
|
||||
)
|
||||
|
||||
|
||||
def test_eigener_deutscher_titel_erzeugt_keinen_titelkonflikt(app):
|
||||
neuheit_anlegen(app, "Catan", verlag="Kosmos", bgg_id=13)
|
||||
ergebnis = pruefe(app, "Die Siedler von Catan", "Kosmos", 13)
|
||||
assert ergebnis.titel_empfehlung == "Die Siedler von Catan"
|
||||
assert "titel" not in konflikt_arten(ergebnis)
|
||||
|
||||
|
||||
def test_bgg_alternativen_fliessen_in_die_empfehlung_ein(app):
|
||||
neuheit_anlegen(app, "Catan", verlag="Kosmos", bgg_id=13)
|
||||
fake = FakeBgg(alternativen_map={
|
||||
13: ["The Settlers of Catan", "Die Siedler von Catan"],
|
||||
})
|
||||
ergebnis = pruefe(app, "Settlers of Catan", "Kosmos", 13, hilfsclient=fake)
|
||||
assert fake.alternativen_aufrufe == [13]
|
||||
assert ergebnis.titel_empfehlung == "Die Siedler von Catan"
|
||||
|
||||
|
||||
# ---------------- Prüfung c) Bereits besprochen / Vorgänger ----------------
|
||||
|
||||
def test_abgeschlossenes_spiel_gilt_als_besprochen(app):
|
||||
lege_benutzer_an(app, "rez1", "rezensent")
|
||||
lege_benutzer_an(app, "rez2", "rezensent")
|
||||
rez1 = benutzer_id(app, "rez1")
|
||||
planung_anlegen(app, "Cascadia", rez1, status="abgeschlossen")
|
||||
|
||||
ergebnis = pruefe(app, "Cascadia") # ohne Benutzerkontext (z. B. Import-Prüfung)
|
||||
assert "besprochen" in konflikt_arten(ergebnis)
|
||||
besprochen = next(k for k in ergebnis.konflikte if k.art == "besprochen")
|
||||
assert "rez1" in besprochen.details["rezensent"] or "Rez1" in besprochen.details["rezensent"]
|
||||
|
||||
|
||||
def test_vorgaenger_wird_ueber_bgg_relation_erkannt(app):
|
||||
lege_benutzer_an(app, "rez1", "rezensent")
|
||||
rez1 = benutzer_id(app, "rez1")
|
||||
planung_anlegen(app, "Catan Grundspiel", rez1, status="abgeschlossen", bgg_id=13)
|
||||
fake = FakeBgg(vorgaenger_map={99: [(13, "Catan Grundspiel")]})
|
||||
|
||||
ergebnis = pruefe(app, "Catan Erweiterung", None, 99, hilfsclient=fake)
|
||||
besprochen = [k for k in ergebnis.konflikte if k.art == "besprochen"]
|
||||
assert besprochen
|
||||
assert any("Vorgänger" in k.beschreibung for k in besprochen)
|
||||
|
||||
|
||||
def test_vorgaenger_wird_zusaetzlich_fuzzy_gematcht(app):
|
||||
lege_benutzer_an(app, "rez1", "rezensent")
|
||||
rez1 = benutzer_id(app, "rez1")
|
||||
planung_anlegen(app, "Catan Grundspiel", rez1, status="abgeschlossen")
|
||||
fake = FakeBgg(vorgaenger_map={99: [(555, "Katan Grundspiel")]})
|
||||
|
||||
ergebnis = pruefe(app, "Catan Erweiterung", None, 99, hilfsclient=fake)
|
||||
assert any(
|
||||
k.art == "besprochen" and "Vorgänger" in k.beschreibung
|
||||
for k in ergebnis.konflikte
|
||||
)
|
||||
|
||||
|
||||
# ---------------- Prüfung d) Fremde Planungsliste ----------------
|
||||
|
||||
def test_fremder_planungseintrag_loest_konflikt_aus(app):
|
||||
lege_benutzer_an(app, "rez1", "rezensent")
|
||||
lege_benutzer_an(app, "rez2", "rezensent")
|
||||
rez1 = benutzer_objekt(app, "rez1")
|
||||
rez2 = benutzer_objekt(app, "rez2")
|
||||
planung_anlegen(app, "Ark Nova", rez1.id)
|
||||
|
||||
ergebnis = pruefe(app, "Ark Nova", user=rez2)
|
||||
assert "planung" in konflikt_arten(ergebnis)
|
||||
planungs_konflikt = next(k for k in ergebnis.konflikte if k.art == "planung")
|
||||
assert rez1.display_name in planungs_konflikt.details["rezensent"]
|
||||
|
||||
# Der eigene Eintrag des gleichen Rezensenten ist kein Konflikt:
|
||||
ergebnis = pruefe(app, "Ark Nova", user=rez1)
|
||||
assert "planung" not in konflikt_arten(ergebnis)
|
||||
|
||||
|
||||
# ---------------- Protokoll ----------------
|
||||
|
||||
def test_pruefung_wird_im_eigenen_protokoll_gespeichert(app, client):
|
||||
lege_benutzer_an(app, "anna", "redakteur")
|
||||
with app.state.session_factory() as db:
|
||||
from redaktionskern.auth.models import User
|
||||
anna = db.scalar(select(User).where(User.username == "anna"))
|
||||
|
||||
neuheit_anlegen(app, "Cascadia", verlag="Lookout Spiele")
|
||||
ergebnis = pruefe(app, "Cascadia", "Anderer Verlag", None, user=anna)
|
||||
assert ergebnis.hat_konflikte
|
||||
|
||||
Pruefung = plugin_modul(app, "dedup").Pruefung
|
||||
with app.state.session_factory() as db:
|
||||
eintrag = db.scalar(select(Pruefung))
|
||||
assert eintrag is not None
|
||||
assert eintrag.titel == "Cascadia"
|
||||
assert eintrag.geprueft_durch == "anna"
|
||||
assert eintrag.konflikte and eintrag.konflikte[0]["art"] == "verlag"
|
||||
|
||||
|
||||
# ---------------- Prüfseite ----------------
|
||||
|
||||
def test_dedup_seite_erreichbar_nach_login(client):
|
||||
melde_an(client)
|
||||
antwort = client.get("/dedup")
|
||||
assert antwort.status_code == 200
|
||||
assert "Titel prüfen" in sichtbar(antwort)
|
||||
assert "Verlags-Konflikt" in antwort.text
|
||||
|
||||
|
||||
def test_dedup_seite_anonym_wird_umgeleitet(client):
|
||||
antwort = client.get("/dedup", follow_redirects=False)
|
||||
assert antwort.status_code == 303
|
||||
assert antwort.headers["location"].startswith("/login")
|
||||
|
||||
|
||||
def test_manuelle_pruefung_ueber_formular(client):
|
||||
melde_an(client)
|
||||
antwort = client.post("/dedup/pruefen", data={
|
||||
"titel": "Ganz neues Spiel", "verlag": "", "bgg_id": "",
|
||||
})
|
||||
assert antwort.status_code == 200
|
||||
assert "Keine Treffer" in sichtbar(antwort)
|
||||
|
||||
|
||||
def test_manuelle_pruefung_zeigt_treffer(client, app):
|
||||
neuheit_anlegen(app, "Cascadia", verlag="Lookout Spiele")
|
||||
melde_an(client)
|
||||
antwort = client.post("/dedup/pruefen", data={
|
||||
"titel": "Cascadia", "verlag": "Feuerland", "bgg_id": "",
|
||||
})
|
||||
assert antwort.status_code == 200
|
||||
text_html = sichtbar(antwort)
|
||||
assert "Verlags-Konflikt" in text_html
|
||||
assert "Lookout Spiele" in text_html
|
||||
|
||||
|
||||
def test_manuelle_pruefung_validiert_eingaben(client):
|
||||
melde_an(client)
|
||||
leer = client.post("/dedup/pruefen", data={"titel": "", "verlag": "", "bgg_id": ""})
|
||||
assert "Bitte einen Spieltitel angeben" in leer.text
|
||||
kaputt = client.post("/dedup/pruefen", data={"titel": "X", "verlag": "", "bgg_id": "abc"})
|
||||
assert "Zahl sein" in kaputt.text
|
||||
|
||||
|
||||
# ---------------- BGG-Hilfsclient (Parsing, netzwerkfrei) ----------------
|
||||
|
||||
THING_XML = """<?xml version="1.0" encoding="utf-8"?>
|
||||
<items total="1" termsofuse="https://boardgamegeek.com/xmlapi/termsofuse">
|
||||
<item type="boardgame" id="99">
|
||||
<name type="primary" sortindex="1" value="Catan Erweiterung"/>
|
||||
<name type="alternate" sortindex="1" value="Catan Expansion"/>
|
||||
<name type="alternate" sortindex="5" value="Siedler-Erweiterung"/>
|
||||
<link type="boardgamecategory" id="1026" value="Settlement"/>
|
||||
<link type="boardgameexpansion" id="13" value="Catan Grundspiel" inbound="false"/>
|
||||
</item>
|
||||
</items>
|
||||
"""
|
||||
|
||||
|
||||
def test_bgg_pruefclient_parst_alternativen_und_vorgaenger():
|
||||
from plugins.dedup.bgg import BggPruefClient
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
assert request.url.params["id"] == "99"
|
||||
return httpx.Response(200, content=THING_XML.encode())
|
||||
|
||||
schlaf_aufrufe: list[float] = []
|
||||
client = BggPruefClient(
|
||||
mindestabstand_sekunden=0.0,
|
||||
transport=httpx.MockTransport(handler),
|
||||
uhr=time.monotonic,
|
||||
schlafen=lambda s: schlaf_aufrufe.append(s),
|
||||
)
|
||||
try:
|
||||
alternativen = client.alternativen(99)
|
||||
vorgaenger = client.vorgaenger(99)
|
||||
finally:
|
||||
client.schliessen()
|
||||
assert alternativen == ["Catan Expansion", "Siedler-Erweiterung"]
|
||||
assert vorgaenger == [(13, "Catan Grundspiel")]
|
||||
assert schlaf_aufrufe == []
|
||||
|
||||
|
||||
def test_bgg_pruefclient_toleroert_fehler():
|
||||
from plugins.dedup.bgg import BggPruefClient
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
raise httpx.ConnectError("offline")
|
||||
|
||||
client = BggPruefClient(mindestabstand_sekunden=0.0, transport=httpx.MockTransport(handler))
|
||||
try:
|
||||
assert client.alternativen(99) == []
|
||||
assert client.vorgaenger(99) == []
|
||||
finally:
|
||||
client.schliessen()
|
||||
575
tests/test_planung.py
Normal file
575
tests/test_planung.py
Normal file
@@ -0,0 +1,575 @@
|
||||
"""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
|
||||
assert [z[0] for z in protokoll] == ["0001_planungsliste"]
|
||||
|
||||
|
||||
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
|
||||
@@ -18,7 +18,7 @@ ERWARTETE_STUBS = {
|
||||
}
|
||||
|
||||
# Voll implementierte Plugins (keine Stubs mehr):
|
||||
ERWARTETE_PLUGINS = ERWARTETE_STUBS | {"neuheiten"}
|
||||
ERWARTETE_PLUGINS = ERWARTETE_STUBS | {"neuheiten", "dedup", "planung"}
|
||||
|
||||
|
||||
def test_alle_plugins_werden_geladen(app):
|
||||
@@ -41,7 +41,7 @@ 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"}
|
||||
VOLL_IMPLEMENTIERT = {"audit-log", "neuheiten", "benachrichtigung", "dedup", "planung"}
|
||||
PLATZHALTER_STUBS = ERWARTETE_STUBS - VOLL_IMPLEMENTIERT
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user