173 lines
8.1 KiB
Python
173 lines
8.1 KiB
Python
import asyncio
|
|
import json
|
|
import logging
|
|
|
|
import httpx
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from app.core.config import Settings, settings
|
|
from app.domain.schemas.place_search import PlaceSearchRequest
|
|
from app.domain.schemas.project_map_config import PlaceSearchConfig
|
|
|
|
CONFIG = PlaceSearchConfig(enabled=True, admin_code="310120", region="奉贤区")
|
|
from app.services.place_search import PlaceSearchError, search_tianditu_places
|
|
|
|
|
|
def poi(**overrides):
|
|
return {
|
|
"hotPointID": "place-1", "name": "南桥路", "address": "上海市奉贤区南桥路",
|
|
"lonlat": "121.441891,31.220188", "county": "奉贤区", "countyCode": "156310120",
|
|
"provinceCode": "156310000", "typeName": "道路名", **overrides,
|
|
}
|
|
|
|
|
|
def payload(**overrides):
|
|
return {"status": {"infocode": 1000}, "resultType": 1, "count": 21, "pois": [poi()], **overrides}
|
|
|
|
|
|
class FakeClient:
|
|
def __init__(self, data=None, status=200, error=None):
|
|
self.data = data if data is not None else payload()
|
|
self.status = status
|
|
self.error = error
|
|
self.calls = []
|
|
|
|
async def get(self, url, *, params):
|
|
self.calls.append((url, params))
|
|
if self.error:
|
|
raise self.error
|
|
return httpx.Response(self.status, json=self.data, request=httpx.Request("GET", url))
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def shared_token(monkeypatch):
|
|
monkeypatch.setattr(settings, "TIANDITU_TOKEN", "shared-token")
|
|
|
|
|
|
def test_search_fixes_fengxian_scope_and_normalizes_candidates():
|
|
client = FakeClient()
|
|
result = asyncio.run(search_tianditu_places(PlaceSearchRequest(keyword=" 南桥路 "), config=CONFIG, client=client))
|
|
url, params = client.calls[0]
|
|
assert url == "https://api.tianditu.gov.cn/v2/search"
|
|
assert params["tk"] == "shared-token"
|
|
assert json.loads(params["postStr"]) == {
|
|
"keyWord": "南桥路", "queryType": 12, "specify": "156310120", "show": 2, "start": 0, "count": 10,
|
|
}
|
|
assert result.items[0].name == "南桥路"
|
|
assert result.items[0].longitude == 121.441891
|
|
assert result.items[0].district == "奉贤区"
|
|
assert result.items[0].category == "道路名"
|
|
assert result.truncated and result.total == 1 and result.region == "奉贤区"
|
|
|
|
|
|
@pytest.mark.parametrize("data", [payload(count=0, pois=[]), {"status": {"infocode": 3001}}])
|
|
def test_no_matches_is_a_successful_empty_result(data):
|
|
result = asyncio.run(search_tianditu_places(PlaceSearchRequest(keyword="未知地名"), config=CONFIG, client=FakeClient(data)))
|
|
assert result.items == [] and result.total == 0 and not result.truncated
|
|
|
|
|
|
@pytest.mark.parametrize("matched_total, size, truncated", [(0, 0, False), (3, 3, False), (10, 10, False), (28, 10, True), (2000, 10, True)])
|
|
def test_search_only_returns_first_ten_results(matched_total, size, truncated):
|
|
client = FakeClient(payload(count=matched_total, pois=[poi(hotPointID=f"place-{index}") for index in range(size)]))
|
|
result = asyncio.run(search_tianditu_places(PlaceSearchRequest(keyword="南桥路"), config=CONFIG, client=client))
|
|
assert len(result.items) == result.total == size and result.truncated == truncated
|
|
params = json.loads(client.calls[0][1]["postStr"])
|
|
assert params["count"] == 10 and params["start"] == 0 and params["specify"] == "156310120"
|
|
assert "has_more" not in result.model_dump() and "offset" not in result.model_dump()
|
|
|
|
|
|
@pytest.mark.parametrize("data", [
|
|
payload(pois=[poi(lonlat="bad")]),
|
|
payload(pois=[poi(lonlat="181,31")]),
|
|
payload(pois=[poi(lonlat="121,NaN")]),
|
|
payload(pois=[poi(name=None)]),
|
|
payload(pois=[poi(provinceCode="156320000")]),
|
|
payload(pois=[poi(countyCode="156310104", county="徐汇区")]),
|
|
payload(pois=[poi(countyCode="156310120", county="静安区")]),
|
|
payload(pois=[poi(countyCode=None, county=None)]),
|
|
payload(count=-1),
|
|
payload(pois=[poi() for _ in range(11)]),
|
|
payload(resultType=4),
|
|
{"status": {"infocode": 2001}},
|
|
])
|
|
def test_invalid_or_unlocatable_provider_response_is_rejected(data):
|
|
with pytest.raises(PlaceSearchError) as exc:
|
|
asyncio.run(search_tianditu_places(PlaceSearchRequest(keyword="南桥路"), config=CONFIG, client=FakeClient(data)))
|
|
assert exc.value.status_code == 502
|
|
|
|
|
|
def test_missing_shared_token_is_configuration_error(monkeypatch):
|
|
monkeypatch.setattr(settings, "TIANDITU_TOKEN", "")
|
|
client = FakeClient()
|
|
with pytest.raises(PlaceSearchError) as exc:
|
|
asyncio.run(search_tianditu_places(PlaceSearchRequest(keyword="南桥路"), config=CONFIG, client=client))
|
|
assert exc.value.status_code == 503 and client.calls == []
|
|
|
|
|
|
def test_http_request_logs_redact_the_shared_token(caplog):
|
|
caplog.set_level(logging.INFO, logger="httpx")
|
|
|
|
async def run():
|
|
transport = httpx.MockTransport(lambda _request: httpx.Response(200, json=payload()))
|
|
async with httpx.AsyncClient(transport=transport) as client:
|
|
await search_tianditu_places(PlaceSearchRequest(keyword="南桥路"), config=CONFIG, client=client)
|
|
|
|
asyncio.run(run())
|
|
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("shared-token" not in message for message in messages)
|
|
|
|
|
|
@pytest.mark.parametrize("failure, status", [
|
|
(httpx.ReadTimeout("secret-token-in-url"), 504),
|
|
(httpx.ConnectError("secret-token-in-url"), 502),
|
|
])
|
|
def test_transport_errors_are_safe_and_actionable(failure, status):
|
|
with pytest.raises(PlaceSearchError) as exc:
|
|
asyncio.run(search_tianditu_places(PlaceSearchRequest(keyword="南桥路"), config=CONFIG, client=FakeClient(error=failure)))
|
|
assert exc.value.status_code == status
|
|
assert "secret-token" not in exc.value.detail
|
|
|
|
|
|
@pytest.mark.parametrize("body", [{"keyword": " "}, {"keyword": "南桥路", "specify": "北京"}, {"keyword": "南桥路", "offset": 0}, {"keyword": "南桥路", "limit": 10}])
|
|
def test_request_rejects_blank_keywords_scope_overrides_and_paging_options(body):
|
|
with pytest.raises(ValidationError):
|
|
PlaceSearchRequest(**body)
|
|
|
|
|
|
@pytest.mark.parametrize("key", ["TIANDITU_TOKEN", "TJWATER_TIANDITU_TOKEN", "TIANDITU_GEOCODER_TOKEN"])
|
|
def test_shared_token_accepts_existing_configuration_names(monkeypatch, key):
|
|
for name in ("TIANDITU_TOKEN", "TJWATER_TIANDITU_TOKEN", "TIANDITU_GEOCODER_TOKEN"):
|
|
monkeypatch.delenv(name, raising=False)
|
|
assert Settings(_env_file=None, **{key: "one-token"}).TIANDITU_TOKEN == "one-token"
|
|
|
|
|
|
@pytest.mark.parametrize('admin_code,region,county_code,county,province', [
|
|
('341126', '凤阳县', '156341126', '凤阳县', '156340000'),
|
|
('320506', '吴中区', '320506', '吴中区', '320000'),
|
|
('440800', '湛江市', '156440803', '霞山区', '156440000'),
|
|
('310000', '上海市', '156310118', '青浦区', '156310000'),
|
|
])
|
|
def test_search_uses_project_scope_for_districts_and_cities(admin_code, region, county_code, county, province):
|
|
config = PlaceSearchConfig(enabled=True, admin_code=admin_code, region=region)
|
|
client = FakeClient(payload(pois=[poi(countyCode=county_code, county=county, provinceCode=province)]))
|
|
result = asyncio.run(search_tianditu_places(PlaceSearchRequest(keyword='地名'), config=config, client=client))
|
|
assert json.loads(client.calls[0][1]['postStr'])['specify'] == '156' + admin_code
|
|
assert result.region == region and result.items[0].district == county
|
|
|
|
|
|
def test_configured_city_rejects_other_city_in_same_province():
|
|
config = PlaceSearchConfig(enabled=True, admin_code='440800', region='湛江市')
|
|
with pytest.raises(PlaceSearchError, match='无效'):
|
|
asyncio.run(search_tianditu_places(PlaceSearchRequest(keyword='地名'), config=config,
|
|
client=FakeClient(payload(pois=[poi(countyCode='156440104', county='越秀区', provinceCode='156440000')]))))
|
|
|
|
|
|
def test_disabled_project_does_not_call_provider():
|
|
client = FakeClient()
|
|
with pytest.raises(PlaceSearchError) as exc:
|
|
asyncio.run(search_tianditu_places(PlaceSearchRequest(keyword='地名'),
|
|
config=CONFIG.model_copy(update={'enabled': False}), client=client))
|
|
assert exc.value.status_code == 503 and client.calls == []
|