--- /dev/null
+# 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
+
--- /dev/null
+# 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'}]
+
--- /dev/null
+# 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]
"""
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=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()
projects / nominatim.git / commitdiff