X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/7c66fef63f5066674e0e76d378c8ace2754ee54c..c70dfccacac90a90bfc193f49262abe7505d7d75:/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 a731e720..8d9f0940 100644 --- a/test/python/api/test_server_glue_v1.py +++ b/test/python/api/test_server_glue_v1.py @@ -2,60 +2,21 @@ # # This file is part of Nominatim. (https://nominatim.org) # -# Copyright (C) 2023 by the Nominatim developer community. +# Copyright (C) 2025 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=None, headers=None, config=None): - self.params = params or {} - self.headers = headers or {} - 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() @@ -97,52 +58,52 @@ def test_adaptor_get_bool_falsish(): 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={'accept-language': 'de'}) - assert a.get_accepted_languages() == '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={'accept-language': 'en'}) - assert a.get_accepted_languages() == 'de' + assert glue.get_accepted_languages(a) == 'de' def test_accepted_languages_header_over_default(monkeypatch): monkeypatch.setenv('NOMINATIM_DEFAULT_LANGUAGE', 'en') a = FakeAdaptor(headers={'accept-language': 'de'}) - assert a.get_accepted_languages() == 'de' + assert glue.get_accepted_languages(a) == 'de' # ASGIAdaptor.raise_error() @@ -152,7 +113,7 @@ class TestAdaptorRaiseError: @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: @@ -160,17 +121,15 @@ class TestAdaptorRaiseError: return excinfo.value - def test_without_content_set(self): 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) @@ -178,7 +137,6 @@ class TestAdaptorRaiseError: assert content['code'] == 501 assert content['message'] == 'TEST' - def test_xml(self): self.adaptor.content_type = 'text/xml; charset=utf-8' @@ -193,7 +151,7 @@ class TestAdaptorRaiseError: 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: @@ -210,7 +168,7 @@ def test_raise_error_during_debug(): # 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 @@ -220,33 +178,33 @@ def test_build_response_without_content_type(): 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 @@ -257,10 +215,10 @@ def test_build_response_jsonp_without_json(): @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() @@ -274,42 +232,38 @@ class TestStatusEndpoint: monkeypatch.setattr(napi.NominatimAPIAsync, 'status', _status) - @pytest.mark.asyncio async def test_status_without_params(self): a = FakeAdaptor() self.status = napi.StatusResult(0, 'foo') - resp = await glue.status_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) + resp = await glue.status_endpoint(napi.NominatimAPIAsync(), a) assert isinstance(resp, FakeResponse) assert resp.status == 200 assert resp.content_type == 'text/plain; charset=utf-8' - @pytest.mark.asyncio async def test_status_with_error(self): a = FakeAdaptor() self.status = napi.StatusResult(405, 'foo') - resp = await glue.status_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) + resp = await glue.status_endpoint(napi.NominatimAPIAsync(), a) assert isinstance(resp, FakeResponse) assert resp.status == 500 assert resp.content_type == 'text/plain; charset=utf-8' - @pytest.mark.asyncio async def test_status_json_with_error(self): a = FakeAdaptor(params={'format': 'json'}) self.status = napi.StatusResult(405, 'foo') - resp = await glue.status_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) + resp = await glue.status_endpoint(napi.NominatimAPIAsync(), a) 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 async def test_status_bad_format(self): @@ -317,7 +271,7 @@ class TestStatusEndpoint: self.status = napi.StatusResult(0, 'foo') with pytest.raises(FakeError): - await glue.status_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) + await glue.status_endpoint(napi.NominatimAPIAsync(), a) # details_endpoint() @@ -337,53 +291,48 @@ class TestDetailsEndpoint: monkeypatch.setattr(napi.NominatimAPIAsync, 'details', _lookup) - @pytest.mark.asyncio async def test_details_no_params(self): a = FakeAdaptor() with pytest.raises(FakeError, match='^400 -- .*Missing'): - await glue.details_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) - + await glue.details_endpoint(napi.NominatimAPIAsync(), a) @pytest.mark.asyncio async def test_details_by_place_id(self): a = FakeAdaptor(params={'place_id': '4573'}) - await glue.details_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) + await glue.details_endpoint(napi.NominatimAPIAsync(), a) assert self.lookup_args[0].place_id == 4573 - @pytest.mark.asyncio async def test_details_by_osm_id(self): a = FakeAdaptor(params={'osmtype': 'N', 'osmid': '45'}) - await glue.details_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) + await glue.details_endpoint(napi.NominatimAPIAsync(), a) assert self.lookup_args[0].osm_type == 'N' assert self.lookup_args[0].osm_id == 45 assert self.lookup_args[0].osm_class is None - @pytest.mark.asyncio async def test_details_with_debugging(self): a = FakeAdaptor(params={'osmtype': 'N', 'osmid': '45', 'debug': '1'}) - resp = await glue.details_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) + resp = await glue.details_endpoint(napi.NominatimAPIAsync(), a) content = ET.fromstring(resp.output) assert resp.content_type == 'text/html; charset=utf-8' assert content.tag == 'html' - @pytest.mark.asyncio async def test_details_no_result(self): a = FakeAdaptor(params={'place_id': '4573'}) self.result = None with pytest.raises(FakeError, match='^404 -- .*found'): - await glue.details_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) + await glue.details_endpoint(napi.NominatimAPIAsync(), a) # reverse_endpoint() @@ -392,14 +341,14 @@ 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)) + ('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): @@ -408,20 +357,7 @@ class TestReverseEndPoint: 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 == '' - + await glue.reverse_endpoint(napi.NominatimAPIAsync(), a) @pytest.mark.asyncio async def test_reverse_success(self): @@ -429,8 +365,7 @@ class TestReverseEndPoint: a.params['lat'] = '56.3' a.params['lon'] = '6.8' - assert await glue.reverse_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) - + assert await glue.reverse_endpoint(napi.NominatimAPIAsync(), a) @pytest.mark.asyncio async def test_reverse_from_search(self): @@ -438,7 +373,7 @@ class TestReverseEndPoint: a.params['q'] = '34.6 2.56' a.params['format'] = 'json' - res = await glue.search_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) + res = await glue.search_endpoint(napi.NominatimAPIAsync(), a) assert len(json.loads(res.output)) == 1 @@ -452,22 +387,21 @@ class TestLookupEndpoint: 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) + res = await glue.lookup_endpoint(napi.NominatimAPIAsync(), a) assert res.output == '[]' - @pytest.mark.asyncio @pytest.mark.parametrize('param', ['w', 'bad', '']) async def test_lookup_bad_params(self, param): @@ -475,11 +409,10 @@ class TestLookupEndpoint: a.params['format'] = 'json' a.params['osm_ids'] = f'W34,{param},N33333' - res = await glue.lookup_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) + res = await glue.lookup_endpoint(napi.NominatimAPIAsync(), 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): @@ -487,18 +420,17 @@ class TestLookupEndpoint: a.params['format'] = 'json' a.params['osm_ids'] = f'W34,{param},N33333' - res = await glue.lookup_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) + res = await glue.lookup_endpoint(napi.NominatimAPIAsync(), 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) + res = await glue.lookup_endpoint(napi.NominatimAPIAsync(), a) assert len(json.loads(res.output)) == 1 @@ -512,44 +444,40 @@ class TestSearchEndPointSearch: 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) + res = await glue.search_endpoint(napi.NominatimAPIAsync(), 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) + res = await glue.search_endpoint(napi.NominatimAPIAsync(), 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' - res = await glue.search_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) - - assert len(json.loads(res.output)) == 1 - + with pytest.raises(FakeError, match='^400 -- .*cannot be used together'): + await glue.search_endpoint(napi.NominatimAPIAsync(), a) @pytest.mark.asyncio @pytest.mark.parametrize('dedupe,numres', [(True, 1), (False, 2)]) @@ -560,7 +488,7 @@ class TestSearchEndPointSearch: if not dedupe: a.params['dedupe'] = '0' - res = await glue.search_endpoint(napi.NominatimAPIAsync(Path('/invalid')), a) + res = await glue.search_endpoint(napi.NominatimAPIAsync(), a) assert len(json.loads(res.output)) == numres @@ -572,18 +500,18 @@ class TestSearchEndPointSearchAddress: 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) + res = await glue.search_endpoint(napi.NominatimAPIAsync(), a) assert len(json.loads(res.output)) == 1 @@ -595,17 +523,17 @@ class TestSearchEndPointSearchCategory: 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) + res = await glue.search_endpoint(napi.NominatimAPIAsync(), a) assert len(json.loads(res.output)) == 1