Plugin neuheiten: BGG-Sync mit APScheduler, Filtern und deutscher UI

- Datenmodell + eigene Migration 0001_neuheiten_tabelle (Tabelle neuheiten: Titel, Verlag, Autor, Erscheinungsjahr, BGG-ID, Status 'neuheit', Quelle, Zeitstempel; bgg_id eindeutig als Merge-Kriterium) - BoardGameGeek XML API2-Client (search + thing, Batches à 20 IDs): Rate-Limit >= 1 s zwischen Requests, Retry mit exponentiellem Backoff bei 5xx/429/Netzwerkfehlern, HTTP 202 gemäß Retry-After, robustes XML-Parsing; Transport/Uhr/Sleep injizierbar (keine echten Calls in Tests) - Filter: Erweiterungen (boardgameexpansion) auf Request- und Elementebene ausgeschlossen; Prototypen per Titel-Heuristik (BGG hat keinen Marker) - Sync-Service mit Update-statt-Duplikat-Logik über die eindeutige BGG-ID (Status bleibt erhalten); Fehler je Suchbegriff brechen den Lauf nicht ab - APScheduler-Hintergrundjob (Standard 24 h) mit Überlappungsschutz, abschaltbar/intervallkonfigurierbar per Env; manueller 'Jetzt synchronisieren'-Endpunkt nur für Admin/Redakteur, optional mit Sofort-Suchbegriff - UI /neuheiten: sortier-/filterbare Tabelle mit Volltextsuche (HTMX-Teilladung, noscript-fähig), deutsche Oberfläche, BGG-Links, Ergebnis-Banner - Plugin-Loader: idempotentes Laden (Modul-Caching), damit mehrere create_app()-Aufrufe dieselben Plugin-Klassen/Tabellen nutzen - Tests: Parsing, Erweiterungs-/Prototyp-Filter, Rate-Limit/Backoff/202, Update-statt-Duplikat, Rollen am Sync-Endpunkt, Scheduler-Lifecycle — ausschließlich mit gemockten BGG-Antworten (uv run pytest: 96 grün) - README/AGENTS: Plugin-Doku, Env-Variablen, Fortschrittstabelle aktualisiert
2026-08-21 19:10:43 +00:00
parent 83d574d95c
commit 56cc5943b1
16 changed files with 1732 additions and 28 deletions
--- a/tests/test_neuheiten_ui.py
+++ b/tests/test_neuheiten_ui.py
@@ -0,0 +1,241 @@
+"""Tests für die Neuheiten-UI: Rollen, Liste, Suche/Filter/Sortierung, Sync-Button.
+
+Alle BGG-Antworten sind gemockt (FakeBggClient), kein echter Netzwerkverkehr.
+"""
+from __future__ import annotations
+
+from fastapi.testclient import TestClient
+
+from redaktionskern.app import create_app
+
+from tests._neuheiten import BggSpiel, Neuheit, SuchTreffer
+
+from tests.conftest import ADMIN_PASSWORD, lege_benutzer_an, melde_an
+
+
+class FakeBggClient:
+    """Achtung: Attribut heißt bewusst NICHT `details`, sonst würde es die
+    gleichnamige Methode verdecken."""
+
+    def __init__(self, treffer=None, spiel_details=None):
+        self.treffer = treffer or []
+        self.spiel_details = spiel_details or []
+        self.suche_aufrufe: list[str] = []
+
+    def suche(self, suchbegriff: str, *, max_treffer: int | None = None):
+        self.suche_aufrufe.append(suchbegriff)
+        return list(self.treffer)
+
+    def details(self, bgg_ids):
+        return list(self.spiel_details)
+
+    def schliessen(self) -> None:
+        pass
+
+
+def _client_mit_fakes(monkeypatch, client: TestClient, fake: FakeBggClient) -> None:
+    plugin = client.app.state.registry.get("neuheiten")
+    # Instanz-Methode ersetzen; monkeypatch stellt nach dem Test wieder her
+    monkeypatch.setattr(plugin, "_neuer_client", lambda: fake)
+
+
+def test_neuheiten_erfordert_anmeldung(client):
+    antwort = client.get("/neuheiten", follow_redirects=False)
+    assert antwort.status_code == 303
+    assert "/login" in antwort.headers["location"]
+
+
+def test_liste_zeigt_eintraege_und_deutsche_beschriftung(app, client):
+    with app.state.session_factory() as db:
+        db.add(
+            Neuheit(
+                titel="Catan",
+                verlag="KOSMOS",
+                autor="Klaus Teuber",
+                erscheinungsjahr=1995,
+                bgg_id=13,
+            )
+        )
+        db.commit()
+
+    melde_an(client)
+    antwort = client.get("/neuheiten")
+
+    assert antwort.status_code == 200
+    for text in (
+        "Spieltitel", "Verlag", "Autor", "Erscheinungsjahr", "Status",
+        "Catan", "KOSMOS", "Klaus Teuber", "1995", "1 Eintr&auml;ge",
+    ):
+        assert text in antwort.text, f"Fehlt in der Seite: {text!r}"
+
+
+def test_leere_liste_hat_hinweis(client):
+    melde_an(client)
+    antwort = client.get("/neuheiten")
+    assert antwort.status_code == 200
+    assert "Keine Eintr&auml;ge gefunden" in antwort.text
+
+
+def test_suche_filtert_nach_titel_verlag_autor(app, client):
+    with app.state.session_factory() as db:
+        db.add_all(
+            [
+                Neuheit(titel="Catan", verlag="KOSMOS", autor="Klaus Teuber", bgg_id=13),
+                Neuheit(titel="Azul", verlag="Next Move Games", autor="Michael Kiesling", bgg_id=2),
+                Neuheit(titel="Ticket to Ride", verlag="Days of Wonder", autor="Alan R. Moon", bgg_id=9),
+            ]
+        )
+        db.commit()
+
+    melde_an(client)
+    assert "Catan" in client.get("/neuheiten?q=catan").text
+    assert "Azul" not in client.get("/neuheiten?q=catan").text
+    assert "Ticket to Ride" in client.get("/neuheiten?q=wonder").text  # Verlag-Treffer
+
+
+def test_status_filter(app, client):
+    with app.state.session_factory() as db:
+        db.add(Neuheit(titel="Catan", bgg_id=13, status="neuheit"))
+        db.add(Neuheit(titel="Azul", bgg_id=2, status="planung"))
+        db.commit()
+
+    melde_an(client)
+    seite = client.get("/neuheiten?status=planung").text
+    assert "Azul" in seite
+    assert ">Catan<" not in seite
+
+
+def test_sortierung_nach_jahr_abwaerts(app, client):
+    with app.state.session_factory() as db:
+        db.add_all(
+            [
+                Neuheit(titel="Alt", bgg_id=1, erscheinungsjahr=1990),
+                Neuheit(titel="Neu", bgg_id=2, erscheinungsjahr=2026),
+                Neuheit(titel="Mittel", bgg_id=3, erscheinungsjahr=2010),
+            ]
+        )
+        db.commit()
+
+    melde_an(client)
+    antwort = client.get("/neuheiten?sort=jahr&richtung=ab")
+    reihenfolge = [zeile for zeile in ("Neu", "Mittel", "Alt") if zeile]
+    position = [antwort.text.index(titel) for titel in reihenfolge]
+    assert position == sorted(position)
+
+
+def test_unbekannte_sortierung_faellt_auf_standard_zurueck(client):
+    melde_an(client)
+    antwort = client.get("/neuheiten?sort=hack&richtung=auf")
+    assert antwort.status_code == 200
+
+
+def test_sync_button_nur_fuer_redaktion_sichtbar(app, client):
+    lege_benutzer_an(app, "rezensentin", role="rezensent")
+
+    melde_an(client)
+    admin_seite = client.get("/neuheiten").text
+    assert "Jetzt synchronisieren" in admin_seite
+
+    client.post("/logout")
+    melde_an(client, "rezensentin", "test-12345678")
+    rezensent_seite = client.get("/neuheiten").text
+    assert "Jetzt synchronisieren" not in rezensent_seite
+
+
+def test_sync_endpunkt_rollen_pruefung(app, settings):
+    lege_benutzer_an(app, "rezensentin", role="rezensent")
+    from fastapi.testclient import TestClient
+
+    with TestClient(create_app(settings)) as client:
+        melde_an(client, "rezensentin", "test-12345678")
+        antwort = client.post("/neuheiten/sync", follow_redirects=False)
+        assert antwort.status_code == 403
+
+
+def test_manueller_sync_legt_eintraege_an(monkeypatch, settings):
+    from fastapi.testclient import TestClient
+
+    fake = FakeBggClient(
+        treffer=[SuchTreffer(13, "Catan", 1995)],
+        spiel_details=[
+            BggSpiel(bgg_id=13, titel="Catan", verlag="KOSMOS", autor="Klaus Teuber",
+                     erscheinungsjahr=1995, typ="boardgame"),
+            BggSpiel(bgg_id=55, titel="Catan: Erweiterung", verlag=None, autor=None,
+                     erscheinungsjahr=1996, typ="boardgameexpansion"),
+        ],
+    )
+
+    with TestClient(create_app(settings)) as client:
+        melde_an(client)
+        _client_mit_fakes(monkeypatch, client, fake)
+
+        antwort = client.post(
+            "/neuheiten/sync",
+            data={"suchbegriff": "catan"},
+            follow_redirects=False,
+        )
+
+        assert antwort.status_code == 303
+        assert "/neuheiten" in antwort.headers["location"]
+        # Meldung über Sync-Ergebnis:
+        folge = client.get(antwort.headers["location"])
+        assert "Sync abgeschlossen" in folge.text
+        assert "1 neu" in folge.text
+
+        with client.app.state.session_factory() as db:
+            eintraege = db.query(Neuheit).all()
+            assert [e.bgg_id for e in eintraege] == [13]  # Erweiterung gefiltert
+
+        assert fake.suche_aufrufe == ["catan"]  # expliziter Begriff statt Standard
+
+
+def test_manueller_sync_ohne_suchbegriff_nutzt_konfigurierte_begriffe(monkeypatch, settings):
+    from fastapi.testclient import TestClient
+
+    monkeypatch.setenv("SPIELE_BGG_SUCHBEGRIFFE", "essen, familien")
+    fake = FakeBggClient()
+    with TestClient(create_app(settings)) as client:
+        melde_an(client)
+        _client_mit_fakes(monkeypatch, client, fake)
+        client.post("/neuheiten/sync", data={"suchbegriff": ""})
+        assert fake.suche_aufrufe == ["essen", "familien"]
+
+
+def test_htmx_request_liefert_nur_teilfragment(app, client):
+    melde_an(client)
+    antwort = client.get("/neuheiten", headers={"HX-Request": "true"})
+    assert antwort.status_code == 200
+    assert "neuheiten-liste" in antwort.text  # Fragment-Anker vorhanden
+    assert "<html" not in antwort.text.lower()
+
+
+def test_scheduler_standardmaessig_deaktiviert_in_tests(settings):
+    app = create_app(settings)
+    plugin = app.state.registry.get("neuheiten")
+    assert plugin._scheduler is None
+
+
+def test_scheduler_laesst_sich_aktivieren(monkeypatch, settings):
+    monkeypatch.setenv("SPIELE_BGG_SYNC_AKTIV", "1")
+    monkeypatch.setenv("SPIELE_BGG_SYNC_INTERVALL_STUNDEN", "24")
+
+    app = create_app(settings)
+    plugin = app.state.registry.get("neuheiten")
+    with TestClient(app):
+        assert plugin._scheduler is not None
+        assert plugin._scheduler.running
+        job = plugin._scheduler.get_job("bgg-neuheiten-sync")
+        assert job is not None
+
+    assert plugin._scheduler is None  # on_unload hat ihn gestoppt
+
+
+def test_migration_legt_tabelle_an(app):
+    from sqlalchemy import inspect
+
+    inspektion = inspect(app.state.engine)
+    spalten = {c["name"] for c in inspektion.get_columns("neuheiten")}
+    assert {
+        "id", "titel", "verlag", "autor", "erscheinungsjahr",
+        "bgg_id", "status", "quelle", "erstellt_am", "aktualisiert_am",
+    } <= spalten