Добавлена dadata
This commit is contained in:
Раис Юсупалиев
@@ -0,0 +1,231 @@
|
||||
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": {
|
||||
"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(
|
||||
provider="dadata",
|
||||
address="107241, г Москва, ул Хабаровская",
|
||||
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()))
|
||||
Reference in New Issue
Block a user