#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2023 by the Nominatim developer community.
+# Copyright (C) 2024 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the Python web frameworks adaptor, v1 API.
"""
-from collections import namedtuple
import json
import xml.etree.ElementTree as ET
from pathlib import Path
import pytest
-from nominatim.config import Configuration
-import nominatim.api.v1.server_glue as glue
-import nominatim.api as napi
-import nominatim.api.logging as loglib
+from fake_adaptor import FakeAdaptor, FakeError, FakeResponse
-class FakeError(BaseException):
-
- def __init__(self, msg, status):
- self.msg = msg
- self.status = status
-
- def __str__(self):
- return f'{self.status} -- {self.msg}'
-
-FakeResponse = namedtuple('FakeResponse', ['status', 'output', 'content_type'])
-
-class FakeAdaptor(glue.ASGIAdaptor):
-
- def __init__(self, params={}, headers={}, config=None):
- self.params = params
- self.headers = headers
- self._config = config or Configuration(None)
-
-
- def get(self, name, default=None):
- return self.params.get(name, default)
-
-
- def get_header(self, name, default=None):
- return self.headers.get(name, default)
-
-
- def error(self, msg, status=400):
- return FakeError(msg, status)
-
-
- def create_response(self, status, output):
- return FakeResponse(status, output, self.content_type)
-
-
- def config(self):
- return self._config
+import nominatim_api.v1.server_glue as glue
+import nominatim_api as napi
+import nominatim_api.logging as loglib
# ASGIAdaptor.get_int/bool()
def test_adaptor_parse_format_use_default():
adaptor = FakeAdaptor()
- assert adaptor.parse_format(napi.StatusResult, 'text') == 'text'
+ assert glue.parse_format(adaptor, napi.StatusResult, 'text') == 'text'
assert adaptor.content_type == 'text/plain; charset=utf-8'
def test_adaptor_parse_format_use_configured():
adaptor = FakeAdaptor(params={'format': 'json'})
- assert adaptor.parse_format(napi.StatusResult, 'text') == 'json'
- assert adaptor.content_type == 'application/json'
+ assert glue.parse_format(adaptor, napi.StatusResult, 'text') == 'json'
+ assert adaptor.content_type == 'application/json; charset=utf-8'
def test_adaptor_parse_format_invalid_value():
adaptor = FakeAdaptor(params={'format': '@!#'})
with pytest.raises(FakeError, match='^400 -- .*must be one of'):
- adaptor.parse_format(napi.StatusResult, 'text')
+ glue.parse_format(adaptor, napi.StatusResult, 'text')
# ASGIAdaptor.get_accepted_languages()
def test_accepted_languages_from_param():
a = FakeAdaptor(params={'accept-language': 'de'})
- assert a.get_accepted_languages() == 'de'
+ assert glue.get_accepted_languages(a) == 'de'
def test_accepted_languages_from_header():
- a = FakeAdaptor(headers={'http_accept_language': 'de'})
- assert a.get_accepted_languages() == 'de'
+ a = FakeAdaptor(headers={'accept-language': 'de'})
+ assert glue.get_accepted_languages(a) == 'de'
def test_accepted_languages_from_default(monkeypatch):
monkeypatch.setenv('NOMINATIM_DEFAULT_LANGUAGE', 'de')
a = FakeAdaptor()
- assert a.get_accepted_languages() == 'de'
+ assert glue.get_accepted_languages(a) == 'de'
def test_accepted_languages_param_over_header():
a = FakeAdaptor(params={'accept-language': 'de'},
- headers={'http_accept_language': 'en'})
- assert a.get_accepted_languages() == 'de'
+ headers={'accept-language': 'en'})
+ assert glue.get_accepted_languages(a) == 'de'
def test_accepted_languages_header_over_default(monkeypatch):
monkeypatch.setenv('NOMINATIM_DEFAULT_LANGUAGE', 'en')
- a = FakeAdaptor(headers={'http_accept_language': 'de'})
- assert a.get_accepted_languages() == 'de'
+ a = FakeAdaptor(headers={'accept-language': 'de'})
+ assert glue.get_accepted_languages(a) == 'de'
# ASGIAdaptor.raise_error()
@pytest.fixture(autouse=True)
def init_adaptor(self):
self.adaptor = FakeAdaptor()
- self.adaptor.setup_debugging()
+ glue.setup_debugging(self.adaptor)
def run_raise_error(self, msg, status):
with pytest.raises(FakeError) as excinfo:
err = self.run_raise_error('TEST', 404)
assert self.adaptor.content_type == 'text/plain; charset=utf-8'
- assert err.msg == 'TEST'
+ assert err.msg == 'ERROR 404: TEST'
assert err.status == 404
def test_json(self):
- self.adaptor.content_type = 'application/json'
+ self.adaptor.content_type = 'application/json; charset=utf-8'
err = self.run_raise_error('TEST', 501)
def test_raise_error_during_debug():
a = FakeAdaptor(params={'debug': '1'})
- a.setup_debugging()
+ glue.setup_debugging(a)
loglib.log().section('Ongoing')
with pytest.raises(FakeError) as excinfo:
- a.raise_error('bad state')
+ a.raise_error('badstate')
content = ET.fromstring(excinfo.value.msg)
assert content.tag == 'html'
assert '>Ongoing<' in excinfo.value.msg
- assert 'bad state' in excinfo.value.msg
+ assert 'badstate' in excinfo.value.msg
# ASGIAdaptor.build_response
def test_build_response_without_content_type():
- resp = FakeAdaptor().build_response('attention')
+ resp = glue.build_response(FakeAdaptor(), 'attention')
assert isinstance(resp, FakeResponse)
assert resp.status == 200
def test_build_response_with_status():
a = FakeAdaptor(params={'format': 'json'})
- a.parse_format(napi.StatusResult, 'text')
+ glue.parse_format(a, napi.StatusResult, 'text')
- resp = a.build_response('stuff\nmore stuff', status=404)
+ resp = glue.build_response(a, 'stuff\nmore stuff', status=404)
assert isinstance(resp, FakeResponse)
assert resp.status == 404
assert resp.output == 'stuff\nmore stuff'
- assert resp.content_type == 'application/json'
+ assert resp.content_type == 'application/json; charset=utf-8'
def test_build_response_jsonp_with_json():
a = FakeAdaptor(params={'format': 'json', 'json_callback': 'test.func'})
- a.parse_format(napi.StatusResult, 'text')
+ glue.parse_format(a, napi.StatusResult, 'text')
- resp = a.build_response('{}')
+ resp = glue.build_response(a, '{}')
assert isinstance(resp, FakeResponse)
assert resp.status == 200
assert resp.output == 'test.func({})'
- assert resp.content_type == 'application/javascript'
+ assert resp.content_type == 'application/javascript; charset=utf-8'
def test_build_response_jsonp_without_json():
a = FakeAdaptor(params={'format': 'text', 'json_callback': 'test.func'})
- a.parse_format(napi.StatusResult, 'text')
+ glue.parse_format(a, napi.StatusResult, 'text')
- resp = a.build_response('{}')
+ resp = glue.build_response(a, '{}')
assert isinstance(resp, FakeResponse)
assert resp.status == 200
@pytest.mark.parametrize('param', ['alert(); func', '\\n', '', 'a b'])
def test_build_response_jsonp_bad_format(param):
a = FakeAdaptor(params={'format': 'json', 'json_callback': param})
- a.parse_format(napi.StatusResult, 'text')
+ glue.parse_format(a, napi.StatusResult, 'text')
with pytest.raises(FakeError, match='^400 -- .*Invalid'):
- a.build_response('{}')
+ glue.build_response(a, '{}')
# status_endpoint()
assert isinstance(resp, FakeResponse)
assert resp.status == 200
- assert resp.content_type == 'application/json'
+ assert resp.content_type == 'application/json; charset=utf-8'
@pytest.mark.asyncio
self.lookup_args.extend(args[1:])
return self.result
- monkeypatch.setattr(napi.NominatimAPIAsync, 'lookup', _lookup)
+ monkeypatch.setattr(napi.NominatimAPIAsync, 'details', _lookup)
@pytest.mark.asyncio
with pytest.raises(FakeError, match='^404 -- .*found'):
await glue.details_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+
+# reverse_endpoint()
+class TestReverseEndPoint:
+
+ @pytest.fixture(autouse=True)
+ def patch_reverse_func(self, monkeypatch):
+ self.result = napi.ReverseResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0))
+ async def _reverse(*args, **kwargs):
+ return self.result
+
+ monkeypatch.setattr(napi.NominatimAPIAsync, 'reverse', _reverse)
+
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize('params', [{}, {'lat': '3.4'}, {'lon': '6.7'}])
+ async def test_reverse_no_params(self, params):
+ a = FakeAdaptor()
+ a.params = params
+ a.params['format'] = 'xml'
+
+ with pytest.raises(FakeError, match='^400 -- (?s:.*)missing'):
+ await glue.reverse_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize('params', [{'lat': '45.6', 'lon': '4563'}])
+ async def test_reverse_success(self, params):
+ a = FakeAdaptor()
+ a.params = params
+ a.params['format'] = 'json'
+
+ res = await glue.reverse_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert res == ''
+
+
+ @pytest.mark.asyncio
+ async def test_reverse_success(self):
+ a = FakeAdaptor()
+ a.params['lat'] = '56.3'
+ a.params['lon'] = '6.8'
+
+ assert await glue.reverse_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+
+ @pytest.mark.asyncio
+ async def test_reverse_from_search(self):
+ a = FakeAdaptor()
+ a.params['q'] = '34.6 2.56'
+ a.params['format'] = 'json'
+
+ res = await glue.search_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert len(json.loads(res.output)) == 1
+
+
+# lookup_endpoint()
+
+class TestLookupEndpoint:
+
+ @pytest.fixture(autouse=True)
+ def patch_lookup_func(self, monkeypatch):
+ self.results = [napi.SearchResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0))]
+ async def _lookup(*args, **kwargs):
+ return napi.SearchResults(self.results)
+
+ monkeypatch.setattr(napi.NominatimAPIAsync, 'lookup', _lookup)
+
+
+ @pytest.mark.asyncio
+ async def test_lookup_no_params(self):
+ a = FakeAdaptor()
+ a.params['format'] = 'json'
+
+ res = await glue.lookup_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert res.output == '[]'
+
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize('param', ['w', 'bad', ''])
+ async def test_lookup_bad_params(self, param):
+ a = FakeAdaptor()
+ a.params['format'] = 'json'
+ a.params['osm_ids'] = f'W34,{param},N33333'
+
+ res = await glue.lookup_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert len(json.loads(res.output)) == 1
+
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize('param', ['p234234', '4563'])
+ async def test_lookup_bad_osm_type(self, param):
+ a = FakeAdaptor()
+ a.params['format'] = 'json'
+ a.params['osm_ids'] = f'W34,{param},N33333'
+
+ res = await glue.lookup_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert len(json.loads(res.output)) == 1
+
+
+ @pytest.mark.asyncio
+ async def test_lookup_working(self):
+ a = FakeAdaptor()
+ a.params['format'] = 'json'
+ a.params['osm_ids'] = 'N23,W34'
+
+ res = await glue.lookup_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert len(json.loads(res.output)) == 1
+
+
+# search_endpoint()
+
+class TestSearchEndPointSearch:
+
+ @pytest.fixture(autouse=True)
+ def patch_lookup_func(self, monkeypatch):
+ self.results = [napi.SearchResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0))]
+ async def _search(*args, **kwargs):
+ return napi.SearchResults(self.results)
+
+ monkeypatch.setattr(napi.NominatimAPIAsync, 'search', _search)
+
+
+ @pytest.mark.asyncio
+ async def test_search_free_text(self):
+ a = FakeAdaptor()
+ a.params['q'] = 'something'
+
+ res = await glue.search_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert len(json.loads(res.output)) == 1
+
+
+ @pytest.mark.asyncio
+ async def test_search_free_text_xml(self):
+ a = FakeAdaptor()
+ a.params['q'] = 'something'
+ a.params['format'] = 'xml'
+
+ res = await glue.search_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert res.status == 200
+ assert res.output.index('something') > 0
+
+
+ @pytest.mark.asyncio
+ async def test_search_free_and_structured(self):
+ a = FakeAdaptor()
+ a.params['q'] = 'something'
+ a.params['city'] = 'ignored'
+
+ with pytest.raises(FakeError, match='^400 -- .*cannot be used together'):
+ res = await glue.search_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+
+ @pytest.mark.asyncio
+ @pytest.mark.parametrize('dedupe,numres', [(True, 1), (False, 2)])
+ async def test_search_dedupe(self, dedupe, numres):
+ self.results = self.results * 2
+ a = FakeAdaptor()
+ a.params['q'] = 'something'
+ if not dedupe:
+ a.params['dedupe'] = '0'
+
+ res = await glue.search_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert len(json.loads(res.output)) == numres
+
+
+class TestSearchEndPointSearchAddress:
+
+ @pytest.fixture(autouse=True)
+ def patch_lookup_func(self, monkeypatch):
+ self.results = [napi.SearchResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0))]
+ async def _search(*args, **kwargs):
+ return napi.SearchResults(self.results)
+
+ monkeypatch.setattr(napi.NominatimAPIAsync, 'search_address', _search)
+
+
+ @pytest.mark.asyncio
+ async def test_search_structured(self):
+ a = FakeAdaptor()
+ a.params['street'] = 'something'
+
+ res = await glue.search_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert len(json.loads(res.output)) == 1
+
+
+class TestSearchEndPointSearchCategory:
+
+ @pytest.fixture(autouse=True)
+ def patch_lookup_func(self, monkeypatch):
+ self.results = [napi.SearchResult(napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0))]
+ async def _search(*args, **kwargs):
+ return napi.SearchResults(self.results)
+
+ monkeypatch.setattr(napi.NominatimAPIAsync, 'search_category', _search)
+
+
+ @pytest.mark.asyncio
+ async def test_search_category(self):
+ a = FakeAdaptor()
+ a.params['q'] = '[shop=fog]'
+
+ res = await glue.search_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a)
+
+ assert len(json.loads(res.output)) == 1
projects / nominatim.git / blobdiff