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

import asyncio
from typing import Any

import httpx
import pytest

from app.adapters.address_suggestions.dadata.client import (
    DadataAddressSuggestionProvider,
    DadataClientError,
    DadataRequestError,
)
from app.config import DadataAddressSuggestionsConfig
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 post(
        self,
        url: str,
        *,
        json: dict[str, Any] | None = None,
        headers: dict[str, str] | None = None,
        timeout: float | None = None,
    ) -> httpx.Response:
        self.calls.append(
            {
                "method": "POST",
                "url": url,
                "json": json,
                "headers": headers,
                "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": "RU",
        "city": "Москва",
        "query": "Хабаровская",
        "limit": 5,
    }
    payload.update(overrides)
    return AddressSuggestRequest(**payload)


def test_dadata_provider_maps_response_to_unified_model() -> None:
    response = httpx.Response(
        200,
        json={
            "suggestions": [
                {
                    "value": "г Москва, ул Хабаровская",
                    "unrestricted_value": "107241, г Москва, ул Хабаровская",
                    "data": {
                        "street": "Хабаровская",
                        "house": "1",
                        "flat": "25",
                        "postal_code": "107241",
                        "country_iso_code": "RU",
                        "geo_lat": "55.821168",
                    },
                }
            ]
        },
        request=httpx.Request(
            "POST",
            "https://suggestions.dadata.test/suggestions/api/4_1/rs/suggest/address",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = DadataAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://suggestions.dadata.test/suggestions/api/4_1/rs/suggest/address",
        api_key="test-api-key",
        timeout_seconds=7.5,
    )

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

    assert result == [
        AddressSuggestion(
            address="107241, г Москва, ул Хабаровская",
            street="Хабаровская",
            house="1",
            flat="25",
            postal_code="107241",
        )
    ]
    assert http_client.calls == [
        {
            "method": "POST",
            "url": "https://suggestions.dadata.test/suggestions/api/4_1/rs/suggest/address",
            "json": {
                "query": "Москва Хабаровская",
                "count": 5,
            },
            "headers": {
                "Authorization": "Token test-api-key",
                "Accept": "application/json",
                "Content-Type": "application/json",
            },
            "timeout": 7.5,
        }
    ]


def test_dadata_provider_uses_foreign_country_payload_and_config_factory() -> None:
    response = httpx.Response(
        200,
        json={"suggestions": []},
        request=httpx.Request(
            "POST",
            "https://suggestions.dadata.test/suggestions/api/4_1/rs/suggest/address",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = DadataAddressSuggestionProvider.from_config(
        http_client=http_client,  # type: ignore[arg-type]
        config=DadataAddressSuggestionsConfig(
            url="https://suggestions.dadata.test/suggestions/api/4_1/rs/suggest/address",
            api_key="config-api-key",
            timeout_seconds=3.0,
        ),
    )

    result = asyncio.run(
        provider.suggest(
            _make_request(
                country_code="BY",
                city="Гомель",
                query="Советская",
                limit=3,
            )
        )
    )

    assert result == []
    assert http_client.calls[0]["json"] == {
        "query": "Гомель Советская",
        "count": 3,
        "locations": [{"country": "*"}],
    }
    assert http_client.calls[0]["headers"]["Authorization"] == "Token config-api-key"
    assert http_client.calls[0]["timeout"] == 3.0


def test_dadata_provider_maps_4xx_to_request_error() -> None:
    response = httpx.Response(
        403,
        json={"detail": "forbidden"},
        request=httpx.Request(
            "POST",
            "https://suggestions.dadata.test/suggestions/api/4_1/rs/suggest/address",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = DadataAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://suggestions.dadata.test/suggestions/api/4_1/rs/suggest/address",
        api_key="test-api-key",
    )

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


def test_dadata_provider_maps_5xx_to_client_error() -> None:
    response = httpx.Response(
        503,
        json={"detail": "unavailable"},
        request=httpx.Request(
            "POST",
            "https://suggestions.dadata.test/suggestions/api/4_1/rs/suggest/address",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = DadataAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://suggestions.dadata.test/suggestions/api/4_1/rs/suggest/address",
        api_key="test-api-key",
    )

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


def test_dadata_provider_maps_transport_error_to_client_error() -> None:
    transport_error = httpx.ReadTimeout(
        "read timeout",
        request=httpx.Request(
            "POST",
            "https://suggestions.dadata.test/suggestions/api/4_1/rs/suggest/address",
        ),
    )
    http_client = SequenceHTTPClient([transport_error])
    provider = DadataAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://suggestions.dadata.test/suggestions/api/4_1/rs/suggest/address",
        api_key="test-api-key",
    )

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


def test_dadata_provider_raises_on_invalid_payload_shape() -> None:
    response = httpx.Response(
        200,
        json={"suggestions": [{"data": {"postal_code": "107241"}}]},
        request=httpx.Request(
            "POST",
            "https://suggestions.dadata.test/suggestions/api/4_1/rs/suggest/address",
        ),
    )
    http_client = SequenceHTTPClient([response])
    provider = DadataAddressSuggestionProvider(
        http_client=http_client,  # type: ignore[arg-type]
        url="https://suggestions.dadata.test/suggestions/api/4_1/rs/suggest/address",
        api_key="test-api-key",
    )

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