141 lines
4.3 KiB
Python
141 lines
4.3 KiB
Python
import asyncio
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
|
|
def _load_geocoding_module():
|
|
module_path = Path(__file__).resolve().parents[2] / "app" / "services" / "geocoding.py"
|
|
spec = importlib.util.spec_from_file_location("tests_geocoding_under_test", module_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec and spec.loader
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
geocoding = _load_geocoding_module()
|
|
|
|
|
|
class FakeClient:
|
|
def __init__(self, response):
|
|
self.response = response
|
|
self.calls = []
|
|
|
|
async def get(self, url, *, params):
|
|
self.calls.append({"url": url, "params": params})
|
|
return self.response
|
|
|
|
|
|
def test_geocode_tianditu_gets_expected_params(monkeypatch):
|
|
monkeypatch.setattr(geocoding.settings, "TIANDITU_GEOCODER_TOKEN", "tk-test")
|
|
monkeypatch.setattr(
|
|
geocoding.settings,
|
|
"TIANDITU_GEOCODER_URL",
|
|
"https://api.tianditu.gov.cn/geocoder",
|
|
)
|
|
response = httpx.Response(
|
|
200,
|
|
json={
|
|
"location": {"lon": "116.407526", "lat": "39.904030", "level": "地名地址"},
|
|
"status": "0",
|
|
"msg": "ok",
|
|
},
|
|
request=httpx.Request("GET", "https://api.tianditu.gov.cn/geocoder"),
|
|
)
|
|
client = FakeClient(response)
|
|
|
|
result = asyncio.run(
|
|
geocoding.geocode_tianditu(
|
|
geocoding.TiandituGeocodeRequest(keyword="北京市人民政府"),
|
|
client=client,
|
|
)
|
|
)
|
|
|
|
assert result["location"] == {
|
|
"lon": "116.407526",
|
|
"lat": "39.904030",
|
|
"level": "地名地址",
|
|
}
|
|
assert client.calls == [
|
|
{
|
|
"url": "https://api.tianditu.gov.cn/geocoder",
|
|
"params": {
|
|
"ds": json.dumps({"keyWord": "北京市人民政府"}, ensure_ascii=False),
|
|
"tk": "tk-test",
|
|
},
|
|
}
|
|
]
|
|
|
|
|
|
def test_geocode_tianditu_accepts_key_word_alias(monkeypatch):
|
|
monkeypatch.setattr(geocoding.settings, "TIANDITU_GEOCODER_TOKEN", "tk-test")
|
|
response = httpx.Response(
|
|
200,
|
|
json={"location": {"lon": "116", "lat": "39"}, "status": "0", "msg": "ok"},
|
|
request=httpx.Request("GET", "https://api.tianditu.gov.cn/geocoder"),
|
|
)
|
|
|
|
result = asyncio.run(
|
|
geocoding.geocode_tianditu(
|
|
geocoding.TiandituGeocodeRequest(keyWord="北京市人民政府"),
|
|
client=FakeClient(response),
|
|
)
|
|
)
|
|
|
|
assert result["status"] == "0"
|
|
|
|
|
|
def test_geocode_tianditu_requires_token(monkeypatch):
|
|
monkeypatch.setattr(geocoding.settings, "TIANDITU_GEOCODER_TOKEN", "")
|
|
|
|
with pytest.raises(geocoding.TiandituGeocodingConfigError):
|
|
asyncio.run(
|
|
geocoding.geocode_tianditu(
|
|
geocoding.TiandituGeocodeRequest(keyword="北京市人民政府"),
|
|
client=FakeClient(httpx.Response(200, json={})),
|
|
)
|
|
)
|
|
|
|
|
|
def test_geocode_tianditu_surfaces_http_error(monkeypatch):
|
|
monkeypatch.setattr(geocoding.settings, "TIANDITU_GEOCODER_TOKEN", "tk-test")
|
|
response = httpx.Response(
|
|
403,
|
|
json={"msg": "invalid tk"},
|
|
request=httpx.Request("GET", "https://api.tianditu.gov.cn/geocoder"),
|
|
)
|
|
|
|
with pytest.raises(geocoding.TiandituGeocodingAPIError) as exc_info:
|
|
asyncio.run(
|
|
geocoding.geocode_tianditu(
|
|
geocoding.TiandituGeocodeRequest(keyword="北京市人民政府"),
|
|
client=FakeClient(response),
|
|
)
|
|
)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
assert exc_info.value.detail == {"msg": "invalid tk"}
|
|
|
|
|
|
def test_geocode_tianditu_surfaces_tianditu_error_status(monkeypatch):
|
|
monkeypatch.setattr(geocoding.settings, "TIANDITU_GEOCODER_TOKEN", "tk-test")
|
|
response = httpx.Response(
|
|
200,
|
|
json={"status": "100", "msg": "bad request"},
|
|
request=httpx.Request("GET", "https://api.tianditu.gov.cn/geocoder"),
|
|
)
|
|
|
|
with pytest.raises(geocoding.TiandituGeocodingAPIError) as exc_info:
|
|
asyncio.run(
|
|
geocoding.geocode_tianditu(
|
|
geocoding.TiandituGeocodeRequest(keyword="北京市人民政府"),
|
|
client=FakeClient(response),
|
|
)
|
|
)
|
|
|
|
assert exc_info.value.status_code == 502
|
|
assert exc_info.value.detail == {"status": "100", "msg": "bad request"}
|