"""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 BggClient, BggFehler, 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]] = [] def handle_request(self, request: httpx.Request) -> httpx.Response: self.anfragen.append((str(request.url), self.uhr())) 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_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_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]