X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/6e81596609518d9202a00bd457af0a75648d30d9..0278ab7f41bf9bc96a87084c04bfec263c6acc40:/test/python/api/test_server_glue_v1.py diff --git a/test/python/api/test_server_glue_v1.py b/test/python/api/test_server_glue_v1.py index 612bac28..5a7430f4 100644 --- a/test/python/api/test_server_glue_v1.py +++ b/test/python/api/test_server_glue_v1.py @@ -7,56 +7,18 @@ """ 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 +from fake_adaptor import FakeAdaptor, FakeError, FakeResponse + import nominatim.api.v1.server_glue as glue import nominatim.api as napi import nominatim.api.logging as loglib -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 - # ASGIAdaptor.get_int/bool() @@ -105,7 +67,7 @@ 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 adaptor.content_type == 'application/json; charset=utf-8' def test_adaptor_parse_format_invalid_value(): @@ -123,7 +85,7 @@ def test_accepted_languages_from_param(): def test_accepted_languages_from_header(): - a = FakeAdaptor(headers={'http_accept_language': 'de'}) + a = FakeAdaptor(headers={'accept-language': 'de'}) assert a.get_accepted_languages() == 'de' @@ -135,13 +97,13 @@ def test_accepted_languages_from_default(monkeypatch): def test_accepted_languages_param_over_header(): a = FakeAdaptor(params={'accept-language': 'de'}, - headers={'http_accept_language': 'en'}) + headers={'accept-language': 'en'}) assert a.get_accepted_languages() == 'de' def test_accepted_languages_header_over_default(monkeypatch): monkeypatch.setenv('NOMINATIM_DEFAULT_LANGUAGE', 'en') - a = FakeAdaptor(headers={'http_accept_language': 'de'}) + a = FakeAdaptor(headers={'accept-language': 'de'}) assert a.get_accepted_languages() == 'de' @@ -170,7 +132,7 @@ class TestAdaptorRaiseError: 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) @@ -197,14 +159,14 @@ def test_raise_error_during_debug(): 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 @@ -227,7 +189,7 @@ def test_build_response_with_status(): 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(): @@ -239,7 +201,7 @@ def test_build_response_jsonp_with_json(): 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(): @@ -308,7 +270,7 @@ class TestStatusEndpoint: 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 @@ -384,3 +346,227 @@ class TestDetailsEndpoint: 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