111 lines
4.7 KiB
Python
111 lines
4.7 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
|
|
import httpx
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from app.core.config import settings
|
|
from app.domain.schemas.place_search import ReverseGeocodingRequest
|
|
from app.domain.schemas.project_map_config import PlaceSearchConfig
|
|
from app.services.place_search import PlaceSearchError
|
|
from app.services.reverse_geocoding import reverse_geocode_tianditu
|
|
|
|
CONFIG = PlaceSearchConfig(enabled=True, admin_code='341126', region='凤阳县')
|
|
POINT = ReverseGeocodingRequest(longitude=117.570309, latitude=32.868029)
|
|
|
|
|
|
def provider_result(**component):
|
|
return {'status': '0', 'result': {
|
|
'formatted_address': '安徽省滁州市凤阳县府城镇朝阳路76号',
|
|
'location': {'lon': 117.57, 'lat': 32.86},
|
|
'addressComponent': {'poi': '府城镇', 'road': '朝阳路', 'county': '凤阳县',
|
|
'county_code': '156341126', 'province_code': '156340000', **component}}}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def token(monkeypatch):
|
|
monkeypatch.setattr(settings, 'TIANDITU_TOKEN', 'test-only-token')
|
|
|
|
|
|
def lookup(data=None, *, config=CONFIG, status=200, failure=None):
|
|
calls = []
|
|
|
|
def transport(request):
|
|
calls.append(request)
|
|
if failure:
|
|
raise failure
|
|
return httpx.Response(status, json=provider_result() if data is None else data)
|
|
|
|
async def run():
|
|
async with httpx.AsyncClient(transport=httpx.MockTransport(transport)) as client:
|
|
return await reverse_geocode_tianditu(POINT, config=config, client=client)
|
|
|
|
return asyncio.run(run()), calls
|
|
|
|
|
|
def test_reverse_lookup_keeps_clicked_coordinate_and_normalizes_the_address():
|
|
result, calls = lookup()
|
|
assert result.name == '府城镇' and result.address == '安徽省滁州市凤阳县府城镇朝阳路76号'
|
|
assert (result.longitude, result.latitude) == (POINT.longitude, POINT.latitude)
|
|
assert result.district == '凤阳县' and result.category == '附近地名'
|
|
params = calls[0].url.params
|
|
assert params['type'] == 'geocode' and params['tk'] == 'test-only-token'
|
|
assert json.loads(params['postStr']) == {'lon': POINT.longitude, 'lat': POINT.latitude, 'ver': 1}
|
|
|
|
|
|
def test_reverse_lookup_supports_city_scope_and_falls_back_to_real_road_name():
|
|
result, _ = lookup(provider_result(poi='', county='青浦区', county_code='156310118', province_code='156310000'),
|
|
config=PlaceSearchConfig(enabled=True, admin_code='310000', region='上海市'))
|
|
assert result.name == '朝阳路' and result.category == '地址' and result.district == '青浦区'
|
|
|
|
|
|
@pytest.mark.parametrize('data,status', [
|
|
({'status': '0', 'result': None}, 404),
|
|
({'status': '0', 'result': {'formatted_address': ''}}, 404),
|
|
({'status': '1'}, 502),
|
|
({'status': '0', 'result': 'bad'}, 502),
|
|
(provider_result(county='奉贤区', county_code='156310120', province_code='156310000'), 422),
|
|
(provider_result(county='定远县', county_code='156341125'), 422),
|
|
])
|
|
def test_reverse_lookup_distinguishes_no_data_bad_response_and_out_of_scope(data, status):
|
|
with pytest.raises(PlaceSearchError) as exc:
|
|
lookup(data)
|
|
assert exc.value.status_code == status
|
|
|
|
|
|
@pytest.mark.parametrize('failure,status', [(httpx.ReadTimeout('private-url'), 504), (httpx.ConnectError('private-url'), 502)])
|
|
def test_reverse_lookup_sanitizes_transport_failures(failure, status):
|
|
with pytest.raises(PlaceSearchError) as exc:
|
|
lookup(failure=failure)
|
|
assert exc.value.status_code == status and 'private-url' not in exc.value.detail
|
|
|
|
|
|
def test_reverse_lookup_redacts_provider_key_from_http_logs(caplog):
|
|
caplog.set_level(logging.INFO, logger='httpx')
|
|
lookup()
|
|
messages = [record.getMessage() for record in caplog.records if record.name == 'httpx']
|
|
assert messages and any('tk=[redacted]' in message for message in messages)
|
|
assert all('test-only-token' not in message for message in messages)
|
|
|
|
|
|
@pytest.mark.parametrize('body', [
|
|
{'longitude': 181, 'latitude': 32}, {'longitude': 117, 'latitude': -91},
|
|
{'longitude': float('nan'), 'latitude': 32}, {'longitude': 117, 'latitude': float('inf')},
|
|
{'longitude': 117, 'latitude': 32, 'region': '全国'},
|
|
])
|
|
def test_reverse_lookup_rejects_invalid_coordinates_and_scope_overrides(body):
|
|
with pytest.raises(ValidationError):
|
|
ReverseGeocodingRequest(**body)
|
|
|
|
|
|
def test_reverse_lookup_requires_configuration_and_shared_key(monkeypatch):
|
|
with pytest.raises(PlaceSearchError) as exc:
|
|
lookup(config=CONFIG.model_copy(update={'enabled': False}))
|
|
assert exc.value.status_code == 503
|
|
monkeypatch.setattr(settings, 'TIANDITU_TOKEN', '')
|
|
with pytest.raises(PlaceSearchError) as exc:
|
|
lookup()
|
|
assert exc.value.status_code == 503
|