335 lines
11 KiB
Python
335 lines
11 KiB
Python
"""Tests für die Sync-Logik: Upsert (Update statt Duplikat), Filter, Robustheit."""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from redaktionskern.db import Base
|
|
|
|
from tests._neuheiten import (
|
|
BggAuthFehler,
|
|
BggFehler,
|
|
BggSpiel,
|
|
MELDUNG_401,
|
|
Neuheit,
|
|
SuchTreffer,
|
|
SyncService,
|
|
ist_moeglicher_prototyp,
|
|
speichere_oder_aktualisiere,
|
|
)
|
|
|
|
|
|
class FakeBggClient:
|
|
"""Duck-typed Ersatz für BggClient mit festen Ergebnissen."""
|
|
|
|
def __init__(
|
|
self,
|
|
suchergebnisse: dict[str, list[SuchTreffer]] | None = None,
|
|
details_pro_id: dict[int, BggSpiel] | None = None,
|
|
) -> None:
|
|
self.suchergebnisse = suchergebnisse or {}
|
|
self.details_pro_id = details_pro_id or {}
|
|
self.suche_aufrufe: list[str] = []
|
|
self.details_aufrufe: list[list[int]] = []
|
|
|
|
def suche(self, suchbegriff: str, *, max_treffer: int | None = None):
|
|
self.suche_aufrufe.append(suchbegriff)
|
|
treffer = list(self.suchergebnisse.get(suchbegriff, []))
|
|
return treffer[:max_treffer] if max_treffer is not None else treffer
|
|
|
|
def details(self, bgg_ids):
|
|
self.details_aufrufe.append(list(bgg_ids))
|
|
return [self.details_pro_id[i] for i in bgg_ids if i in self.details_pro_id]
|
|
|
|
|
|
@pytest.fixture
|
|
def session_factory():
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False})
|
|
Base.metadata.create_all(engine)
|
|
fabrik = sessionmaker(bind=engine, expire_on_commit=False, autoflush=False)
|
|
yield fabrik
|
|
engine.dispose()
|
|
|
|
|
|
def _spiel(bgg_id: int = 13, titel: str = "Catan", **aenderungen) -> BggSpiel:
|
|
basis = dict(
|
|
bgg_id=bgg_id,
|
|
titel=titel,
|
|
verlag="KOSMOS",
|
|
autor="Klaus Teuber",
|
|
erscheinungsjahr=1995,
|
|
typ="boardgame",
|
|
)
|
|
basis.update(aenderungen)
|
|
return BggSpiel(**basis)
|
|
|
|
|
|
def _alle_eintraege(session_factory) -> list[Neuheit]:
|
|
with session_factory() as db:
|
|
return list(db.scalars(select(Neuheit).order_by(Neuheit.bgg_id)).all())
|
|
|
|
|
|
# ---------- Update statt Duplikat ----------
|
|
|
|
|
|
def test_sync_reicht_trefferlimit_weiter(session_factory):
|
|
client = FakeBggClient(
|
|
suchergebnisse={
|
|
"viel": [SuchTreffer(i, f"Spiel {i}", None) for i in range(100)]
|
|
},
|
|
details_pro_id={},
|
|
)
|
|
service = SyncService(session_factory, client, max_treffer_pro_suchbegriff=7)
|
|
|
|
ergebnis = service.synchronisiere(["viel"])
|
|
|
|
# Das Limit begrenzt die Trefferliste vor der Detailabfrage:
|
|
assert len(client.details_aufrufe[0]) <= 20 # Batch-Größe bleibt intakt
|
|
assert sum(len(b) for b in client.details_aufrufe) == 7
|
|
|
|
|
|
def test_sync_legt_eintraege_an(session_factory):
|
|
client = FakeBggClient(
|
|
suchergebnisse={"catan": [SuchTreffer(13, "Catan", 1995)]},
|
|
details_pro_id={13: _spiel()},
|
|
)
|
|
service = SyncService(
|
|
session_factory, client, jahr_filter=False, deutscher_titel_filter=False
|
|
)
|
|
|
|
ergebnis = service.synchronisiere(["catan"])
|
|
|
|
assert ergebnis.neu == 1
|
|
assert ergebnis.aktualisiert == 0
|
|
eintraege = _alle_eintraege(session_factory)
|
|
assert len(eintraege) == 1
|
|
eintrag = eintraege[0]
|
|
assert eintrag.titel == "Catan"
|
|
assert eintrag.verlag == "KOSMOS"
|
|
assert eintrag.autor == "Klaus Teuber"
|
|
assert eintrag.erscheinungsjahr == 1995
|
|
assert eintrag.bgg_id == 13
|
|
assert eintrag.status == "neuheit"
|
|
assert eintrag.quelle == "boardgamegeek"
|
|
|
|
|
|
def test_sync_aktualisiert_bestehende_eintraege_statt_duplikate(session_factory):
|
|
client = FakeBggClient(
|
|
suchergebnisse={"catan": [SuchTreffer(13, "Catan", 1995)]},
|
|
details_pro_id={13: _spiel()},
|
|
)
|
|
service = SyncService(
|
|
session_factory, client, jahr_filter=False, deutscher_titel_filter=False
|
|
)
|
|
|
|
service.synchronisiere(["catan"])
|
|
|
|
# Zweiter Lauf mit geänderten Daten (z. B. korrigiertes Jahr, neuer Verlag)
|
|
client.details_pro_id[13] = _spiel(
|
|
verlag="Catan Studio", erscheinungsjahr=2020
|
|
)
|
|
ergebnis = service.synchronisiere(["catan"])
|
|
|
|
assert ergebnis.neu == 0
|
|
assert ergebnis.aktualisiert == 1
|
|
eintraege = _alle_eintraege(session_factory)
|
|
assert len(eintraege) == 1, "Update statt Duplikat verletzt"
|
|
eintrag = eintraege[0]
|
|
assert eintrag.verlag == "Catan Studio"
|
|
assert eintrag.erscheinungsjahr == 2020
|
|
assert eintrag.status == "neuheit", "Status darf beim Update nicht verloren gehen"
|
|
|
|
|
|
def test_sync_veraenderter_titel_gleiche_bgg_id_kein_duplikat(session_factory):
|
|
client = FakeBggClient(
|
|
suchergebnisse={"catan": [SuchTreffer(13, "Catan", 1995)]},
|
|
details_pro_id={13: _spiel()},
|
|
)
|
|
service = SyncService(
|
|
session_factory, client, jahr_filter=False, deutscher_titel_filter=False
|
|
)
|
|
service.synchronisiere(["catan"])
|
|
|
|
client.details_pro_id[13] = _spiel(titel="Catan (Neuausgabe)")
|
|
service.synchronisiere(["catan"])
|
|
|
|
eintraege = _alle_eintraege(session_factory)
|
|
assert len(eintraege) == 1
|
|
assert eintraege[0].titel == "Catan (Neuausgabe)"
|
|
|
|
|
|
def test_speichere_oder_aktualisiere_rückgabe_werte(session_factory):
|
|
with session_factory() as db:
|
|
assert speichere_oder_aktualisiere(db, _spiel()) is True
|
|
db.flush()
|
|
assert speichere_oder_aktualisiere(db, _spiel(erscheinungsjahr=1996)) is False
|
|
|
|
|
|
# ---------- Filter: Erweiterungen & Prototypen ----------
|
|
|
|
|
|
def test_sync_filtert_erweiterungen(session_factory):
|
|
client = FakeBggClient(
|
|
suchergebnisse={
|
|
"catan": [
|
|
SuchTreffer(13, "Catan", 1995),
|
|
SuchTreffer(55, "Catan: Erweiterung", 1996),
|
|
]
|
|
},
|
|
details_pro_id={
|
|
13: _spiel(),
|
|
55: _spiel(55, "Catan: Erweiterung", typ="boardgameexpansion"),
|
|
},
|
|
)
|
|
service = SyncService(
|
|
session_factory, client, jahr_filter=False, deutscher_titel_filter=False
|
|
)
|
|
|
|
ergebnis = service.synchronisiere(["catan"])
|
|
|
|
assert ergebnis.gefiltert == 1
|
|
eintraege = _alle_eintraege(session_factory)
|
|
assert [e.bgg_id for e in eintraege] == [13]
|
|
|
|
|
|
def test_sync_filtert_prototypen(session_factory):
|
|
client = FakeBggClient(
|
|
suchergebnisse={
|
|
"essen": [
|
|
SuchTreffer(13, "Catan", 1995),
|
|
SuchTreffer(77, "Fabrik — Prototyp 2026", None),
|
|
SuchTreffer(78, "My Prototype Game", None),
|
|
]
|
|
},
|
|
details_pro_id={
|
|
13: _spiel(),
|
|
77: _spiel(77, "Fabrik — Prototyp 2026"),
|
|
78: _spiel(78, "My Prototype Game"),
|
|
},
|
|
)
|
|
service = SyncService(
|
|
session_factory, client, jahr_filter=False, deutscher_titel_filter=False
|
|
)
|
|
|
|
ergebnis = service.synchronisiere(["essen"])
|
|
|
|
assert ergebnis.gefiltert == 2
|
|
assert [e.bgg_id for e in _alle_eintraege(session_factory)] == [13]
|
|
|
|
|
|
def test_prototyp_heuristik():
|
|
assert ist_moeglicher_prototyp("Ein Prototyp von etwas")
|
|
assert ist_moeglicher_prototyp("Prototype X")
|
|
assert ist_moeglicher_prototyp("PROTOTYPISCHES Spiel")
|
|
assert not ist_moeglicher_prototyp("Catan")
|
|
assert not ist_moeglicher_prototyp("Die Burgen von Burgund")
|
|
|
|
|
|
# ---------- Filter: Jahr und deutscher Titel ----------
|
|
|
|
|
|
def test_sync_filtert_nach_jahr_und_deutschem_titel(session_factory):
|
|
aktuelles = 2026
|
|
client = FakeBggClient(
|
|
suchergebnisse={
|
|
"mix": [
|
|
SuchTreffer(13, "Catan", aktuelles - 1),
|
|
SuchTreffer(42, "Die neuen Siedler", aktuelles),
|
|
SuchTreffer(99, "Future Game", aktuelles + 1),
|
|
SuchTreffer(88, "The Old English Game", aktuelles),
|
|
]
|
|
},
|
|
details_pro_id={
|
|
13: _spiel(13, "Catan", erscheinungsjahr=aktuelles - 1),
|
|
42: _spiel(42, "Die neuen Siedler", erscheinungsjahr=aktuelles),
|
|
99: _spiel(99, "Future Game", erscheinungsjahr=aktuelles + 1),
|
|
88: _spiel(88, "The Old English Game", erscheinungsjahr=aktuelles),
|
|
},
|
|
)
|
|
service = SyncService(session_factory, client)
|
|
service._min_jahr = aktuelles
|
|
|
|
ergebnis = service.synchronisiere(["mix"])
|
|
|
|
assert ergebnis.neu == 1 # nur "Die neuen Siedler" überlebt beide Filter
|
|
assert ergebnis.gefiltert == 3 # 1 zu alt + 1 zukünftig aber englisch + 1 aktuell aber englisch
|
|
eintraege = _alle_eintraege(session_factory)
|
|
assert [e.bgg_id for e in eintraege] == [42]
|
|
|
|
|
|
def test_sync_ohne_filter_behält_alle_jahre_und_sprachen(session_factory):
|
|
client = FakeBggClient(
|
|
suchergebnisse={
|
|
"mix": [
|
|
SuchTreffer(13, "Catan", 1995),
|
|
SuchTreffer(88, "The English Game", 2026),
|
|
]
|
|
},
|
|
details_pro_id={
|
|
13: _spiel(13, "Catan", erscheinungsjahr=1995),
|
|
88: _spiel(88, "The English Game", erscheinungsjahr=2026),
|
|
},
|
|
)
|
|
service = SyncService(
|
|
session_factory, client, jahr_filter=False, deutscher_titel_filter=False
|
|
)
|
|
|
|
ergebnis = service.synchronisiere(["mix"])
|
|
|
|
assert ergebnis.neu == 2
|
|
assert ergebnis.gefiltert == 0
|
|
|
|
|
|
# ---------- Robustheit ----------
|
|
|
|
|
|
def test_401_bricht_sync_ab_mit_klarer_meldung(session_factory):
|
|
"""BGG-401: Lauf stoppt, Meldung landet im Ergebnis, Teilergebnisse bleiben."""
|
|
|
|
class AuthKaputterClient(FakeBggClient):
|
|
def suche(self, suchbegriff: str, *, max_treffer: int | None = None):
|
|
self.suche_aufrufe.append(suchbegriff)
|
|
if suchbegriff == "abgelehnt":
|
|
raise BggAuthFehler(MELDUNG_401)
|
|
return list(self.suchergebnisse.get(suchbegriff, []))
|
|
|
|
client = AuthKaputterClient(
|
|
suchergebnisse={"gut": [SuchTreffer(13, "Catan", 1995)]},
|
|
details_pro_id={13: _spiel()},
|
|
)
|
|
service = SyncService(
|
|
session_factory, client, jahr_filter=False, deutscher_titel_filter=False
|
|
)
|
|
|
|
ergebnis = service.synchronisiere(["gut", "abgelehnt", "dritter"])
|
|
|
|
assert ergebnis.abbruch == MELDUNG_401
|
|
assert ergebnis.als_text() == MELDUNG_401 # klare Meldung statt Fehlerzähler
|
|
assert ergebnis.neu == 1 # Teilergebnis vor dem 401 bleibt erhalten
|
|
assert client.suche_aufrufe == ["gut", "abgelehnt"] # kein weiterer Begriff
|
|
assert len(_alle_eintraege(session_factory)) == 1 # Commit trotz Abbruch
|
|
|
|
|
|
def test_fehler_bei_einem_suchbegriff_bricht_lauf_nicht_ab(session_factory):
|
|
class HalbKaputterClient(FakeBggClient):
|
|
def suche(self, suchbegriff: str, *, max_treffer: int | None = None):
|
|
if suchbegriff == "kaputt":
|
|
raise BggFehler("BGG nicht erreichbar")
|
|
return super().suche(suchbegriff, max_treffer=max_treffer)
|
|
|
|
client = HalbKaputterClient(
|
|
suchergebnisse={"gut": [SuchTreffer(13, "Catan", 1995)]},
|
|
details_pro_id={13: _spiel()},
|
|
)
|
|
service = SyncService(
|
|
session_factory, client, jahr_filter=False, deutscher_titel_filter=False
|
|
)
|
|
|
|
ergebnis = service.synchronisiere(["kaputt", "gut"])
|
|
|
|
assert len(ergebnis.fehler) == 1
|
|
assert "kaputt" in ergebnis.fehler[0]
|
|
assert ergebnis.neu == 1
|
|
assert len(_alle_eintraege(session_factory)) == 1
|