483 lines
17 KiB
Python
483 lines
17 KiB
Python
"""Tests für die Coverbilder-Erweiterung (bild_url).
|
|
|
|
Abgedeckt:
|
|
- Migrationen: Spalte bild_url existiert in allen vier Tabellen
|
|
(neuheiten, planungsliste, archiv_neuheiten, archiv_planung);
|
|
bestehende Zeilen bleiben beim ALTER TABLE erhalten.
|
|
- BGG-Parsing: `<image>` wird bevorzugt, sonst `<thumbnail>`;
|
|
protokoll-relative URLs werden auf https normalisiert.
|
|
- Sync-Upsert: bild_url wird angelegt und beim Update mitgepflegt;
|
|
ein fehlendes Bild in der Antwort löscht kein gespeichertes.
|
|
- Web-Quellen: img im Listenelement wird übernommen und gespeichert.
|
|
- Templates: img-Tag mit loading="lazy"/decoding="async" bei Bild,
|
|
Platzhalter-Div ohne Bild; große Ansicht in Detailseiten.
|
|
|
|
Kein echter Netzwerkverkehr — XML/HTML sind gemockt.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine, select
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from redaktionskern.db import Base
|
|
from redaktionskern.migrationen import run_migrations
|
|
|
|
from tests._neuheiten import (
|
|
BggSpiel,
|
|
Neuheit,
|
|
QuellenTreffer,
|
|
SpielboxQuelle,
|
|
WebQuellenClient,
|
|
WebQuellenSyncService,
|
|
parse_things,
|
|
)
|
|
from tests.conftest import melde_an
|
|
|
|
# ---------- Hilfsbausteine ----------
|
|
|
|
|
|
class FakeUhr:
|
|
def __init__(self) -> None:
|
|
self.zeit = 0.0
|
|
|
|
def __call__(self) -> float:
|
|
return self.zeit
|
|
|
|
|
|
@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 _spalten(conn, tabelle: str) -> set[str]:
|
|
info = conn.exec_driver_sql(f"PRAGMA table_info({tabelle})").fetchall()
|
|
return {zeile[1] for zeile in info}
|
|
|
|
|
|
def plugin_modul(app, name: str):
|
|
"""Das Modul des geladenen Plugins (Registry bevorzugt)."""
|
|
plugin_instanz = app.state.registry.get(name) or sys.modules.get(
|
|
f"spiele_redaktion_plugins.{name}"
|
|
)
|
|
assert plugin_instanz is not None, f"Plugin {name} nicht geladen"
|
|
return sys.modules[type(plugin_instanz).__module__]
|
|
|
|
|
|
# ---------- Migrationen ----------
|
|
|
|
|
|
def test_migrations_legen_bild_url_in_allen_tabellen_an(app):
|
|
with app.state.engine.connect() as conn:
|
|
for tabelle in ("neuheiten", "planungsliste", "archiv_neuheiten",
|
|
"archiv_planung"):
|
|
assert "bild_url" in _spalten(conn, tabelle), \
|
|
f"bild_url fehlt in {tabelle}"
|
|
|
|
|
|
def test_bild_url_migration_erhaelt_bestehende_daten(app, tmp_path):
|
|
"""ALTER-TABLE-Pfad: alte DB ohne bild_url → Migration fügt die Spalte
|
|
hinzu und erhält alle bestehenden Zeilen."""
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'mig.db'}")
|
|
|
|
# Alte Migrationen (ohne die bild_url-Schritte) je Plugin ausführen.
|
|
alte_schritte = {
|
|
"neuheiten": [m for m in app.state.registry.get("neuheiten").migrations()
|
|
if not m.version.startswith("0004")],
|
|
"planung": [m for m in app.state.registry.get("planung").migrations()
|
|
if not m.version.startswith("0002")],
|
|
"archiv": [m for m in app.state.registry.get("archiv").migrations()
|
|
if not m.version.startswith("0002")],
|
|
}
|
|
with engine.begin() as conn:
|
|
for plugin_name, schritte in alte_schritte.items():
|
|
run_migrations(conn, plugin_name, schritte)
|
|
# Alten Stand simulieren: die aktuellen Modelle erzeugen die Spalte
|
|
# schon mit, daher für den Test wieder entfernen.
|
|
for tabelle in ("neuheiten", "planungsliste",
|
|
"archiv_neuheiten", "archiv_planung"):
|
|
conn.exec_driver_sql(f"ALTER TABLE {tabelle} DROP COLUMN bild_url")
|
|
assert "bild_url" not in _spalten(conn, tabelle)
|
|
|
|
# Bestehende Zeilen einfügen (NOT-NULL-Pflichtspalten gesetzt; die
|
|
# Defaults der Modelle sind Python-seitig, daher hier explizit).
|
|
with engine.begin() as conn:
|
|
conn.exec_driver_sql(
|
|
"INSERT INTO neuheiten (titel, status, quelle)"
|
|
" VALUES ('Alter BGG-Titel', 'neuheit', 'boardgamegeek')"
|
|
)
|
|
conn.exec_driver_sql(
|
|
"INSERT INTO planungsliste"
|
|
" (titel, rezensent_id, status, quelle, ausgabe)"
|
|
" VALUES ('Alter Plan', 1, 'offen', 'manuell', '')"
|
|
)
|
|
conn.exec_driver_sql(
|
|
"INSERT INTO archiv_neuheiten"
|
|
" (quell_id, titel, status, quelle, archiviert_von, grund)"
|
|
" VALUES (1, 'Alt Neuheit', 'neuheit', 'boardgamegeek', 'System', '')"
|
|
)
|
|
conn.exec_driver_sql(
|
|
"INSERT INTO archiv_planung"
|
|
" (quell_id, titel, rezensent_id, status, quelle, ausgabe,"
|
|
" archiviert_von, grund)"
|
|
" VALUES (2, 'Alt Planung', 1, 'offen', 'manuell', '', 'System', '')"
|
|
)
|
|
|
|
# Neue bild_url-Migrationen anwenden (nur diese sind noch offen).
|
|
neue_schritte = {
|
|
"neuheiten": [m for m in app.state.registry.get("neuheiten").migrations()
|
|
if m.version == "0004_bild_url"],
|
|
"planung": [m for m in app.state.registry.get("planung").migrations()
|
|
if m.version == "0002_bild_url"],
|
|
"archiv": [m for m in app.state.registry.get("archiv").migrations()
|
|
if m.version == "0002_bild_url"],
|
|
}
|
|
with engine.begin() as conn:
|
|
for plugin_name, schritte in neue_schritte.items():
|
|
run_migrations(conn, plugin_name, schritte)
|
|
|
|
# Spalte vorhanden, Daten vollständig erhalten.
|
|
with engine.connect() as conn:
|
|
for tabelle in ("neuheiten", "planungsliste",
|
|
"archiv_neuheiten", "archiv_planung"):
|
|
assert "bild_url" in _spalten(conn, tabelle)
|
|
titel = {
|
|
zeile[0]
|
|
for zeile in conn.exec_driver_sql(
|
|
"SELECT titel FROM neuheiten"
|
|
).fetchall()
|
|
}
|
|
assert titel == {"Alter BGG-Titel"}
|
|
assert conn.exec_driver_sql(
|
|
"SELECT COUNT(*) FROM planungsliste"
|
|
).scalar() == 1
|
|
assert conn.exec_driver_sql(
|
|
"SELECT COUNT(*) FROM archiv_neuheiten"
|
|
).scalar() == 1
|
|
assert conn.exec_driver_sql(
|
|
"SELECT COUNT(*) FROM archiv_planung"
|
|
).scalar() == 1
|
|
|
|
|
|
# ---------- BGG-Parsing ----------
|
|
|
|
THING_MIT_BEIDEN = """<?xml version="1.0"?>
|
|
<items termsofuse="https://boardgamegeek.com/xmlapi/termsofuse">
|
|
<item type="boardgame" id="13">
|
|
<thumbnail>https://cf.geekdo-images.com/klein_thumb.jpg</thumbnail>
|
|
<image>https://cf.geekdo-images.com/gross.png</image>
|
|
<name type="primary" value="Catan"/>
|
|
</item>
|
|
</items>
|
|
"""
|
|
|
|
THING_NUR_THUMBNAIL = """<?xml version="1.0"?>
|
|
<items termsofuse="https://boardgamegeek.com/xmlapi/termsofuse">
|
|
<item type="boardgame" id="13">
|
|
<thumbnail>https://cf.geekdo-images.com/nur_thumb.jpg</thumbnail>
|
|
<name type="primary" value="Catan"/>
|
|
</item>
|
|
</items>
|
|
"""
|
|
|
|
THING_PROTOKOLL_RELATIV = """<?xml version="1.0"?>
|
|
<items termsofuse="https://boardgamegeek.com/xmlapi/termsofuse">
|
|
<item type="boardgame" id="13">
|
|
<image>//cf.geekdo-images.com/relativ.jpg</image>
|
|
<name type="primary" value="Catan"/>
|
|
</item>
|
|
</items>
|
|
"""
|
|
|
|
THING_OHNE_BILD = """<?xml version="1.0"?>
|
|
<items termsofuse="https://boardgamegeek.com/xmlapi/termsofuse">
|
|
<item type="boardgame" id="13">
|
|
<name type="primary" value="Catan"/>
|
|
</item>
|
|
</items>
|
|
"""
|
|
|
|
|
|
def test_parse_things_bevorzugt_grosses_image():
|
|
spiele = parse_things(THING_MIT_BEIDEN)
|
|
assert len(spiele) == 1
|
|
assert spiele[0].bild_url == "https://cf.geekdo-images.com/gross.png"
|
|
|
|
|
|
def test_parse_things_faellt_auf_thumbnail_zurueck():
|
|
spiele = parse_things(THING_NUR_THUMBNAIL)
|
|
assert spiele[0].bild_url == "https://cf.geekdo-images.com/nur_thumb.jpg"
|
|
|
|
|
|
def test_parse_things_normalisiert_protokoll_relative_urls():
|
|
spiele = parse_things(THING_PROTOKOLL_RELATIV)
|
|
assert spiele[0].bild_url == "https://cf.geekdo-images.com/relativ.jpg"
|
|
|
|
|
|
def test_parse_things_ohne_bild_liefert_none():
|
|
spiele = parse_things(THING_OHNE_BILD)
|
|
assert spiele[0].bild_url is None
|
|
|
|
|
|
# ---------- Sync-Upsert pflegt bild_url ----------
|
|
|
|
|
|
def test_upsert_legt_und_pflegt_bild_url(session_factory):
|
|
from tests._neuheiten import speichere_oder_aktualisiere
|
|
|
|
spiel = BggSpiel(
|
|
bgg_id=13, titel="Catan", verlag="KOSMOS", autor="Klaus Teuber",
|
|
erscheinungsjahr=1995, bild_url="https://cf.geekdo-images.com/a.jpg",
|
|
)
|
|
with session_factory() as db:
|
|
assert speichere_oder_aktualisiere(db, spiel) is True
|
|
db.commit()
|
|
|
|
# Update: neues Bild wird übernommen …
|
|
update = BggSpiel(
|
|
bgg_id=13, titel="Catan", verlag="KOSMOS", autor="Klaus Teuber",
|
|
erscheinungsjahr=1995, bild_url="https://cf.geekdo-images.com/b.jpg",
|
|
)
|
|
assert speichere_oder_aktualisiere(db, update) is False
|
|
db.commit()
|
|
eintrag = db.scalar(select(Neuheit).where(Neuheit.bgg_id == 13))
|
|
assert eintrag.bild_url == "https://cf.geekdo-images.com/b.jpg"
|
|
|
|
# … ein fehlendes Bild in der Antwort löscht das gespeicherte nicht.
|
|
ohne = BggSpiel(
|
|
bgg_id=13, titel="Catan", verlag="KOSMOS", autor="Klaus Teuber",
|
|
erscheinungsjahr=1995, bild_url=None,
|
|
)
|
|
speichere_oder_aktualisiere(db, ohne)
|
|
db.commit()
|
|
db.expire_all()
|
|
eintrag = db.scalar(select(Neuheit).where(Neuheit.bgg_id == 13))
|
|
assert eintrag.bild_url == "https://cf.geekdo-images.com/b.jpg"
|
|
|
|
|
|
# ---------- Web-Quellen: Bild aus Listenelement übernehmen ----------
|
|
|
|
SPIELBOX_MIT_BILD = b"""
|
|
<div class="blog">
|
|
<div class="item column-1">
|
|
<div class="page-header"><h2><a href="/beitrag/x">Pegasus Spiele: Next Station Berlin</a></h2></div>
|
|
<img src="/media/next-station.jpg" alt="Next Station Berlin">
|
|
<p><strong>24.07.2026</strong> - Vorstellung.</p>
|
|
</div>
|
|
<div class="item column-2">
|
|
<div class="page-header"><h2><a href="/beitrag/y">MICROMACRO Kids</a></h2></div>
|
|
<p><strong>02.03.2026</strong> - Neu.</p>
|
|
</div>
|
|
</div>
|
|
"""
|
|
|
|
|
|
def test_spielbox_adapter_uebernimmt_img_url():
|
|
quelle = SpielboxQuelle(WebQuellenClient(schlaf=lambda s: None, uhr=FakeUhr()))
|
|
treffer, folge_links = quelle.seite_verarbeiten(
|
|
SPIELBOX_MIT_BILD, "https://www.spielbox.de/neuheiten-neue-spiele"
|
|
)
|
|
assert [t.titel for t in treffer] == [
|
|
"Pegasus Spiele: Next Station Berlin", "MICROMACRO Kids",
|
|
]
|
|
# Erstes Element liefert ein Bild, das zweite keins → leer lassen.
|
|
assert treffer[0].bild_url == "https://www.spielbox.de/media/next-station.jpg"
|
|
assert treffer[1].bild_url is None
|
|
assert folge_links == []
|
|
|
|
|
|
class _StatischerAdapter:
|
|
"""Adapter-Ersatz mit fester Trefferliste (für Service-Tests)."""
|
|
|
|
name = "spielbox"
|
|
anzeigename = "spielbox Test"
|
|
start_urls = ("https://spielbox.example.org/liste",)
|
|
|
|
def __init__(self, treffer):
|
|
self._treffer = treffer
|
|
|
|
def __call__(self, client):
|
|
return self
|
|
|
|
def sammle(self):
|
|
from tests._neuheiten import SammelErgebnis
|
|
|
|
return SammelErgebnis(treffer=list(self._treffer), seiten=1)
|
|
|
|
|
|
def test_quellen_sync_speichert_und_pflegt_bild_url(session_factory):
|
|
quelle = _StatischerAdapter([
|
|
QuellenTreffer(
|
|
titel="Ringträger", verlag="Schmidt Spiele",
|
|
quellen_url="https://x.example.org/r",
|
|
bild_url="https://bilder.example.org/ringtraeger.jpg",
|
|
),
|
|
])
|
|
service = WebQuellenSyncService(
|
|
session_factory,
|
|
adapter_klassen=(quelle,),
|
|
client_fabrik=lambda validatoren: WebQuellenClient(
|
|
schlaf=lambda s: None, uhr=FakeUhr(), validatoren=validatoren
|
|
),
|
|
)
|
|
zeilen = service.laufe()
|
|
assert zeilen[0].neu == 1
|
|
|
|
with session_factory() as db:
|
|
eintrag = db.query(Neuheit).one()
|
|
assert eintrag.quelle == "spielbox"
|
|
assert eintrag.bild_url == "https://bilder.example.org/ringtraeger.jpg"
|
|
|
|
# Zweiter Lauf aktualisiert das Bild statt es zu duplizieren.
|
|
quelle._treffer = [
|
|
QuellenTreffer(
|
|
titel="Ringträger", verlag="Schmidt Spiele",
|
|
quellen_url="https://x.example.org/r",
|
|
bild_url="https://bilder.example.org/ringtraeger-neu.jpg",
|
|
),
|
|
]
|
|
zeilen = service.laufe()
|
|
assert zeilen[0].aktualisiert == 1 and zeilen[0].neu == 0
|
|
with session_factory() as db:
|
|
assert db.query(Neuheit).count() == 1
|
|
assert db.query(Neuheit).one().bild_url == \
|
|
"https://bilder.example.org/ringtraeger-neu.jpg"
|
|
|
|
|
|
def test_spielessen_extrahiert_bild_aus_bilder_feld(app):
|
|
klasse = plugin_modul(app, "neuheiten").quellen.spielessen.SpielEssenQuelle
|
|
adapter = klasse(client=None) # Parsing-Hilfen brauchen keinen Client
|
|
|
|
produkt = {
|
|
"TITEL": "-60°",
|
|
"BILDER": [{"URL": "//maps.eyeled-services.de/media/bild.jpg"}],
|
|
}
|
|
treffer = adapter._produkt(produkt, {})
|
|
assert treffer is not None
|
|
assert treffer.bild_url == "https://maps.eyeled-services.de/media/bild.jpg"
|
|
|
|
treffer_ohne = adapter._produkt({"TITEL": "X", "BILDER": []}, {})
|
|
assert treffer_ohne is not None
|
|
assert treffer_ohne.bild_url is None
|
|
|
|
|
|
# ---------- Planung: Übernahme trägt das Bild mit ----------
|
|
|
|
|
|
def test_uebernehmen_kopiert_bild_url_in_die_planung(app, client):
|
|
with app.state.session_factory() as db:
|
|
db.add(
|
|
Neuheit(
|
|
titel="Cascadia",
|
|
bgg_id=295947,
|
|
status="neuheit",
|
|
bild_url="https://cf.geekdo-images.com/cascadia.jpg",
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
melde_an(client)
|
|
antwort = client.post("/planung/uebernehmen/1", follow_redirects=False)
|
|
assert antwort.status_code in (303, 200)
|
|
|
|
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
|
with app.state.session_factory() as db:
|
|
eintrag = db.scalars(select(Planungseintrag)).one()
|
|
assert eintrag.titel == "Cascadia"
|
|
assert eintrag.bild_url == "https://cf.geekdo-images.com/cascadia.jpg"
|
|
|
|
# Die Neuheit selbst ist jetzt „planung“ und behält ihr Bild.
|
|
neuheit = db.scalar(select(Neuheit).where(Neuheit.titel == "Cascadia"))
|
|
assert neuheit.status == "planung"
|
|
assert neuheit.bild_url == "https://cf.geekdo-images.com/cascadia.jpg"
|
|
|
|
|
|
# ---------- Template-Darstellung ----------
|
|
|
|
|
|
def test_neuheitenliste_rendert_img_mit_lazy_loading(app, client):
|
|
with app.state.session_factory() as db:
|
|
db.add(
|
|
Neuheit(
|
|
titel="Catan", bgg_id=13, status="neuheit",
|
|
bild_url="https://cf.geekdo-images.com/catan.jpg",
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
melde_an(client)
|
|
antwort = client.get("/neuheiten")
|
|
assert antwort.status_code == 200
|
|
text = antwort.text
|
|
assert '<img src="https://cf.geekdo-images.com/catan.jpg"' in text
|
|
assert 'loading="lazy"' in text
|
|
assert 'decoding="async"' in text
|
|
assert 'alt="Cover: Catan"' in text
|
|
# Fallback-Zeichen darf daneben nicht auftauchen, wenn Bilder da sind …
|
|
# (Doch: andere Einträge könnten bildlos sein — hier nur einer.)
|
|
|
|
# Mobil-Kartenansicht und Desktop-Tabelle sind beide im Markup.
|
|
assert 'md:hidden' in text
|
|
assert 'hidden md:block' in text
|
|
|
|
|
|
def test_neuheitenliste_rendert_platzhalter_ohne_bild(app, client):
|
|
with app.state.session_factory() as db:
|
|
db.add(Neuheit(titel="Ohne Bild", status="neuheit"))
|
|
db.commit()
|
|
|
|
melde_an(client)
|
|
antwort = client.get("/neuheiten")
|
|
assert antwort.status_code == 200
|
|
liste = antwort.text.split('<div id="neuheiten-liste"')[1].split("</main>")[0]
|
|
assert "<img" not in liste
|
|
assert "🎲" in antwort.text # Spielicon-Zeichen im Platzhalter-Div
|
|
|
|
|
|
def test_planungsliste_zeigt_cover(app, client):
|
|
from redaktionskern.auth.models import User
|
|
|
|
Planungseintrag = plugin_modul(app, "planung").Planungseintrag
|
|
with app.state.session_factory() as db:
|
|
benutzer = db.query(User).filter_by(username="admin").first()
|
|
db.add(
|
|
Planungseintrag(
|
|
titel="Cascadia",
|
|
rezensent_id=benutzer.id,
|
|
status="offen",
|
|
bild_url="https://cf.geekdo-images.com/cascadia.jpg",
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
melde_an(client)
|
|
antwort = client.get("/planung")
|
|
assert antwort.status_code == 200
|
|
assert 'src="https://cf.geekdo-images.com/cascadia.jpg"' in antwort.text
|
|
assert 'loading="lazy"' in antwort.text
|
|
assert 'decoding="async"' in antwort.text
|
|
|
|
|
|
def test_archiv_zeigt_cover(app, client):
|
|
ArchivNeuheit = plugin_modul(app, "archiv").ArchivNeuheit
|
|
with app.state.session_factory() as db:
|
|
db.add(
|
|
ArchivNeuheit(
|
|
quell_id=1, titel="Altspiel",
|
|
bild_url="https://cf.geekdo-images.com/altspiel.jpg",
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
melde_an(client)
|
|
antwort = client.get("/archiv")
|
|
assert antwort.status_code == 200
|
|
assert 'src="https://cf.geekdo-images.com/altspiel.jpg"' in antwort.text
|
|
assert 'loading="lazy"' in antwort.text
|