414 lines
15 KiB
Python
414 lines
15 KiB
Python
"""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()
|
|
|
|
|
|
def test_bgg_pruefclient_sendet_token_header():
|
|
from plugins.dedup.bgg import BggPruefClient
|
|
|
|
gesehen: list[httpx.Request] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
gesehen.append(request)
|
|
return httpx.Response(200, content=THING_XML.encode())
|
|
|
|
client = BggPruefClient(
|
|
mindestabstand_sekunden=0.0,
|
|
token="mein-token",
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
try:
|
|
assert client.alternativen(99) == ["Catan Expansion", "Siedler-Erweiterung"]
|
|
finally:
|
|
client.schliessen()
|
|
assert gesehen
|
|
assert gesehen[0].headers["Authorization"] == "Bearer mein-token"
|