Добавлен yandex geosuggest
This commit is contained in:
Раис Юсупалиев
@@ -0,0 +1,266 @@
|
||||
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()))
|
||||
Reference in New Issue
Block a user