spiele-redaktion/tests/test_neuheiten_bgg.py

"""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 = """<?xml version="1.0" encoding="utf-8"?>
<items total="3" termsofuse="https://boardgamegeek.com/xmlapi/termsofuse">
  <item type="boardgame" id="13">
    <name value="Catan"/>
    <yearpublished value="1995"/>
  </item>
  <item type="boardgameexpansion" id="55">
    <name value="Catan: Erweiterung"/>
    <yearpublished value="1996"/>
  </item>
  <item type="boardgame" id="9999">
    <name value="Mysterium"/>
  </item>
</items>
"""

THING_XML = """<?xml version="1.0" encoding="utf-8"?>
<items termsofuse="https://boardgamegeek.com/xmlapi/termsofuse">
  <item type="boardgame" id="13">
    <thumbnail>https://example.org/thumb.jpg</thumbnail>
    <name type="primary" sortindex="1" value="Catan"/>
    <yearpublished value="1995"/>
    <link type="boardgamepublisher" id="4" value="KOSMOS"/>
    <link type="boardgamepublisher" id="27" value="Catan Studio"/>
    <link type="boardgamedesigner" id="2" value="Klaus Teuber"/>
  </item>
  <item type="boardgameexpansion" id="55">
    <name type="primary" value="Catan: Erweiterung"/>
    <yearpublished value="1996"/>
  </item>
</items>
"""


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 = '<items><item type="boardgame" id="1"><name value="Schätze von Trullala"/></item></items>'
    treffer = parse_suche(xml)
    assert treffer[0].titel == "Schätze von Trullala"

    with pytest.raises(BggFehler):
        parse_suche("<items><item></items>")


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('<items total="0"/>') == []
    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'<item type="boardgame" id="{i}"><name value="Spiel {i}"/></item>'
            for i in ids
        )
        return _antwort(f"<items>{items}</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]