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

import asyncio
from typing import Any

import httpx
import pytest

from app.adapters.address_suggestions.yandex_geosuggest.client import (
    YandexGeosuggestAddressSuggestionProvider,
    YandexGeosuggestClientError,
    YandexGeosuggestRequestError,
)
from app.config import YandexGeosuggestAddressSuggestionsConfig
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": "AM",
        "city": "Yerevan",
        "query": "Tumanyan 12",
        "limit": 5,
    }
    payload.update(overrides)
    return AddressSuggestRequest(**payload)


def test_yandex_geosuggest_provider_maps_response_to_unified_model() -> None:
    response = httpx.Response(
        200,
        json={
            "results": [
                {
                    "title": {"text": "Tumanyan Street, 12, Yerevan"},
                    "address": {
                        "formatted_address": "Tumanyan Street, 12, apt 34, Yerevan",
                        "component": [
                            {"name": "Tumanyan Street", "kind": ["STREET"]},
                            {"name": "12", "kind": ["HOUSE"]},
                            {"name": "34", "kind": ["APARTMENT"]},
                            {"name": "0001", "kind": ["POSTAL_CODE"]},
                        ],
                    },
                }
            ]
        },
        request=httpx.Request(
            "GET",
            "https://suggest-maps.yandex.test/v1/suggest",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = YandexGeosuggestAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://suggest-maps.yandex.test/v1/suggest",
        api_key="test-api-key",
        timeout_seconds=6.5,
    )

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

    assert result == [
        AddressSuggestion(
            address="Tumanyan Street, 12, apt 34, Yerevan",
            street="Tumanyan Street",
            house="12",
            flat="34",
            postal_code=None,
        )
    ]
    assert http_client.calls == [
        {
            "method": "GET",
            "url": "https://suggest-maps.yandex.test/v1/suggest",
            "params": {
                "apikey": "test-api-key",
                "text": "Yerevan Tumanyan 12",
                "print_address": 1,
                "results": 5,
            },
            "timeout": 6.5,
        }
    ]


def test_yandex_geosuggest_provider_uses_config_factory_and_omits_results_without_limit() -> None:
    response = httpx.Response(
        200,
        json={
            "results": [
                {
                    "title": {"text": "Republic Square, Yerevan"},
                    "address": {
                        "formatted_address": "Republic Square, Yerevan",
                        "component": [
                            {"name": "Republic Square", "kind": ["street"]},
                        ],
                    },
                }
            ]
        },
        request=httpx.Request(
            "GET",
            "https://suggest-maps.yandex.test/v1/suggest",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = YandexGeosuggestAddressSuggestionProvider.from_config(
        http_client=http_client,  # type: ignore[arg-type]
        config=YandexGeosuggestAddressSuggestionsConfig(
            url="https://suggest-maps.yandex.test/v1/suggest",
            api_key="config-api-key",
            timeout_seconds=4.5,
        ),
    )

    result = asyncio.run(
        provider.suggest(
            _make_request(
                query="Republic Square",
                limit=None,
            )
        )
    )

    assert result == [
        AddressSuggestion(
            address="Republic Square, Yerevan",
            street="Republic Square",
        )
    ]
    assert http_client.calls == [
        {
            "method": "GET",
            "url": "https://suggest-maps.yandex.test/v1/suggest",
            "params": {
                "apikey": "config-api-key",
                "text": "Yerevan Republic Square",
                "print_address": 1,
            },
            "timeout": 4.5,
        }
    ]


def test_yandex_geosuggest_provider_maps_400_to_request_error() -> None:
    response = httpx.Response(
        400,
        json={"error": "bad request"},
        request=httpx.Request(
            "GET",
            "https://suggest-maps.yandex.test/v1/suggest",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = YandexGeosuggestAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://suggest-maps.yandex.test/v1/suggest",
        api_key="test-api-key",
    )

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


@pytest.mark.parametrize("status_code", [403, 429, 503])
def test_yandex_geosuggest_provider_maps_unavailable_statuses_to_client_error(
    status_code: int,
) -> None:
    response = httpx.Response(
        status_code,
        json={"error": "provider error"},
        request=httpx.Request(
            "GET",
            "https://suggest-maps.yandex.test/v1/suggest",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = YandexGeosuggestAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://suggest-maps.yandex.test/v1/suggest",
        api_key="test-api-key",
    )

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


def test_yandex_geosuggest_provider_maps_transport_error_to_client_error() -> None:
    transport_error = httpx.ReadTimeout(
        "read timeout",
        request=httpx.Request(
            "GET",
            "https://suggest-maps.yandex.test/v1/suggest",
        ),
    )
    http_client = SequenceHTTPClient([transport_error])
    provider = YandexGeosuggestAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://suggest-maps.yandex.test/v1/suggest",
        api_key="test-api-key",
    )

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


def test_yandex_geosuggest_provider_raises_on_invalid_payload_shape() -> None:
    response = httpx.Response(
        200,
        json={
            "results": [
                {
                    "address": {
                        "formatted_address": "Tumanyan Street, 12, Yerevan",
                        "component": [{"name": 12, "kind": ["HOUSE"]}],
                    }
                }
            ]
        },
        request=httpx.Request(
            "GET",
            "https://suggest-maps.yandex.test/v1/suggest",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = YandexGeosuggestAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://suggest-maps.yandex.test/v1/suggest",
        api_key="test-api-key",
    )

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