330 lines
11 KiB
Python
330 lines
11 KiB
Python
"""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 datetime import datetime, timezone
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from redaktionskern.app import create_app
|
|
|
|
from tests._neuheiten import (
|
|
BggAuthFehler,
|
|
BggSpiel,
|
|
MELDUNG_401,
|
|
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
|
|
|
|
|
|
class FehlerwerfenderFake(FakeBggClient):
|
|
"""Wirft bei der Suche einen Fehler (z. B. BggAuthFehler bei HTTP 401)."""
|
|
|
|
def __init__(self, fehler: Exception):
|
|
super().__init__()
|
|
self.fehler = fehler
|
|
|
|
def suche(self, suchbegriff: str, *, max_treffer: int | None = None):
|
|
self.suche_aufrufe.append(suchbegriff)
|
|
raise self.fehler
|
|
|
|
|
|
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?jahr=1990")
|
|
|
|
assert antwort.status_code == 200
|
|
for text in (
|
|
"Spieltitel", "Verlag", "Autor", "Erscheinungsjahr", "Status",
|
|
"Catan", "KOSMOS", "Klaus Teuber", "1995", "1 Einträ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ä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&jahr=1990")
|
|
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_standard_jahr_filter_zeigt_nur_aktuelles_jahr_und_neuer(app, client):
|
|
"""Ohne jahr-Parameter: nur aktuelles Jahr und neuer (plus ohne Jahr)."""
|
|
with app.state.session_factory() as db:
|
|
db.add_all(
|
|
[
|
|
Neuheit(titel="Alt", bgg_id=1, erscheinungsjahr=1990),
|
|
Neuheit(titel="Aktuell", bgg_id=2, erscheinungsjahr=datetime.now(timezone.utc).year),
|
|
Neuheit(titel="Ohne Jahr", bgg_id=3),
|
|
]
|
|
)
|
|
db.commit()
|
|
|
|
melde_an(client)
|
|
antwort = client.get("/neuheiten")
|
|
assert "Aktuell" in antwort.text
|
|
assert "Ohne Jahr" in antwort.text
|
|
assert ">Alt<" not in antwort.text
|
|
|
|
|
|
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"),
|
|
],
|
|
)
|
|
|
|
monkeypatch.setenv("SPIELE_BGG_TOKEN", "test-token")
|
|
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")
|
|
monkeypatch.setenv("SPIELE_BGG_TOKEN", "test-token")
|
|
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_sync_ohne_token_wird_uebersprungen(monkeypatch, settings):
|
|
from fastapi.testclient import TestClient
|
|
|
|
monkeypatch.delenv("SPIELE_BGG_TOKEN", raising=False)
|
|
fake = FakeBggClient(treffer=[SuchTreffer(13, "Catan", 1995)])
|
|
|
|
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
|
|
folge = client.get(antwort.headers["location"])
|
|
assert (
|
|
"Kein BGG-API-Token konfiguriert (SPIELE_BGG_TOKEN) — "
|
|
"Sync übersprungen." in folge.text
|
|
)
|
|
assert fake.suche_aufrufe == [] # kein Request an die BGG-API gefeuert
|
|
|
|
|
|
def test_sync_401_zeigt_klare_fehlermeldung(monkeypatch, settings):
|
|
from fastapi.testclient import TestClient
|
|
|
|
monkeypatch.setenv("SPIELE_BGG_TOKEN", "falscher-token")
|
|
fake = FehlerwerfenderFake(BggAuthFehler(MELDUNG_401))
|
|
|
|
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
|
|
folge = client.get(antwort.headers["location"])
|
|
assert MELDUNG_401 in folge.text
|
|
assert "Unerwartete HTTP-Antwort" not in folge.text
|
|
|
|
|
|
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
|