forked from hans/Nominatim
add tests for new endpoints
This commit is contained in:
52
test/python/api/fake_adaptor.py
Normal file
52
test/python/api/fake_adaptor.py
Normal file
@@ -0,0 +1,52 @@
|
|||||||
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||||
|
#
|
||||||
|
# This file is part of Nominatim. (https://nominatim.org)
|
||||||
|
#
|
||||||
|
# Copyright (C) 2023 by the Nominatim developer community.
|
||||||
|
# For a full list of authors see the git log.
|
||||||
|
"""
|
||||||
|
Provides dummy implementations of ASGIAdaptor for testing.
|
||||||
|
"""
|
||||||
|
from collections import namedtuple
|
||||||
|
|
||||||
|
import nominatim.api.v1.server_glue as glue
|
||||||
|
from nominatim.config import Configuration
|
||||||
|
|
||||||
|
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
|
||||||
|
|
||||||
67
test/python/api/test_api_deletable_v1.py
Normal file
67
test/python/api/test_api_deletable_v1.py
Normal file
@@ -0,0 +1,67 @@
|
|||||||
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||||
|
#
|
||||||
|
# This file is part of Nominatim. (https://nominatim.org)
|
||||||
|
#
|
||||||
|
# Copyright (C) 2023 by the Nominatim developer community.
|
||||||
|
# For a full list of authors see the git log.
|
||||||
|
"""
|
||||||
|
Tests for the deletable v1 API call.
|
||||||
|
"""
|
||||||
|
import json
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
import pytest_asyncio
|
||||||
|
|
||||||
|
import psycopg2.extras
|
||||||
|
|
||||||
|
from fake_adaptor import FakeAdaptor, FakeError, FakeResponse
|
||||||
|
|
||||||
|
import nominatim.api.v1.server_glue as glue
|
||||||
|
import nominatim.api as napi
|
||||||
|
|
||||||
|
@pytest_asyncio.fixture
|
||||||
|
async def api():
|
||||||
|
api = napi.NominatimAPIAsync(Path('/invalid'))
|
||||||
|
yield api
|
||||||
|
await api.close()
|
||||||
|
|
||||||
|
|
||||||
|
class TestDeletableEndPoint:
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def setup_deletable_table(self, temp_db_cursor, table_factory, temp_db_with_extensions):
|
||||||
|
psycopg2.extras.register_hstore(temp_db_cursor)
|
||||||
|
table_factory('import_polygon_delete',
|
||||||
|
definition='osm_id bigint, osm_type char(1), class text, type text',
|
||||||
|
content=[(345, 'N', 'boundary', 'administrative'),
|
||||||
|
(781, 'R', 'landuse', 'wood'),
|
||||||
|
(781, 'R', 'landcover', 'grass')])
|
||||||
|
table_factory('placex',
|
||||||
|
definition="""place_id bigint, osm_id bigint, osm_type char(1),
|
||||||
|
class text, type text, name HSTORE, country_code char(2)""",
|
||||||
|
content=[(1, 345, 'N', 'boundary', 'administrative', {'old_name': 'Former'}, 'ab'),
|
||||||
|
(2, 781, 'R', 'landuse', 'wood', {'name': 'Wood'}, 'cd'),
|
||||||
|
(3, 781, 'R', 'landcover', 'grass', None, 'cd')])
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_deletable(self, api):
|
||||||
|
a = FakeAdaptor()
|
||||||
|
|
||||||
|
resp = await glue.deletable_endpoint(api, a)
|
||||||
|
results = json.loads(resp.output)
|
||||||
|
|
||||||
|
results.sort(key=lambda r: r['place_id'])
|
||||||
|
|
||||||
|
assert results == [{'place_id': 1, 'country_code': 'ab', 'name': None,
|
||||||
|
'osm_id': 345, 'osm_type': 'N',
|
||||||
|
'class': 'boundary', 'type': 'administrative'},
|
||||||
|
{'place_id': 2, 'country_code': 'cd', 'name': 'Wood',
|
||||||
|
'osm_id': 781, 'osm_type': 'R',
|
||||||
|
'class': 'landuse', 'type': 'wood'},
|
||||||
|
{'place_id': 3, 'country_code': 'cd', 'name': None,
|
||||||
|
'osm_id': 781, 'osm_type': 'R',
|
||||||
|
'class': 'landcover', 'type': 'grass'}]
|
||||||
|
|
||||||
111
test/python/api/test_api_polygons_v1.py
Normal file
111
test/python/api/test_api_polygons_v1.py
Normal file
@@ -0,0 +1,111 @@
|
|||||||
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
||||||
|
#
|
||||||
|
# This file is part of Nominatim. (https://nominatim.org)
|
||||||
|
#
|
||||||
|
# Copyright (C) 2023 by the Nominatim developer community.
|
||||||
|
# For a full list of authors see the git log.
|
||||||
|
"""
|
||||||
|
Tests for the deletable v1 API call.
|
||||||
|
"""
|
||||||
|
import json
|
||||||
|
import datetime as dt
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
import pytest_asyncio
|
||||||
|
|
||||||
|
import psycopg2.extras
|
||||||
|
|
||||||
|
from fake_adaptor import FakeAdaptor, FakeError, FakeResponse
|
||||||
|
|
||||||
|
import nominatim.api.v1.server_glue as glue
|
||||||
|
import nominatim.api as napi
|
||||||
|
|
||||||
|
@pytest_asyncio.fixture
|
||||||
|
async def api():
|
||||||
|
api = napi.NominatimAPIAsync(Path('/invalid'))
|
||||||
|
yield api
|
||||||
|
await api.close()
|
||||||
|
|
||||||
|
|
||||||
|
class TestPolygonsEndPoint:
|
||||||
|
|
||||||
|
@pytest.fixture(autouse=True)
|
||||||
|
def setup_deletable_table(self, temp_db_cursor, table_factory, temp_db_with_extensions):
|
||||||
|
psycopg2.extras.register_hstore(temp_db_cursor)
|
||||||
|
|
||||||
|
self.now = dt.datetime.now()
|
||||||
|
self.recent = dt.datetime.now() - dt.timedelta(days=3)
|
||||||
|
|
||||||
|
table_factory('import_polygon_error',
|
||||||
|
definition="""osm_id bigint,
|
||||||
|
osm_type character(1),
|
||||||
|
class text,
|
||||||
|
type text,
|
||||||
|
name hstore,
|
||||||
|
country_code character varying(2),
|
||||||
|
updated timestamp without time zone,
|
||||||
|
errormessage text,
|
||||||
|
prevgeometry geometry(Geometry,4326),
|
||||||
|
newgeometry geometry(Geometry,4326)""",
|
||||||
|
content=[(345, 'N', 'boundary', 'administrative',
|
||||||
|
{'name': 'Foo'}, 'xx', self.recent,
|
||||||
|
'some text', None, None),
|
||||||
|
(781, 'R', 'landuse', 'wood',
|
||||||
|
None, 'ds', self.now,
|
||||||
|
'Area reduced by lots', None, None)])
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_polygons_simple(self, api):
|
||||||
|
a = FakeAdaptor()
|
||||||
|
|
||||||
|
resp = await glue.polygons_endpoint(api, a)
|
||||||
|
results = json.loads(resp.output)
|
||||||
|
|
||||||
|
results.sort(key=lambda r: (r['osm_type'], r['osm_id']))
|
||||||
|
|
||||||
|
assert results == [{'osm_type': 'N', 'osm_id': 345,
|
||||||
|
'class': 'boundary', 'type': 'administrative',
|
||||||
|
'name': 'Foo', 'country_code': 'xx',
|
||||||
|
'errormessage': 'some text',
|
||||||
|
'updated': self.recent.isoformat(sep=' ', timespec='seconds')},
|
||||||
|
{'osm_type': 'R', 'osm_id': 781,
|
||||||
|
'class': 'landuse', 'type': 'wood',
|
||||||
|
'name': None, 'country_code': 'ds',
|
||||||
|
'errormessage': 'Area reduced by lots',
|
||||||
|
'updated': self.now.isoformat(sep=' ', timespec='seconds')}]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_polygons_days(self, api):
|
||||||
|
a = FakeAdaptor()
|
||||||
|
a.params['days'] = '2'
|
||||||
|
|
||||||
|
resp = await glue.polygons_endpoint(api, a)
|
||||||
|
results = json.loads(resp.output)
|
||||||
|
|
||||||
|
assert [r['osm_id'] for r in results] == [781]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_polygons_class(self, api):
|
||||||
|
a = FakeAdaptor()
|
||||||
|
a.params['class'] = 'landuse'
|
||||||
|
|
||||||
|
resp = await glue.polygons_endpoint(api, a)
|
||||||
|
results = json.loads(resp.output)
|
||||||
|
|
||||||
|
assert [r['osm_id'] for r in results] == [781]
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_polygons_reduced(self, api):
|
||||||
|
a = FakeAdaptor()
|
||||||
|
a.params['reduced'] = '1'
|
||||||
|
|
||||||
|
resp = await glue.polygons_endpoint(api, a)
|
||||||
|
results = json.loads(resp.output)
|
||||||
|
|
||||||
|
assert [r['osm_id'] for r in results] == [781]
|
||||||
@@ -7,56 +7,18 @@
|
|||||||
"""
|
"""
|
||||||
Tests for the Python web frameworks adaptor, v1 API.
|
Tests for the Python web frameworks adaptor, v1 API.
|
||||||
"""
|
"""
|
||||||
from collections import namedtuple
|
|
||||||
import json
|
import json
|
||||||
import xml.etree.ElementTree as ET
|
import xml.etree.ElementTree as ET
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
import pytest
|
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.v1.server_glue as glue
|
||||||
import nominatim.api as napi
|
import nominatim.api as napi
|
||||||
import nominatim.api.logging as loglib
|
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=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
|
|
||||||
|
|
||||||
|
|
||||||
# ASGIAdaptor.get_int/bool()
|
# ASGIAdaptor.get_int/bool()
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user