901 lines
32 KiB
Python
901 lines
32 KiB
Python
"""Tests für die Web-Quellen-Erweiterung des Neuheiten-Plugins.
|
||
|
||
Alle HTTP-Antworten sind gemockt (Fake-Transport / Client-Fabrik) — es
|
||
gibt keine echten Netzwerkaufrufe. Abgedeckt: Parser-Extraktion je
|
||
Quelle, Pagination-Durchlauf, 304-Behandlung (If-None-Match), Rate-
|
||
Limit, Fehler-Isolation pro Quelle, Duplikatprüfung gegen BGG-Titel,
|
||
Env-Schalter und Sync-Übersicht im UI.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
|
||
import httpx
|
||
import pytest
|
||
|
||
from tests._neuheiten import (
|
||
ADAPTER_KLASSEN,
|
||
BrettspielboxQuelle,
|
||
CliquenabendQuelle,
|
||
Neuheit,
|
||
QuellenFehler,
|
||
QuellenStatus,
|
||
QuellenTreffer,
|
||
QuellenLaufZeile,
|
||
SammelErgebnis,
|
||
SpielEssenQuelle,
|
||
SpielboxQuelle,
|
||
USER_AGENT,
|
||
WebQuellenClient,
|
||
WebQuellenSyncService,
|
||
jahr_aus_datumsangabe,
|
||
quelle_aktiv,
|
||
titel_aehnlichkeit,
|
||
)
|
||
|
||
# ---------- Gemockte Seiten (realistische Struktur-Anleihen der Quellen) ----
|
||
|
||
SPIELBOX_SEITE_1 = b"""
|
||
<div class="blog">
|
||
<div class="item column-1">
|
||
<div class="page-header"><h2 itemprop="name">
|
||
<a href="/neuheiten-neue-spiele/1369-pegasus-next-station-berlin">Pegasus Spiele: Next Station Berlin</a>
|
||
</h2></div>
|
||
<p><strong>24.07.2026</strong> - <em>Next Station Berlin</em> von Matthew Dunstan, das neu bei Pegasus Spiele erschienen ist.</p>
|
||
</div>
|
||
<div class="item column-2">
|
||
<div class="page-header"><h2><a href="/neuheiten-neue-spiele/1247-micromacro-kids">MICROMACRO Kids</a></h2></div>
|
||
<p><strong>02.03.2026</strong> - Neu im März.</p>
|
||
</div>
|
||
</div>
|
||
<ul class="pagination"><li><a href="/neuheiten-neue-spiele?start=6">2</a></li></ul>
|
||
"""
|
||
|
||
SPIELBOX_SEITE_2 = b"""
|
||
<div class="item column-1">
|
||
<div class="page-header"><h2><a href="/neuheiten-neue-spiele/1123-schmidt-ringtraeger">Schmidt Spiele: Ringträger</a></h2></div>
|
||
<p><strong>15.08.2026</strong> - Neue Titel von Schmidt Spiele.</p>
|
||
</div>
|
||
<ul class="pagination"></ul>
|
||
"""
|
||
|
||
BRETTSPIELBOX_SEITE = """
|
||
<div class="wp-block-columns is-layout-flex">
|
||
<div class="wp-block-column">
|
||
<p>0-9</p>
|
||
<ul>
|
||
<li>-60 Grad (dlp games)</li>
|
||
<li>5Pairs (<a href="https://brettspielbox.de/vorschau-bsl/">BSL</a>)</li>
|
||
</ul>
|
||
<p>A</p>
|
||
<ul><li>Abaku (Albi)</li></ul>
|
||
</div>
|
||
<div class="wp-block-column">
|
||
<p>M</p>
|
||
<ul>
|
||
<li>Machu Pichu (HUCH!)</li>
|
||
<li>Titel ohne Verlag</li>
|
||
<li>Fadenfroh (1 More Time Games)</li>
|
||
</ul>
|
||
</div>
|
||
</div>
|
||
"""
|
||
|
||
SPIELESSEN_PRODUKTE = {
|
||
"path": "static/spiel26/x/logos/",
|
||
"picturePath": "static/spiel26/x/products/",
|
||
"products": [
|
||
{
|
||
"ID": "188",
|
||
"TITEL": "-60°",
|
||
"UNTERTITEL": "Nanox Games",
|
||
"FIRMA_ID": "436",
|
||
"INFO": (
|
||
"<table><tr><td>Autor: </td><td>Alexander Pfister & Peter Prinz</td></tr>"
|
||
"<tr><td>Verlag: </td><td>Nanox Games</td></tr>"
|
||
"<tr><td>Vertrieb: </td><td>dlp games Verlag GmbH</td></tr>"
|
||
"<tr><td>Erscheinungsdatum:</td><td>10/2026</td></tr>"
|
||
"<tr><td>Spielalter:</td><td>ab 12 Jahren</td></tr></table>"
|
||
),
|
||
},
|
||
{"ID": "189", "TITEL": "", "INFO": ""},
|
||
{"ID": "190", "TITEL": "Nur Aussteller-Match", "UNTERTITEL": "", "FIRMA_ID": "777", "INFO": ""},
|
||
],
|
||
}
|
||
|
||
SPIELESSEN_AUSSTELLER = {"exhibitors": [{"ID": "777", "NAME": "Mysterium Verlag"}]}
|
||
|
||
CLIQUENABEND_SEITE = """
|
||
<table>
|
||
<tr><td>Update</td><td>Verlag</td><td>Stand</td><td>Spiel</td><td>Autor</td></tr>
|
||
<tr>
|
||
<td>15.10.25</td>
|
||
<td>
|
||
<div style="font-weight:bold;">Dranda Games</div>
|
||
<a class="grey" href="/profile/067900.html">Profil</a>,
|
||
<div style="font-weight:bold;">PD Verlag</div>
|
||
<a class="grey" href="/profile/152000.html">Profil</a>,
|
||
</td>
|
||
<td>3-H400</td>
|
||
<td><div><a href="/spiele/790610-Galactic-Cruise.html" target="_blank"><b style="font-size:1.1em;">Galactic Cruise</b></a>
|
||
<div class="gameinfos">100,- €</div></div></td>
|
||
<td>T.K. King und Koltin Thompson</td>
|
||
</tr>
|
||
<tr><td>26.09.25</td><td>Piatnik Profil ,</td><td>6-G200</td>
|
||
<td><div><a href="/spiele/790700-Elixirus.html"><b>Elixirus</b></a></div></td>
|
||
<td>Dickie Chapin</td></tr>
|
||
<tr><td>01.10.25</td><td>ohne Link</td><td>—</td><td>Nur Texttitel</td><td>Unbekannt</td></tr>
|
||
</table>
|
||
"""
|
||
|
||
CLIQUENABEND_DETAIL_1 = """
|
||
<html><body>
|
||
<h1>Galactic Cruise</h1>
|
||
<div class="right_bar"><center>
|
||
<a href="/spiele/790610-Galactic-Cruise.html"><img src="/images/db/62452_360x240.jpg" border="0"></a><br>
|
||
</center></div>
|
||
<b>Release:</b><br>
|
||
2025 <br>
|
||
<b>Anzahl der Spieler:</b><br>
|
||
1 bis 4 Spieler<br>
|
||
</body></html>
|
||
"""
|
||
|
||
CLIQUENABEND_DETAIL_FEHLT = "<html><body><p>Seite ohne Release-Feld</p></body></html>"
|
||
|
||
|
||
class FakeUhr:
|
||
"""Steuerbare Monoton-Uhr."""
|
||
|
||
def __init__(self) -> None:
|
||
self.zeit = 0.0
|
||
|
||
def __call__(self) -> float:
|
||
return self.zeit
|
||
|
||
|
||
class KartenTransport(httpx.BaseTransport):
|
||
"""Antworten je URL; zeichnet Requests (URL, Header, Uhrzeit) auf."""
|
||
|
||
def __init__(self, antworten: dict[str, httpx.Response], uhr: FakeUhr):
|
||
self.antworten = antworten
|
||
self.uhr = uhr
|
||
self.anfragen: list[tuple[str, float]] = []
|
||
self.header: list[dict[str, str]] = []
|
||
|
||
def handle_request(self, request: httpx.Request) -> httpx.Response:
|
||
url = str(request.url)
|
||
self.anfragen.append((url, self.uhr()))
|
||
self.header.append(dict(request.headers))
|
||
# 1) exakte URL, 2) URL ohne Query (dynamische Parameter wie
|
||
# ?columns=… bei JSON-Endpunkten). URLs mit Query fallen NIE auf
|
||
# eine längere gemockte Basis-URL zurück (?start=N wäre sonst
|
||
# ein Präfix der Basis-URL und würde endlos Seite 1 liefern).
|
||
antwort = self.antworten.get(url)
|
||
if antwort is None and "?" in url:
|
||
antwort = self.antworten.get(url.split("?")[0])
|
||
if antwort is None and "?" not in url:
|
||
antwort = self.antworten.get(url.split("?")[0])
|
||
if antwort is None or isinstance(antwort, Exception):
|
||
raise QuellenFehler(f"Keine gemockte Antwort für {url}")
|
||
return antwort
|
||
|
||
|
||
def _antwort(inhalt: bytes | str, etag: str | None = None) -> httpx.Response:
|
||
header = {}
|
||
if etag:
|
||
header["ETag"] = etag
|
||
return httpx.Response(
|
||
200, content=inhalt.encode("utf-8") if isinstance(inhalt, str) else inhalt,
|
||
headers=header, request=httpx.Request("GET", "https://example.org/mock"),
|
||
)
|
||
|
||
|
||
def _client(transport, uhr, schlaf=None, validatoren=None) -> WebQuellenClient:
|
||
return WebQuellenClient(
|
||
transport=transport,
|
||
mindestabstand_sekunden=1.0,
|
||
schlaf=schlaf or (lambda s: None),
|
||
uhr=uhr,
|
||
validatoren=validatoren,
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def session_factory():
|
||
from sqlalchemy import create_engine
|
||
from sqlalchemy.orm import sessionmaker
|
||
|
||
from redaktionskern.db import Base
|
||
|
||
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 _adapter(klasse, seiten: dict[str, bytes | str], etag: str | None = None):
|
||
uhr = FakeUhr()
|
||
transport = KartenTransport(
|
||
{u: _antwort(i, etag) for u, i in seiten.items()}, uhr
|
||
)
|
||
return klasse(_client(transport, uhr)), transport
|
||
|
||
|
||
# ---------- Hilfsfunktionen ----------
|
||
|
||
|
||
def test_jahr_aus_datumsangabe():
|
||
assert jahr_aus_datumsangabe("24.07.2026") == 2026
|
||
assert jahr_aus_datumsangabe("10/2026") == 2026
|
||
assert jahr_aus_datumsangabe("Herbst 2026") == 2026
|
||
assert jahr_aus_datumsangabe("Q3 2026") == 2026
|
||
assert jahr_aus_datumsangabe(None) is None
|
||
assert jahr_aus_datumsangabe("ohne Jahr") is None
|
||
|
||
|
||
def test_titel_aehnlichkeit_erkennt_duplikate_keine_erweiterungen():
|
||
# Reale BGG-Duplikatfälle werden erkannt …
|
||
assert titel_aehnlichkeit("Die Siedler von Catan", "Catan") >= 92
|
||
assert titel_aehnlichkeit("Fadenfroh!", "Fadenfroh") >= 92
|
||
assert titel_aehnlichkeit("Machu Pichu", "Machu Picchu") >= 92
|
||
# … echte Erweiterungen/andere Spiele nicht.
|
||
assert titel_aehnlichkeit("Catan", "Catan: Erweiterung") < 92
|
||
assert titel_aehnlichkeit("Azul", "Azul: Summer Pavilion") < 92
|
||
|
||
|
||
# ---------- Parser je Quelle ----------
|
||
|
||
|
||
def test_parser_spielbox_extrahiert_titel_verlag_datum_url():
|
||
# Ohne Pagination-Link — der Parser-Test interessiert nur Seite 1.
|
||
seite_ohne_pagination = SPIELBOX_SEITE_1.replace(
|
||
b'<ul class="pagination"><li><a href="/neuheiten-neue-spiele?start=6">2</a></li></ul>',
|
||
b'<ul class="pagination"></ul>',
|
||
)
|
||
adapter, _ = _adapter(SpielboxQuelle, {"https://www.spielbox.de/neuheiten-neue-spiele": seite_ohne_pagination})
|
||
ergebnis = adapter.sammle()
|
||
|
||
assert len(ergebnis.treffer) == 2
|
||
erster = ergebnis.treffer[0]
|
||
assert erster.titel == "Pegasus Spiele: Next Station Berlin"
|
||
assert erster.verlag == "Pegasus Spiele"
|
||
assert erster.erscheinungsdatum_oder_quartal == "24.07.2026"
|
||
assert erster.quellen_url == (
|
||
"https://www.spielbox.de/neuheiten-neue-spiele/1369-pegasus-next-station-berlin"
|
||
)
|
||
assert ergebnis.treffer[1].verlag is None # kein Doppelpunkt im Titel
|
||
|
||
|
||
def test_parser_brettspielbox_zerlegt_titel_verlag_und_links():
|
||
adapter, _ = _adapter(
|
||
BrettspielboxQuelle,
|
||
{"https://brettspielbox.de/brettspiel-neuheiten-spiel-2026-herbst-2026-a-z/": BRETTSPIELBOX_SEITE},
|
||
)
|
||
ergebnis = adapter.sammle()
|
||
|
||
titel_verlage = {(t.titel, t.verlag) for t in ergebnis.treffer}
|
||
assert ("-60 Grad", "dlp games") in titel_verlage
|
||
assert ("5Pairs", "BSL") in titel_verlage
|
||
assert ("Abaku", "Albi") in titel_verlage
|
||
assert ("Titel ohne Verlag", None) in titel_verlage
|
||
# Eintrag mit Vorstellungs-Link nutzt diesen als quellen_url …
|
||
fuenf = next(t for t in ergebnis.treffer if t.titel == "5Pairs")
|
||
assert fuenf.quellen_url == "https://brettspielbox.de/vorschau-bsl/"
|
||
# … alle anderen die Listenseite.
|
||
ohne = next(t for t in ergebnis.treffer if t.titel == "Abaku")
|
||
assert ohne.quellen_url.endswith("a-z/")
|
||
|
||
|
||
def test_parser_spielessen_nutzt_json_api_endpunkt():
|
||
adapter, transport = _adapter(
|
||
lambda client: SpielEssenQuelle(client, jahr=2026),
|
||
{}, # Endpunkte werden dynamisch gebaut → direkt setzen
|
||
)
|
||
transport.antworten.update(
|
||
{
|
||
# Der Adapter hängt ?columns=… an — Mock matcht per Präfix.
|
||
"https://maps.eyeled-services.de/de/spiel26/products": _antwort(
|
||
json.dumps(SPIELESSEN_PRODUKTE)
|
||
),
|
||
"https://maps.eyeled-services.de/de/spiel26/exhibitors": _antwort(
|
||
json.dumps(SPIELESSEN_AUSSTELLER)
|
||
),
|
||
}
|
||
)
|
||
|
||
ergebnis = adapter.sammle()
|
||
|
||
assert len(ergebnis.treffer) == 2 # leerer TITEL wird übersprungen
|
||
erster = ergebnis.treffer[0]
|
||
assert erster.titel == "-60°"
|
||
assert erster.verlag == "Nanox Games"
|
||
assert erster.autor == "Alexander Pfister & Peter Prinz"
|
||
assert erster.erscheinungsdatum_oder_quartal == "10/2026"
|
||
assert erster.quellen_url == "https://spiel-essen.de/de/die-spiel/neuheiten"
|
||
# Aussteller-Fallback für Produkte ohne INFO-Verlag:
|
||
assert ergebnis.treffer[1].verlag == "Mysterium Verlag"
|
||
|
||
|
||
def test_parser_cliquenabend_liest_tabellenspalten():
|
||
adapter, _ = _adapter(
|
||
CliquenabendQuelle,
|
||
{
|
||
"https://www.cliquenabend.de/events/130000-SPIEL-2025-Essen.html": CLIQUENABEND_SEITE,
|
||
"https://www.cliquenabend.de/spiele/790610-Galactic-Cruise.html": CLIQUENABEND_DETAIL_1,
|
||
"https://www.cliquenabend.de/spiele/790700-Elixirus.html": CLIQUENABEND_DETAIL_FEHLT,
|
||
},
|
||
)
|
||
ergebnis = adapter.sammle()
|
||
|
||
assert [t.titel for t in ergebnis.treffer] == ["Galactic Cruise", "Elixirus"]
|
||
erster = ergebnis.treffer[0]
|
||
assert erster.verlag == "Dranda Games, PD Verlag"
|
||
assert erster.autor == "T.K. King und Koltin Thompson"
|
||
# Release-Feld der Detailseite schlägt das Update-Datum der Tabelle:
|
||
assert erster.erscheinungsdatum_oder_quartal == "2025"
|
||
assert erster.quellen_url == "https://www.cliquenabend.de/spiele/790610-Galactic-Cruise.html"
|
||
# Cover-Bild aus der Detailseite (statt Messe-Foto):
|
||
assert erster.bild_url == "https://www.cliquenabend.de/images/db/62452_360x240.jpg"
|
||
assert ergebnis.treffer[1].verlag == "Piatnik"
|
||
# Ohne Release-Feld auf der Detailseite fällt der Parser auf das
|
||
# Update-Datum zurück:
|
||
assert ergebnis.treffer[1].erscheinungsdatum_oder_quartal == "26.09.25"
|
||
|
||
|
||
def test_parser_cliquenabend_detailseite_fehler_blockiert_nicht():
|
||
adapter, _ = _adapter(
|
||
CliquenabendQuelle,
|
||
{
|
||
"https://www.cliquenabend.de/events/130000-SPIEL-2025-Essen.html": CLIQUENABEND_SEITE,
|
||
# Detailseiten nicht erreichbar → Update-Datum bleibt erhalten.
|
||
},
|
||
)
|
||
ergebnis = adapter.sammle()
|
||
assert len(ergebnis.treffer) == 2
|
||
assert ergebnis.treffer[0].erscheinungsdatum_oder_quartal == "15.10.25"
|
||
|
||
|
||
def test_parser_cliquenabend_toleriert_fehlende_tabelle():
|
||
adapter, _ = _adapter(
|
||
CliquenabendQuelle,
|
||
{"https://www.cliquenabend.de/events/130000-SPIEL-2025-Essen.html": "<html><body><p>Wartung</p></body></html>"},
|
||
)
|
||
ergebnis = adapter.sammle()
|
||
assert ergebnis.treffer == []
|
||
|
||
|
||
# ---------- Pagination ----------
|
||
|
||
|
||
def test_spielbox_durchlaeuft_alle_pagination_seiten():
|
||
basis_url = "https://www.spielbox.de/neuheiten-neue-spiele"
|
||
adapter, transport = _adapter(
|
||
SpielboxQuelle,
|
||
{basis_url: SPIELBOX_SEITE_1, f"{basis_url}?start=6": SPIELBOX_SEITE_2},
|
||
)
|
||
|
||
ergebnis = adapter.sammle()
|
||
|
||
assert len(ergebnis.treffer) == 3 # 2 + 1 über die zweite Seite
|
||
angefragt = [u for u, _ in transport.anfragen]
|
||
assert angefragt.count(basis_url) == 1
|
||
assert f"{basis_url}?start=6" in angefragt
|
||
|
||
|
||
# ---------- Rate-Limit & User-Agent ----------
|
||
|
||
|
||
def test_rate_limit_mindestens_eine_sekunde_zwischen_requests():
|
||
uhr = FakeUhr()
|
||
wartezeiten: list[float] = []
|
||
basis_url = "https://www.spielbox.de/neuheiten-neue-spiele"
|
||
transport = KartenTransport(
|
||
{basis_url: _antwort(SPIELBOX_SEITE_1), f"{basis_url}?start=6": _antwort(SPIELBOX_SEITE_2)},
|
||
uhr,
|
||
)
|
||
|
||
def schlaf(sekunden: float) -> None:
|
||
wartezeiten.append(sekunden)
|
||
uhr.zeit += sekunden
|
||
|
||
client = WebQuellenClient(transport=transport, mindestabstand_sekunden=1.0, schlaf=schlaf, uhr=uhr)
|
||
SpielboxQuelle(client).sammle()
|
||
|
||
assert len(transport.anfragen) == 2
|
||
erste_zeit = transport.anfragen[0][1]
|
||
zweite_zeit = transport.anfragen[1][1]
|
||
assert zweite_zeit - erste_zeit >= 1.0
|
||
assert wartezeiten and wartezeiten[0] > 0
|
||
|
||
|
||
def test_client_sendet_redaktions_user_agent():
|
||
uhr = FakeUhr()
|
||
basis_url = "https://www.spielbox.de/neuheiten-neue-spiele"
|
||
transport = KartenTransport({basis_url: _antwort(SPIELBOX_SEITE_1)}, uhr)
|
||
client = WebQuellenClient(transport=transport, schlaf=lambda s: None, uhr=uhr)
|
||
client.hole(basis_url)
|
||
assert transport.header[0]["user-agent"] == USER_AGENT
|
||
|
||
|
||
# ---------- ETag / 304 ----------
|
||
|
||
|
||
def test_etag_wird_gesendet_und_304_ueberspringt_die_quelle():
|
||
basis_url = "https://www.spielbox.de/neuheiten-neue-spiele"
|
||
uhr = FakeUhr()
|
||
transport = KartenTransport(
|
||
{basis_url: httpx.Response(
|
||
304, request=httpx.Request("GET", basis_url)
|
||
)},
|
||
uhr,
|
||
)
|
||
client = WebQuellenClient(
|
||
transport=transport, schlaf=lambda s: None, uhr=uhr,
|
||
validatoren={basis_url: ('"abc123"', None)},
|
||
)
|
||
|
||
ergebnis = SpielboxQuelle(client).sammle()
|
||
|
||
assert transport.header[0].get("if-none-match") == '"abc123"'
|
||
assert ergebnis.unveraendert is True
|
||
assert ergebnis.treffer == []
|
||
|
||
|
||
def test_etag_wird_aus_antwort_gespeichert():
|
||
basis_url = "https://www.spielbox.de/neuheiten-neue-spiele"
|
||
adapter, _ = _adapter(SpielboxQuelle, {basis_url: SPIELBOX_SEITE_1}, etag='"etag-42"')
|
||
adapter.client.hole(basis_url)
|
||
assert adapter.client.validatoren[basis_url][0] == '"etag-42"'
|
||
|
||
|
||
# ---------- Sync-Orchestrierung ----------
|
||
|
||
|
||
class StatischerAdapter:
|
||
"""Adapter-Ersatz mit fester Trefferliste (für Service-Tests)."""
|
||
|
||
def __init__(self, name: str, treffer: list[QuellenTreffer]):
|
||
self.name = name
|
||
self.anzeigename = name.title()
|
||
self.start_urls = (f"https://{name}.example.org/liste",)
|
||
self._treffer = treffer
|
||
|
||
def __call__(self, client):
|
||
return self
|
||
|
||
def sammle(self) -> SammelErgebnis:
|
||
return SammelErgebnis(treffer=list(self._treffer), seiten=1)
|
||
|
||
|
||
class KaputterAdapter:
|
||
def __init__(self, name: str):
|
||
self.name = name
|
||
self.anzeigename = name.title()
|
||
self.start_urls = (f"https://{name}.example.org/liste",)
|
||
|
||
def __call__(self, client):
|
||
return self
|
||
|
||
def sammle(self) -> SammelErgebnis:
|
||
raise QuellenFehler("Seite umgebaut")
|
||
|
||
|
||
def _service(session_factory, klassen, dedup=None):
|
||
return WebQuellenSyncService(
|
||
session_factory, adapter_klassen=klassen, dedup=dedup,
|
||
client_fabrik=lambda validatoren: WebQuellenClient(
|
||
schlaf=lambda s: None, uhr=FakeUhr(), validatoren=validatoren
|
||
),
|
||
)
|
||
|
||
|
||
def test_sync_legt_web_eintraege_mit_quelle_an(session_factory):
|
||
quelle = StatischerAdapter(
|
||
"spielbox",
|
||
[QuellenTreffer(titel="Ringträger", verlag="Schmidt Spiele",
|
||
autor=None, erscheinungsdatum_oder_quartal="15.08.2026",
|
||
quellen_url="https://www.spielbox.de/x")],
|
||
)
|
||
service = _service(session_factory, (quelle,))
|
||
|
||
zeilen = service.laufe()
|
||
|
||
assert zeilen[0].neu == 1
|
||
assert zeilen[0].fehlermeldung is None
|
||
with session_factory() as db:
|
||
eintrag = db.query(Neuheit).one()
|
||
assert eintrag.quelle == "spielbox"
|
||
assert eintrag.bgg_id is None
|
||
assert eintrag.status == "neuheit"
|
||
assert eintrag.erscheinungsjahr == 2026
|
||
statuszeile = db_status(session_factory, "spielbox")
|
||
assert statuszeile.anzahl_neu == 1
|
||
assert statuszeile.letzte_laufzeit is not None
|
||
|
||
|
||
def db_status(session_factory, quelle: str) -> QuellenStatus:
|
||
with session_factory() as db:
|
||
zeile = db.query(QuellenStatus).filter_by(quelle=quelle).one()
|
||
db.expunge(zeile)
|
||
return zeile
|
||
|
||
|
||
def test_sync_filtert_duplikate_gegen_bggtitel(session_factory):
|
||
with session_factory() as db:
|
||
db.add(Neuheit(titel="Catan", verlag="KOSMOS", autor="Klaus Teuber",
|
||
erscheinungsjahr=1995, bgg_id=13, quelle="boardgamegeek"))
|
||
db.commit()
|
||
|
||
quelle = StatischerAdapter(
|
||
"spielbox",
|
||
[
|
||
QuellenTreffer(titel="Die Siedler von Catan", verlag="KOSMOS",
|
||
quellen_url="https://www.spielbox.de/catan"),
|
||
QuellenTreffer(titel="Ein ganz neues Spiel", verlag="Anderer Verlag",
|
||
quellen_url="https://www.spielbox.de/neu"),
|
||
],
|
||
)
|
||
service = _service(session_factory, (quelle,))
|
||
zeilen = service.laufe()
|
||
|
||
assert zeilen[0].gefiltert == 1 # BGG-Duplikat per Titel-Match
|
||
assert zeilen[0].neu == 1
|
||
with session_factory() as db:
|
||
titel = {e.titel for e in db.query(Neuheit).all()}
|
||
assert titel == {"Catan", "Ein ganz neues Spiel"}
|
||
neu = db.query(Neuheit).filter_by(titel="Ein ganz neues Spiel").one()
|
||
assert neu.bgg_id is None and neu.quelle == "spielbox"
|
||
|
||
|
||
def test_zweiter_lauf_aktualisiert_statt_duplikat(session_factory):
|
||
quelle = StatischerAdapter(
|
||
"brettspielbox",
|
||
[QuellenTreffer(titel="Fadenfroh", verlag="1 More Time Games",
|
||
quellen_url="https://brettspielbox.de/fadenfroh")],
|
||
)
|
||
service = _service(session_factory, (quelle,))
|
||
service.laufe()
|
||
|
||
zeilen = service.laufe()
|
||
|
||
assert zeilen[0].aktualisiert == 1
|
||
assert zeilen[0].neu == 0
|
||
with session_factory() as db:
|
||
assert db.query(Neuheit).count() == 1
|
||
|
||
|
||
def test_dedup_konflikt_wird_gefiltert_registry_api(session_factory):
|
||
class FakeDedup:
|
||
async def check_titel(self, titel, verlag, bgg_id, **kwargs):
|
||
class Ergebnis:
|
||
hat_konflikte = titel.startswith("Konflikt")
|
||
|
||
return Ergebnis()
|
||
|
||
quelle = StatischerAdapter(
|
||
"cliquenabend",
|
||
[
|
||
QuellenTreffer(titel="Konfliktfall", verlag="V", quellen_url="u"),
|
||
QuellenTreffer(titel="Saubere Ware", verlag="V", quellen_url="u"),
|
||
],
|
||
)
|
||
service = _service(session_factory, (quelle,), dedup=FakeDedup())
|
||
zeilen = service.laufe()
|
||
|
||
assert zeilen[0].gefiltert == 1
|
||
assert zeilen[0].neu == 1
|
||
assert [e.titel for e in alle(session_factory)] == ["Saubere Ware"]
|
||
|
||
|
||
def test_dedup_ausfall_blockiert_nicht(session_factory):
|
||
class KaputterDedup:
|
||
async def check_titel(self, titel, verlag, bgg_id, **kwargs):
|
||
raise RuntimeError("dedup nicht erreichbar")
|
||
|
||
quelle = StatischerAdapter(
|
||
"cliquenabend",
|
||
[QuellenTreffer(titel="Läuft trotzdem", verlag="V", quellen_url="u")],
|
||
)
|
||
service = _service(session_factory, (quelle,), dedup=KaputterDedup())
|
||
zeilen = service.laufe()
|
||
assert zeilen[0].neu == 1
|
||
|
||
|
||
def alle(session_factory):
|
||
from sqlalchemy import select
|
||
|
||
with session_factory() as db:
|
||
return list(db.scalars(select(Neuheit)).all())
|
||
|
||
|
||
def test_fehler_isolation_eine_quelle_crasht_andere_laeuft_weiter(session_factory):
|
||
kaputt = KaputterAdapter("spielbox")
|
||
gut = StatischerAdapter(
|
||
"cliquenabend",
|
||
[QuellenTreffer(titel="Galactic Cruise", verlag="Dranda Games",
|
||
autor="T.K. King", erscheinungsdatum_oder_quartal="15.10.25",
|
||
quellen_url="https://www.cliquenabend.de/spiele/x")],
|
||
)
|
||
service = _service(session_factory, (kaputt, gut))
|
||
|
||
zeilen = service.laufe()
|
||
|
||
assert zeilen[0].fehleranzahl == 1
|
||
assert "Seite umgebaut" in zeilen[0].fehlermeldung
|
||
assert zeilen[1].neu == 1 # andere Quelle lief normal weiter
|
||
status_kaputt = db_status(session_factory, "spielbox")
|
||
assert status_kaputt.fehlermeldung and status_kaputt.anzahl_fehler == 1
|
||
status_gut = db_status(session_factory, "cliquenabend")
|
||
assert status_gut.anzahl_neu == 1 and not status_gut.fehlermeldung
|
||
|
||
|
||
def test_env_schalter_deaktiviert_einzelne_quelle(session_factory, monkeypatch):
|
||
monkeypatch.setenv("SPIELE_NEUHEITEN_QUELLE_SPIELBOX_AKTIV", "0")
|
||
|
||
aus = StatischerAdapter("spielbox", [QuellenTreffer(titel="X", quellen_url="u")])
|
||
an = StatischerAdapter("cliquenabend", [QuellenTreffer(titel="Y", quellen_url="u")])
|
||
|
||
class BeobachtenderService(WebQuellenSyncService):
|
||
aufgerufen = []
|
||
|
||
def _client_fabrik_guard(self): # pragma: no cover - nur Signal
|
||
raise AssertionError("ausgeschaltete Quelle darf keinen Client bauen")
|
||
|
||
service = _service(session_factory, (aus, an))
|
||
zeilen = service.laufe()
|
||
|
||
assert zeilen[0].aktiv is False
|
||
assert zeilen[0].neu == 0
|
||
assert zeilen[1].neu == 1 # angeschaltete Quelle läuft
|
||
assert quelle_aktiv("spielbox") is False
|
||
assert quelle_aktiv("brettspielbox") is True # Standard „1“
|
||
|
||
|
||
def test_ui_schalter_uebersteuert_env(session_factory, monkeypatch):
|
||
"""Der DB-Schalter schaltet eine per Env aktive Quelle ab (und umgekehrt)."""
|
||
monkeypatch.delenv("SPIELE_NEUHEITEN_QUELLE_SPIELBOX_AKTIV", raising=False)
|
||
|
||
spielbox = StatischerAdapter("spielbox", [QuellenTreffer(titel="X", quellen_url="u")])
|
||
service = _service(session_factory, (spielbox,))
|
||
|
||
# Standard: an
|
||
assert service.quelle_laeuft("spielbox") is True
|
||
|
||
# Schalter „aus“ → Sync überspringt die Quelle
|
||
service.schalter_gesetzt("spielbox", False)
|
||
assert service.quelle_laeuft("spielbox") is False
|
||
zeilen = service.laufe()
|
||
assert zeilen[0].aktiv is False and zeilen[0].neu == 0
|
||
|
||
# Schalter „an“ → läuft wieder
|
||
service.schalter_gesetzt("spielbox", True)
|
||
assert service.quelle_laeuft("spielbox") is True
|
||
zeilen = service.laufe()
|
||
assert zeilen[0].aktiv is True and zeilen[0].neu == 1
|
||
|
||
|
||
def test_einzellauf_ignoriert_schalter(session_factory, monkeypatch):
|
||
"""`laufе(nur_quellen=…)` läuft auch bei ausgeschalteter Quelle."""
|
||
monkeypatch.delenv("SPIELE_NEUHEITEN_QUELLE_SPIELBOX_AKTIV", raising=False)
|
||
|
||
spielbox = StatischerAdapter("spielbox", [QuellenTreffer(titel="X", quellen_url="u")])
|
||
andere = StatischerAdapter("cliquenabend", [QuellenTreffer(titel="Y", quellen_url="u")])
|
||
service = _service(session_factory, (spielbox, andere))
|
||
|
||
service.schalter_gesetzt("spielbox", False)
|
||
|
||
zeilen = service.laufe(nur_quellen={"spielbox"})
|
||
assert len(zeilen) == 1
|
||
assert zeilen[0].quelle == "spielbox"
|
||
assert zeilen[0].aktiv is True
|
||
assert zeilen[0].neu == 1
|
||
|
||
|
||
def test_validatoren_werden_persistiert_und_wieder_verwendet(session_factory):
|
||
"""304-Behandlung über Läufe hinweg: ETag wird gespeichert und gesendet."""
|
||
import tests._neuheiten as h
|
||
|
||
basis_url = "https://www.spielbox.de/neuheiten-neuheiten"
|
||
uhr = FakeUhr()
|
||
|
||
class ValidatorenTransport(httpx.BaseTransport):
|
||
def __init__(self):
|
||
self.etag_gesendet: list[str | None] = []
|
||
self.runde = 0
|
||
|
||
def handle_request(self, request):
|
||
self.etag_gesendet.append(request.headers.get("if-none-match"))
|
||
self.runde += 1
|
||
if self.runde == 1:
|
||
return httpx.Response(200, content=b"<div class='item'></div>",
|
||
headers={"ETag": '"runde-1"'},
|
||
request=httpx.Request("GET", basis_url))
|
||
return httpx.Response(304, request=httpx.Request("GET", basis_url))
|
||
|
||
transport = ValidatorenTransport()
|
||
|
||
def fabrik(validatoren):
|
||
return WebQuellenClient(transport=transport, schlaf=lambda s: None,
|
||
uhr=uhr, validatoren=validatoren)
|
||
|
||
quelle = type("SpielboxPersistenz", (), {
|
||
"name": "spielbox",
|
||
"anzeigename": "Spielbox",
|
||
"start_urls": (basis_url,),
|
||
})
|
||
# Adapter-Klasse so bauen, dass sie den Transport-Parser nutzt:
|
||
from tests._neuheiten import SpielboxQuelle as SB
|
||
|
||
class PersistenteSpielbox(SB):
|
||
pass
|
||
|
||
PersistenteSpielbox.name = "spielbox"
|
||
PersistenteSpielbox.start_urls = (basis_url,)
|
||
|
||
service = WebQuellenSyncService(
|
||
session_factory, adapter_klassen=(PersistenteSpielbox,),
|
||
dedup=None, client_fabrik=fabrik,
|
||
)
|
||
|
||
erste = service.laufe()
|
||
zweite = service.laufe()
|
||
|
||
assert erste[0].unveraendert is False
|
||
assert transport.etag_gesendet[0] is None
|
||
assert zweite[0].unveraendert is True # 304 → Quelle übersprungen
|
||
assert transport.etag_gesendet[1] == '"runde-1"' # gespeicherter Validator
|
||
statuszeile = db_status(session_factory, "spielbox")
|
||
assert statuszeile.validatoren[basis_url][0] == '"runde-1"'
|
||
|
||
|
||
# ---------- UI-Übersicht ----------
|
||
|
||
|
||
def _test_app_und_client(monkeypatch, tmp_path):
|
||
from fastapi.testclient import TestClient
|
||
|
||
from redaktionskern.app import create_app
|
||
from redaktionskern.config import Settings
|
||
from tests.conftest import ADMIN_PASSWORD, PROJEKT_WURZEL
|
||
|
||
settings = Settings(
|
||
database_url=f"sqlite:///{tmp_path/'q.db'}",
|
||
session_secret="test-secret",
|
||
initial_admin_password=ADMIN_PASSWORD,
|
||
plugins_dir=PROJEKT_WURZEL / "plugins",
|
||
)
|
||
app = create_app(settings)
|
||
client = TestClient(app)
|
||
antwort = client.post("/login", data={"username": "admin", "password": ADMIN_PASSWORD},
|
||
follow_redirects=False)
|
||
assert antwort.status_code == 303
|
||
return app, client
|
||
|
||
|
||
def test_ui_zeigt_sync_uebersicht_der_quellen(monkeypatch, tmp_path):
|
||
monkeypatch.setenv("SPIELE_NEUHEITEN_QUELLE_SPIELBOX_AKTIV", "0")
|
||
app, client = _test_app_und_client(monkeypatch, tmp_path)
|
||
|
||
antwort = client.get("/neuheiten")
|
||
assert antwort.status_code == 200
|
||
for text in ("Web-Quellen", "spielbox.de Neuheiten", "Brettspielbox Messevorschau",
|
||
"SPIEL Essen Neuheiten", "Cliquenabend Messe-Übersicht"):
|
||
assert text in antwort.text, f"Fehlt: {text!r}"
|
||
|
||
|
||
def test_manueller_quellen_sync_endpunkt_mit_fakes(monkeypatch, tmp_path):
|
||
app, client = _test_app_und_client(monkeypatch, tmp_path)
|
||
plugin = app.state.registry.get("neuheiten")
|
||
|
||
fake_zeile = QuellenLaufZeile("spielbox", "spielbox.de Neuheiten")
|
||
fake_zeile.neu = 3
|
||
monkeypatch.setattr(plugin, "_quellen_sync_ausfuehren", lambda: [fake_zeile])
|
||
|
||
antwort = client.post("/neuheiten/quellen-sync", follow_redirects=False)
|
||
assert antwort.status_code == 303
|
||
assert "/neuheiten" in antwort.headers["location"]
|
||
# Meldung steckt im Redirect-Ziel — der Folge-Request muss die
|
||
# Location (inkl. ?meldung=…) abrufen, nicht die nackte Liste.
|
||
folge = client.get(antwort.headers["location"])
|
||
assert "Web-Quellen-Sync abgeschlossen" in folge.text
|
||
assert "3 neu" in folge.text
|
||
|
||
|
||
def test_rezensent_kann_quellen_sync_nicht_starten(monkeypatch, tmp_path):
|
||
from tests.conftest import lege_benutzer_an, melde_an
|
||
|
||
app, client = _test_app_und_client(monkeypatch, tmp_path)
|
||
lege_benutzer_an(app, "resi", "rezensent")
|
||
|
||
melde_an(client, username="resi", password="test-12345678")
|
||
antwort = client.post("/neuheiten/quellen-sync", follow_redirects=False)
|
||
assert antwort.status_code == 403
|
||
|
||
|
||
# ---------- Migration: bgg_id nullable ----------
|
||
|
||
|
||
def test_migration_0002_macht_bgg_id_nullable_erhaelt_daten(tmp_path):
|
||
"""Bestehende DB mit NOT NULL-bgg_id wird migriert, Daten bleiben erhalten."""
|
||
from sqlalchemy import create_engine, inspect, text
|
||
|
||
from redaktionskern.migrationen import run_migrations
|
||
from tests._neuheiten import models as m
|
||
|
||
engine = create_engine(f"sqlite:///{tmp_path/'mig.db'}")
|
||
with engine.begin() as conn:
|
||
conn.exec_driver_sql("""
|
||
CREATE TABLE neuheiten (
|
||
id INTEGER NOT NULL PRIMARY KEY,
|
||
titel VARCHAR(300) NOT NULL,
|
||
verlag VARCHAR(300),
|
||
autor VARCHAR(300),
|
||
erscheinungsjahr INTEGER,
|
||
bgg_id INTEGER NOT NULL,
|
||
status VARCHAR(50),
|
||
quelle VARCHAR(100),
|
||
erstellt_am DATETIME DEFAULT (CURRENT_TIMESTAMP),
|
||
aktualisiert_am DATETIME DEFAULT (CURRENT_TIMESTAMP),
|
||
CONSTRAINT uq_bgg UNIQUE (bgg_id)
|
||
)
|
||
""")
|
||
conn.exec_driver_sql(
|
||
"INSERT INTO neuheiten (titel, bgg_id, status, quelle) "
|
||
"VALUES ('Catan', 13, 'neuheit', 'boardgamegeek')"
|
||
)
|
||
plugin_name = "neuheiten"
|
||
|
||
plugin_instanz = _plugin_instance()
|
||
migrationen = plugin_instanz.migrations()
|
||
nur_alt = [migr for migr in migrationen if migr.version == "0001_neuheiten_tabelle"]
|
||
with engine.begin() as conn:
|
||
run_migrations(conn, plugin_name, nur_alt) # Stand: alte Version
|
||
with engine.begin() as conn:
|
||
neue = run_migrations(conn, plugin_name, migrationen)
|
||
|
||
assert "0002_bgg_id_nullable" in neue
|
||
inspektor = inspect(engine)
|
||
spalten = {c["name"]: c for c in inspektor.get_columns("neuheiten")}
|
||
assert spalten["bgg_id"]["nullable"] is True
|
||
with engine.connect() as conn:
|
||
zeile = conn.execute(text("SELECT titel, bgg_id FROM neuheiten")).one()
|
||
assert zeile == ("Catan", 13)
|
||
# NULL-bgg_id ist mehrfach erlaubt:
|
||
conn.execute(text(
|
||
"INSERT INTO neuheiten (titel, bgg_id, status, quelle) "
|
||
"VALUES ('Web-Titel A', NULL, 'neuheit', 'spielbox')"
|
||
))
|
||
conn.execute(text(
|
||
"INSERT INTO neuheiten (titel, bgg_id, status, quelle) "
|
||
"VALUES ('Web-Titel B', NULL, 'neuheit', 'spielbox')"
|
||
))
|
||
|
||
|
||
def _plugin_instance():
|
||
from tests._neuheiten import sys_modules_plugin
|
||
|
||
return sys_modules_plugin()
|
||
|
||
|
||
def test_migration_0003_legt_quellen_status_tabelle_an(tmp_path):
|
||
from sqlalchemy import create_engine, inspect
|
||
|
||
from redaktionskern.migrationen import run_migrations
|
||
|
||
engine = create_engine(f"sqlite:///{tmp_path/'mig3.db'}")
|
||
plugin_instanz = _plugin_instance()
|
||
with engine.begin() as conn:
|
||
run_migrations(conn, "neuheiten", plugin_instanz.migrations())
|
||
inspektor = inspect(engine)
|
||
assert "neuheiten_quellen_status" in inspektor.get_table_names()
|
||
|
||
|
||
# ---------- Registrierung ----------
|
||
|
||
|
||
def test_alle_vier_quellen_registriert():
|
||
namen = {klasse.name for klasse in ADAPTER_KLASSEN}
|
||
assert namen == {"spielbox", "brettspielbox", "spielessen", "cliquenabend"}
|