test/python/api/test_api_polygons_v1.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2024 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Tests for the deletable v1 API call.
   9 """
  10 import json
  11 import datetime as dt
  12 from pathlib import Path
  13
  14 import pytest
  15 import pytest_asyncio
  16
  17 import psycopg2.extras
  18
  19 from fake_adaptor import FakeAdaptor, FakeError, FakeResponse
  20
  21 import nominatim_api.v1.server_glue as glue
  22 import nominatim_api as napi
  23
  24 @pytest_asyncio.fixture
  25 async def api():
  26     api = napi.NominatimAPIAsync(Path('/invalid'))
  27     yield api
  28     await api.close()
  29
  30
  31 class TestPolygonsEndPoint:
  32
  33     @pytest.fixture(autouse=True)
  34     def setup_deletable_table(self, temp_db_cursor, table_factory, temp_db_with_extensions):
  35         psycopg2.extras.register_hstore(temp_db_cursor)
  36
  37         self.now = dt.datetime.now()
  38         self.recent = dt.datetime.now() - dt.timedelta(days=3)
  39
  40         table_factory('import_polygon_error',
  41                       definition="""osm_id bigint,
  42                                     osm_type character(1),
  43                                     class text,
  44                                     type text,
  45                                     name hstore,
  46                                     country_code character varying(2),
  47                                     updated timestamp without time zone,
  48                                     errormessage text,
  49                                     prevgeometry geometry(Geometry,4326),
  50                                     newgeometry geometry(Geometry,4326)""",
  51                     content=[(345, 'N', 'boundary', 'administrative',
  52                               {'name': 'Foo'}, 'xx', self.recent,
  53                               'some text', None, None),
  54                              (781, 'R', 'landuse', 'wood',
  55                               None, 'ds', self.now,
  56                               'Area reduced by lots', None, None)])
  57
  58
  59     @pytest.mark.asyncio
  60     async def test_polygons_simple(self, api):
  61         a = FakeAdaptor()
  62
  63         resp = await glue.polygons_endpoint(api, a)
  64         results = json.loads(resp.output)
  65
  66         results.sort(key=lambda r: (r['osm_type'], r['osm_id']))
  67
  68         assert results == [{'osm_type': 'N', 'osm_id': 345,
  69                             'class': 'boundary', 'type': 'administrative',
  70                             'name': 'Foo', 'country_code': 'xx',
  71                             'errormessage': 'some text',
  72                             'updated': self.recent.isoformat(sep=' ', timespec='seconds')},
  73                            {'osm_type': 'R', 'osm_id': 781,
  74                             'class': 'landuse', 'type': 'wood',
  75                             'name': None, 'country_code': 'ds',
  76                             'errormessage': 'Area reduced by lots',
  77                             'updated': self.now.isoformat(sep=' ', timespec='seconds')}]
  78
  79
  80     @pytest.mark.asyncio
  81     async def test_polygons_days(self, api):
  82         a = FakeAdaptor()
  83         a.params['days'] = '2'
  84
  85         resp = await glue.polygons_endpoint(api, a)
  86         results = json.loads(resp.output)
  87
  88         assert [r['osm_id'] for r in results] == [781]
  89
  90
  91     @pytest.mark.asyncio
  92     async def test_polygons_class(self, api):
  93         a = FakeAdaptor()
  94         a.params['class'] = 'landuse'
  95
  96         resp = await glue.polygons_endpoint(api, a)
  97         results = json.loads(resp.output)
  98
  99         assert [r['osm_id'] for r in results] == [781]
 100
 101
 102
 103     @pytest.mark.asyncio
 104     async def test_polygons_reduced(self, api):
 105         a = FakeAdaptor()
 106         a.params['reduced'] = '1'
 107
 108         resp = await glue.polygons_endpoint(api, a)
 109         results = json.loads(resp.output)
 110
 111         assert [r['osm_id'] for r in results] == [781]