]> git.openstreetmap.org Git - nominatim.git/blob - test/python/api/test_api_polygons_v1.py
port code to psycopg3
[nominatim.git] / 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 from fake_adaptor import FakeAdaptor, FakeError, FakeResponse
18
19 import nominatim_api.v1.server_glue as glue
20 import nominatim_api as napi
21
22 @pytest_asyncio.fixture
23 async def api():
24     api = napi.NominatimAPIAsync(Path('/invalid'))
25     yield api
26     await api.close()
27
28
29 class TestPolygonsEndPoint:
30
31     @pytest.fixture(autouse=True)
32     def setup_deletable_table(self, temp_db_cursor, table_factory, temp_db_with_extensions):
33         self.now = dt.datetime.now()
34         self.recent = dt.datetime.now() - dt.timedelta(days=3)
35
36         table_factory('import_polygon_error',
37                       definition="""osm_id bigint,
38                                     osm_type character(1),
39                                     class text,
40                                     type text,
41                                     name hstore,
42                                     country_code character varying(2),
43                                     updated timestamp without time zone,
44                                     errormessage text,
45                                     prevgeometry geometry(Geometry,4326),
46                                     newgeometry geometry(Geometry,4326)""",
47                     content=[(345, 'N', 'boundary', 'administrative',
48                               {'name': 'Foo'}, 'xx', self.recent,
49                               'some text', None, None),
50                              (781, 'R', 'landuse', 'wood',
51                               None, 'ds', self.now,
52                               'Area reduced by lots', None, None)])
53
54
55     @pytest.mark.asyncio
56     async def test_polygons_simple(self, api):
57         a = FakeAdaptor()
58
59         resp = await glue.polygons_endpoint(api, a)
60         results = json.loads(resp.output)
61
62         results.sort(key=lambda r: (r['osm_type'], r['osm_id']))
63
64         assert results == [{'osm_type': 'N', 'osm_id': 345,
65                             'class': 'boundary', 'type': 'administrative',
66                             'name': 'Foo', 'country_code': 'xx',
67                             'errormessage': 'some text',
68                             'updated': self.recent.isoformat(sep=' ', timespec='seconds')},
69                            {'osm_type': 'R', 'osm_id': 781,
70                             'class': 'landuse', 'type': 'wood',
71                             'name': None, 'country_code': 'ds',
72                             'errormessage': 'Area reduced by lots',
73                             'updated': self.now.isoformat(sep=' ', timespec='seconds')}]
74
75
76     @pytest.mark.asyncio
77     async def test_polygons_days(self, api):
78         a = FakeAdaptor()
79         a.params['days'] = '2'
80
81         resp = await glue.polygons_endpoint(api, a)
82         results = json.loads(resp.output)
83
84         assert [r['osm_id'] for r in results] == [781]
85
86
87     @pytest.mark.asyncio
88     async def test_polygons_class(self, api):
89         a = FakeAdaptor()
90         a.params['class'] = 'landuse'
91
92         resp = await glue.polygons_endpoint(api, a)
93         results = json.loads(resp.output)
94
95         assert [r['osm_id'] for r in results] == [781]
96
97
98
99     @pytest.mark.asyncio
100     async def test_polygons_reduced(self, api):
101         a = FakeAdaptor()
102         a.params['reduced'] = '1'
103
104         resp = await glue.polygons_endpoint(api, a)
105         results = json.loads(resp.output)
106
107         assert [r['osm_id'] for r in results] == [781]