g2s-aggregator/tests/adapters/address_suggestions/tomtom/test_client.py

import asyncio
from typing import Any

import httpx
import pytest

from app.adapters.address_suggestions.tomtom.client import (
    TomTomAddressSuggestionProvider,
    TomTomClientError,
    TomTomRequestError,
)
from app.config import TomTomAddressSuggestionsConfig
from app.schemas.request import AddressSuggestRequest
from app.schemas.response import AddressSuggestion


class SequenceHTTPClient:
    def __init__(self, results: list[Any]) -> None:
        self._results = results
        self.calls: list[dict[str, Any]] = []

    def _next_result(self) -> Any:
        return self._results[len(self.calls) - 1]

    async def get(
        self,
        url: str,
        *,
        params: dict[str, Any] | None = None,
        timeout: float | None = None,
    ) -> httpx.Response:
        self.calls.append(
            {
                "method": "GET",
                "url": url,
                "params": params,
                "timeout": timeout,
            }
        )
        result = self._next_result()
        if isinstance(result, Exception):
            raise result
        return result


def _make_request(**overrides: object) -> AddressSuggestRequest:
    payload: dict[str, object] = {
        "country_code": "DE",
        "city": "Berlin",
        "query": "Alexanderplatz 1",
        "limit": 5,
    }
    payload.update(overrides)
    return AddressSuggestRequest(**payload)


def test_tomtom_provider_maps_response_to_unified_model() -> None:
    response = httpx.Response(
        200,
        json={
            "results": [
                {
                    "address": {
                        "freeformAddress": "Alexanderplatz 1, 10178 Berlin",
                        "streetName": "Alexanderplatz",
                        "streetNumber": "1",
                        "postalCode": "10178",
                    }
                }
            ]
        },
        request=httpx.Request(
            "GET",
            "https://api.tomtom.test/search/2/search/Berlin%20Alexanderplatz%201.json",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = TomTomAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://api.tomtom.test/search/2/search",
        api_key="test-api-key",
        timeout_seconds=5.5,
    )

    result = asyncio.run(provider.suggest(_make_request()))

    assert result == [
        AddressSuggestion(
            address="Alexanderplatz 1, 10178 Berlin",
            street="Alexanderplatz",
            house="1",
            flat=None,
            postal_code="10178",
        )
    ]
    assert http_client.calls == [
        {
            "method": "GET",
            "url": "https://api.tomtom.test/search/2/search/Berlin%20Alexanderplatz%201.json",
            "params": {
                "key": "test-api-key",
                "countrySet": "DE",
                "typeahead": "true",
                "idxSet": "PAD,Addr,Str,EPP",
                "limit": 5,
            },
            "timeout": 5.5,
        }
    ]


def test_tomtom_provider_uses_config_factory_and_omits_limit_when_missing() -> None:
    response = httpx.Response(
        200,
        json={
            "results": [
                {
                    "address": {
                        "freeformAddress": "Dam Square, 1012 JS Amsterdam",
                        "streetName": "Dam Square",
                        "postalCode": "1012 JS",
                    }
                }
            ]
        },
        request=httpx.Request(
            "GET",
            "https://api.tomtom.test/search/2/search/Amsterdam%20Dam%20Square.json",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = TomTomAddressSuggestionProvider.from_config(
        http_client=http_client,  # type: ignore[arg-type]
        config=TomTomAddressSuggestionsConfig(
            url="https://api.tomtom.test/search/2/search",
            api_key="config-api-key",
            timeout_seconds=4.25,
        ),
    )

    result = asyncio.run(
        provider.suggest(
            _make_request(
                country_code="NL",
                city="Amsterdam",
                query="Dam Square",
                limit=None,
            )
        )
    )

    assert result == [
        AddressSuggestion(
            address="Dam Square, 1012 JS Amsterdam",
            street="Dam Square",
            house=None,
            flat=None,
            postal_code="1012 JS",
        )
    ]
    assert http_client.calls == [
        {
            "method": "GET",
            "url": "https://api.tomtom.test/search/2/search/Amsterdam%20Dam%20Square.json",
            "params": {
                "key": "config-api-key",
                "countrySet": "NL",
                "typeahead": "true",
                "idxSet": "PAD,Addr,Str,EPP",
            },
            "timeout": 4.25,
        }
    ]


def test_tomtom_provider_maps_400_to_request_error() -> None:
    response = httpx.Response(
        400,
        json={"errorText": "bad request"},
        request=httpx.Request(
            "GET",
            "https://api.tomtom.test/search/2/search/Berlin%20Alexanderplatz%201.json",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = TomTomAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://api.tomtom.test/search/2/search",
        api_key="test-api-key",
    )

    with pytest.raises(TomTomRequestError, match="status 400"):
        asyncio.run(provider.suggest(_make_request()))


@pytest.mark.parametrize("status_code", [403, 429, 503])
def test_tomtom_provider_maps_unavailable_statuses_to_client_error(
    status_code: int,
) -> None:
    response = httpx.Response(
        status_code,
        json={"errorText": "provider error"},
        request=httpx.Request(
            "GET",
            "https://api.tomtom.test/search/2/search/Berlin%20Alexanderplatz%201.json",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = TomTomAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://api.tomtom.test/search/2/search",
        api_key="test-api-key",
    )

    with pytest.raises(TomTomClientError, match=f"status {status_code}"):
        asyncio.run(provider.suggest(_make_request()))


def test_tomtom_provider_maps_transport_error_to_client_error() -> None:
    transport_error = httpx.ReadTimeout(
        "read timeout",
        request=httpx.Request(
            "GET",
            "https://api.tomtom.test/search/2/search/Berlin%20Alexanderplatz%201.json",
        ),
    )
    http_client = SequenceHTTPClient([transport_error])
    provider = TomTomAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://api.tomtom.test/search/2/search",
        api_key="test-api-key",
    )

    with pytest.raises(TomTomClientError, match="request failed"):
        asyncio.run(provider.suggest(_make_request()))


def test_tomtom_provider_raises_on_invalid_payload_shape() -> None:
    response = httpx.Response(
        200,
        json={"results": [{"address": {"freeformAddress": 1}}]},
        request=httpx.Request(
            "GET",
            "https://api.tomtom.test/search/2/search/Berlin%20Alexanderplatz%201.json",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = TomTomAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://api.tomtom.test/search/2/search",
        api_key="test-api-key",
    )

    with pytest.raises(
        TomTomClientError,
        match="response payload is invalid",
    ):
        asyncio.run(provider.suggest(_make_request()))