b0rbor4d/spiele-redaktion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8b38bc6c893c035d492ad8cb5a96546128716875
spiele-redaktion/tests/test_neuheiten_quellen.py

861 lines
30 KiB
Python
Raw Blame History

"""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&auml;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&auml;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 &amp; 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,- &euro;</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>&mdash;</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_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"}
Reference in New Issue View Git Blame Copy Permalink