Plugin neuheiten: Bearer-Token-Pflicht der BGG-XML-API2 (Fix für HTTP 401)

BoardGameGeek verlangt für die XML API2 jetzt Bearer-Authentifizierung
(Thread 3602374); ohne Token antwortet die API mit 401.

- BggClient: neuer Parameter token, sendet Authorization: Bearer … auf
  allen Requests; Token kommt aus SPIELE_BGG_TOKEN (vor jedem Sync gelesen)
- HTTP 401 → klarer BggAuthFehler ('BGG hat die Anfrage abgelehnt (401) —
  bitte gültigen API-Token in SPIELE_BGG_TOKEN hinterlegen.') statt
  generischer Meldung, kein Retry
- Ohne konfigurierten Token wird der Sync übersprungen ('Kein BGG-API-Token
  konfiguriert (SPIELE_BGG_TOKEN) — Sync übersprungen.'), statt Requests zu
  feuern; Meldung erscheint im Sync-Ergebnis, UI-Banner und Protokoll
- SyncService: BggAuthFehler bricht den Lauf ab (weitere Suchbegriffe
  scheitern zwangsläufig gleich), Teilergebnisse bleiben erhalten
- Tests: Bearer-Header, 401-Fall (Client/Sync/UI), Skip ohne Token;
  bestehende Sync-UI-Tests setzen Test-Token
- README: neue Env-Variable SPIELE_BGG_TOKEN + Hinweis auf Token-Pflicht
This commit is contained in:
Flo Hartmann
2026-08-21 23:14:58 +00:00
parent 55f6f32f73
commit 60c4b6e818
8 changed files with 259 additions and 9 deletions

View File

@@ -8,7 +8,14 @@ from __future__ import annotations
import httpx
import pytest
from tests._neuheiten import BggClient, BggFehler, parse_suche, parse_things
from tests._neuheiten import (
BggAuthFehler,
BggClient,
BggFehler,
MELDUNG_401,
parse_suche,
parse_things,
)
SUCHE_XML = """<?xml version="1.0" encoding="utf-8"?>
<items total="3" termsofuse="https://boardgamegeek.com/xmlapi/termsofuse">
@@ -61,9 +68,11 @@ class AufzeichnenderTransport(httpx.BaseTransport):
self.antworten = list(antworten)
self.uhr = uhr
self.anfragen: list[tuple[str, float]] = []
self.auth_header: list[str | None] = [] # Authorization je Request
def handle_request(self, request: httpx.Request) -> httpx.Response:
self.anfragen.append((str(request.url), self.uhr()))
self.auth_header.append(request.headers.get("Authorization"))
element = self.antworten.pop(0)
if isinstance(element, Exception):
raise element
@@ -157,6 +166,39 @@ def test_suche_sendet_type_boardgame():
assert "query=catan" in url
def test_token_wird_als_bearer_header_gesendet():
uhr = FakeUhr()
transport = AufzeichnenderTransport([_antwort(SUCHE_XML), _antwort(THING_XML)], uhr)
client = BggClient(
transport=transport,
token="geheim-123",
schlaf=lambda s: None,
uhr=uhr,
)
client.suche("catan")
client.details([13])
# Authorization-Header auf allen Requests (Suche und Thing):
assert transport.auth_header == ["Bearer geheim-123", "Bearer geheim-123"]
def test_leerer_token_gilt_als_nicht_konfiguriert():
uhr = FakeUhr()
transport = AufzeichnenderTransport([_antwort(SUCHE_XML)], uhr)
client = BggClient(
transport=transport,
token=" ",
schlaf=lambda s: None,
uhr=uhr,
)
assert client.token is None
client.suche("catan")
assert transport.auth_header == [None]
def test_details_sendet_thing_mit_ids_und_filtert_erweiterungen():
uhr = FakeUhr()
transport = AufzeichnenderTransport([_antwort(THING_XML)], uhr)
@@ -266,6 +308,34 @@ def test_4xx_fuehrt_zu_fehler_ohne_weitere_versuche():
assert wartezeiten == []
def test_401_fuehrt_zu_klarer_auth_fehlermeldung_ohne_retry():
uhr = FakeUhr()
wartezeiten: list[float] = []
transport = AufzeichnenderTransport(
[_antwort("unauthorized", status_code=401)], uhr
)
client = _client(transport, uhr, _fortschreitender_schlaf(uhr, wartezeiten))
with pytest.raises(BggAuthFehler) as aufgetreten:
client.suche("catan")
assert str(aufgetreten.value) == MELDUNG_401
assert MELDUNG_401.startswith("BGG hat die Anfrage abgelehnt (401)")
assert len(transport.anfragen) == 1 # kein Retry bei dauerhaftem 401
assert wartezeiten == []
def test_401_beim_thing_request_gleiche_meldung():
uhr = FakeUhr()
transport = AufzeichnenderTransport([_antwort("nope", status_code=401)], uhr)
client = _client(transport, uhr, lambda s: None)
with pytest.raises(BggAuthFehler) as aufgetreten:
client.details([13])
assert str(aufgetreten.value) == MELDUNG_401
def test_max_versuche_erschoepft_mit_fehler():
uhr = FakeUhr()
transport = AufzeichnenderTransport([_antwort("kaputt", status_code=500)], uhr)