"""Tests für den BGG-XML-API2-Client — ausschließlich mit gemockten Antworten.
Kein echter Netzwerkverkehr: Der HTTP-Transport wird durch Fakes ersetzt,
Uhr und Schlaf-Funktion sind injiziert.
"""
from __future__ import annotations
import httpx
import pytest
from tests._neuheiten import (
BggAuthFehler,
BggClient,
BggFehler,
MELDUNG_401,
parse_suche,
parse_things,
)
SUCHE_XML = """
-
-
-
"""
THING_XML = """
-
https://example.org/thumb.jpg
-
"""
class FakeUhr:
"""Steuerbare Monoton-Uhr."""
def __init__(self) -> None:
self.zeit = 0.0
def __call__(self) -> float:
return self.zeit
class AufzeichnenderTransport(httpx.BaseTransport):
"""Liefert eine feste Antwortkette; zeichnet URLs und Uhrzeiten auf."""
def __init__(self, antworten: list, uhr: FakeUhr):
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
if not self.antworten:
self.antworten.append(element) # letzte Antwort zyklisch wiederholen
return element
def _antwort(
xml: str, status_code: int = 200, headers: dict | None = None
) -> httpx.Response:
return httpx.Response(
status_code,
content=xml.encode("utf-8"),
headers=headers or {},
request=httpx.Request("GET", "https://boardgamegeek.com/xmlapi2/test"),
)
def _client(
transport: httpx.BaseTransport, uhr: FakeUhr, schlaf
) -> BggClient:
return BggClient(
transport=transport,
mindestabstand_sekunden=1.0,
backoff_basis_sekunden=0.5,
backoff_maximum_sekunden=4.0,
schlaf=schlaf,
uhr=uhr,
)
def _fortschreitender_schlaf(uhr: FakeUhr, wartezeiten: list[float]):
def schlaf(sekunden: float) -> None:
wartezeiten.append(sekunden)
uhr.zeit += sekunden
return schlaf
# ---------- Parsing ----------
def test_parse_suche_nur_brettspiele():
treffer = parse_suche(SUCHE_XML)
assert [t.bgg_id for t in treffer] == [13, 9999] # Erweiterung raus
assert treffer[0].titel == "Catan"
assert treffer[0].erscheinungsjahr == 1995
assert treffer[1].erscheinungsjahr is None # fehlendes Jahr toleriert
def test_parse_suche_umlaute_und_kaputtes_xml():
xml = ' '
treffer = parse_suche(xml)
assert treffer[0].titel == "Schätze von Trullala"
with pytest.raises(BggFehler):
parse_suche(" ")
def test_parse_things_felder_und_erweiterungsfilter():
spiele = parse_things(THING_XML)
assert len(spiele) == 1 # boardgameexpansion wird verworfen
spiel = spiele[0]
assert spiel.typ == "boardgame"
assert spiel.titel == "Catan"
assert spiel.verlag == "KOSMOS, Catan Studio"
assert spiel.autor == "Klaus Teuber"
assert spiel.erscheinungsjahr == 1995
def test_parse_things_leere_und_fehlerhafte_antwort():
assert parse_things('') == []
with pytest.raises(BggFehler):
parse_things("kein xml")
# ---------- HTTP-Verhalten ----------
def test_suche_sendet_type_boardgame():
uhr = FakeUhr()
transport = AufzeichnenderTransport([_antwort(SUCHE_XML)], uhr)
client = _client(transport, uhr, lambda s: None)
client.suche("catan")
url = transport.anfragen[0][0]
assert "/search" in url
assert "type=boardgame" in url
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)
client = _client(transport, uhr, lambda s: None)
spiele = client.details([13, 55])
url = transport.anfragen[0][0]
assert "/thing" in url
assert "id=13%2C55" in url or "id=13,55" in url
assert "type=boardgame" in url
assert [s.bgg_id for s in spiele] == [13]
def test_rate_limit_mindestens_eine_sekunde_zwischen_requests():
uhr = FakeUhr()
wartezeiten: list[float] = []
transport = AufzeichnenderTransport([_antwort(SUCHE_XML)], uhr)
client = _client(transport, uhr, _fortschreitender_schlaf(uhr, wartezeiten))
client.suche("a")
client.suche("b")
zeitpunkt_1 = transport.anfragen[0][1]
zeitpunkt_2 = transport.anfragen[1][1]
assert zeitpunkt_2 - zeitpunkt_1 >= 1.0
assert wartezeiten # es wurde aktiv gewartet
def test_retry_bei_5xx_mit_backoff_dann_erfolg():
uhr = FakeUhr()
wartezeiten: list[float] = []
transport = AufzeichnenderTransport(
[_antwort("server error", status_code=500),
_antwort("server error", status_code=500),
_antwort(SUCHE_XML)], uhr
)
client = _client(transport, uhr, _fortschreitender_schlaf(uhr, wartezeiten))
treffer = client.suche("catan")
assert len(treffer) == 2
assert len(transport.anfragen) == 3 # 2 × 500, dann Erfolg
# Backoff 0,5 s → Rate-Limit-Rest 0,5 s → Backoff 1,0 s:
assert wartezeiten == [0.5, 0.5, 1.0]
def test_202_wird_gemaess_retry_after_erneut_versucht():
uhr = FakeUhr()
wartezeiten: list[float] = []
transport = AufzeichnenderTransport(
[
_antwort("queued", status_code=202, headers={"Retry-After": "3"}),
_antwort(SUCHE_XML),
],
uhr,
)
client = _client(transport, uhr, _fortschreitender_schlaf(uhr, wartezeiten))
treffer = client.suche("catan")
assert len(treffer) == 2
assert wartezeiten == [3.0] # Retry-After wird beachtet
assert len(transport.anfragen) == 2
def test_429_ohne_retry_after_nutzt_backoff():
uhr = FakeUhr()
wartezeiten: list[float] = []
transport = AufzeichnenderTransport(
[_antwort("zu schnell", status_code=429), _antwort(SUCHE_XML)], uhr
)
client = _client(transport, uhr, _fortschreitender_schlaf(uhr, wartezeiten))
treffer = client.suche("catan")
assert len(treffer) == 2
# Backoff nach 429 plus Rate-Limit-Abstand vor dem zweiten Request:
assert wartezeiten == [0.5, 0.5]
def test_netzwerkfehler_wird_mit_backoff_wiederholt():
uhr = FakeUhr()
wartezeiten: list[float] = []
transport = AufzeichnenderTransport(
[httpx.ConnectError("Verbindung fehlgeschlagen"), _antwort(SUCHE_XML)], uhr
)
client = _client(transport, uhr, _fortschreitender_schlaf(uhr, wartezeiten))
treffer = client.suche("catan")
assert len(treffer) == 2
assert wartezeiten == [0.5]
assert len(transport.anfragen) == 2
def test_4xx_fuehrt_zu_fehler_ohne_weitere_versuche():
uhr = FakeUhr()
wartezeiten: list[float] = []
transport = AufzeichnenderTransport([_antwort("nope", status_code=404)], uhr)
client = _client(transport, uhr, _fortschreitender_schlaf(uhr, wartezeiten))
with pytest.raises(BggFehler):
client.suche("catan")
assert len(transport.anfragen) == 1
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)
client = BggClient(
transport=transport,
max_versuche=3,
backoff_basis_sekunden=0.1,
schlaf=lambda s: uhr.__setattr__("zeit", uhr.zeit + s),
uhr=uhr,
)
with pytest.raises(BggFehler):
client.suche("catan")
assert len(transport.anfragen) == 3
def test_details_batched_ueber_20_ids_pro_request():
uhr = FakeUhr()
aufgerufene_id_listen: list[list[str]] = []
def handler(request: httpx.Request) -> httpx.Response:
ids = request.url.params["id"].split(",")
aufgerufene_id_listen.append(ids)
items = "".join(
f' '
for i in ids
)
return _antwort(f"{items}")
client = BggClient(
transport=httpx.MockTransport(handler),
schlaf=lambda s: None,
uhr=uhr,
)
spiele = client.details(list(range(1, 26))) # 25 IDs → 2 Requests
assert len(spiele) == 25
assert [len(batch) for batch in aufgerufene_id_listen] == [20, 5]