]> git.openstreetmap.org Git - nominatim.git/blob - test/python/conftest.py
4b7cccc39f09992b73c48d348ebd150a4942d8b9
[nominatim.git] / test / python / conftest.py
1 import itertools
2 import sys
3 from pathlib import Path
4
5 import psycopg2
6 import psycopg2.extras
7 import pytest
8
9 SRC_DIR = Path(__file__) / '..' / '..' / '..'
10
11 # always test against the source
12 sys.path.insert(0, str(SRC_DIR.resolve()))
13
14 from nominatim.config import Configuration
15 from nominatim.db import connection
16 from nominatim.db.sql_preprocessor import SQLPreprocessor
17
18 class _TestingCursor(psycopg2.extras.DictCursor):
19     """ Extension to the DictCursor class that provides execution
20         short-cuts that simplify writing assertions.
21     """
22
23     def scalar(self, sql, params=None):
24         """ Execute a query with a single return value and return this value.
25             Raises an assertion when not exactly one row is returned.
26         """
27         self.execute(sql, params)
28         assert self.rowcount == 1
29         return self.fetchone()[0]
30
31     def row_set(self, sql, params=None):
32         """ Execute a query and return the result as a set of tuples.
33         """
34         self.execute(sql, params)
35         if self.rowcount == 1:
36             return set(tuple(self.fetchone()))
37
38         return set((tuple(row) for row in self))
39
40     def table_exists(self, table):
41         """ Check that a table with the given name exists in the database.
42         """
43         num = self.scalar("""SELECT count(*) FROM pg_tables
44                              WHERE tablename = %s""", (table, ))
45         return num == 1
46
47     def table_rows(self, table):
48         """ Return the number of rows in the given table.
49         """
50         return self.scalar('SELECT count(*) FROM ' + table)
51
52
53 @pytest.fixture
54 def temp_db(monkeypatch):
55     """ Create an empty database for the test. The database name is also
56         exported into NOMINATIM_DATABASE_DSN.
57     """
58     name = 'test_nominatim_python_unittest'
59     conn = psycopg2.connect(database='postgres')
60
61     conn.set_isolation_level(0)
62     with conn.cursor() as cur:
63         cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
64         cur.execute('CREATE DATABASE {}'.format(name))
65
66     conn.close()
67
68     monkeypatch.setenv('NOMINATIM_DATABASE_DSN' , 'dbname=' + name)
69
70     yield name
71
72     conn = psycopg2.connect(database='postgres')
73
74     conn.set_isolation_level(0)
75     with conn.cursor() as cur:
76         cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
77
78     conn.close()
79
80
81 @pytest.fixture
82 def dsn(temp_db):
83     return 'dbname=' + temp_db
84
85
86 @pytest.fixture
87 def temp_db_with_extensions(temp_db):
88     conn = psycopg2.connect(database=temp_db)
89     with conn.cursor() as cur:
90         cur.execute('CREATE EXTENSION hstore; CREATE EXTENSION postgis;')
91     conn.commit()
92     conn.close()
93
94     return temp_db
95
96 @pytest.fixture
97 def temp_db_conn(temp_db):
98     """ Connection to the test database.
99     """
100     with connection.connect('dbname=' + temp_db) as conn:
101         yield conn
102
103
104 @pytest.fixture
105 def temp_db_cursor(temp_db):
106     """ Connection and cursor towards the test database. The connection will
107         be in auto-commit mode.
108     """
109     conn = psycopg2.connect('dbname=' + temp_db)
110     conn.set_isolation_level(0)
111     with conn.cursor(cursor_factory=_TestingCursor) as cur:
112         yield cur
113     conn.close()
114
115
116 @pytest.fixture
117 def table_factory(temp_db_cursor):
118     def mk_table(name, definition='id INT', content=None):
119         temp_db_cursor.execute('CREATE TABLE {} ({})'.format(name, definition))
120         if content is not None:
121             if not isinstance(content, str):
122                 content = '),('.join([str(x) for x in content])
123             temp_db_cursor.execute("INSERT INTO {} VALUES ({})".format(name, content))
124
125     return mk_table
126
127
128 @pytest.fixture
129 def def_config():
130     return Configuration(None, SRC_DIR.resolve() / 'settings')
131
132 @pytest.fixture
133 def src_dir():
134     return SRC_DIR.resolve()
135
136 @pytest.fixture
137 def status_table(temp_db_conn):
138     """ Create an empty version of the status table and
139         the status logging table.
140     """
141     with temp_db_conn.cursor() as cur:
142         cur.execute("""CREATE TABLE import_status (
143                            lastimportdate timestamp with time zone NOT NULL,
144                            sequence_id integer,
145                            indexed boolean
146                        )""")
147         cur.execute("""CREATE TABLE import_osmosis_log (
148                            batchend timestamp,
149                            batchseq integer,
150                            batchsize bigint,
151                            starttime timestamp,
152                            endtime timestamp,
153                            event text
154                            )""")
155     temp_db_conn.commit()
156
157
158 @pytest.fixture
159 def place_table(temp_db_with_extensions, temp_db_conn):
160     """ Create an empty version of the place table.
161     """
162     with temp_db_conn.cursor() as cur:
163         cur.execute("""CREATE TABLE place (
164                            osm_id int8 NOT NULL,
165                            osm_type char(1) NOT NULL,
166                            class text NOT NULL,
167                            type text NOT NULL,
168                            name hstore,
169                            admin_level smallint,
170                            address hstore,
171                            extratags hstore,
172                            geometry Geometry(Geometry,4326) NOT NULL)""")
173     temp_db_conn.commit()
174
175
176 @pytest.fixture
177 def place_row(place_table, temp_db_cursor):
178     """ A factory for rows in the place table. The table is created as a
179         prerequisite to the fixture.
180     """
181     idseq = itertools.count(1001)
182     def _insert(osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
183                 admin_level=None, address=None, extratags=None, geom=None):
184         temp_db_cursor.execute("INSERT INTO place VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
185                                (osm_id or next(idseq), osm_type, cls, typ, names,
186                                 admin_level, address, extratags,
187                                 geom or 'SRID=4326;POINT(0 0)'))
188
189     return _insert
190
191 @pytest.fixture
192 def placex_table(temp_db_with_extensions, temp_db_conn):
193     """ Create an empty version of the place table.
194     """
195     with temp_db_conn.cursor() as cur:
196         cur.execute("""CREATE TABLE placex (
197                            place_id BIGINT,
198                            parent_place_id BIGINT,
199                            linked_place_id BIGINT,
200                            importance FLOAT,
201                            indexed_date TIMESTAMP,
202                            geometry_sector INTEGER,
203                            rank_address SMALLINT,
204                            rank_search SMALLINT,
205                            partition SMALLINT,
206                            indexed_status SMALLINT,
207                            osm_id int8,
208                            osm_type char(1),
209                            class text,
210                            type text,
211                            name hstore,
212                            admin_level smallint,
213                            address hstore,
214                            extratags hstore,
215                            geometry Geometry(Geometry,4326),
216                            wikipedia TEXT,
217                            country_code varchar(2),
218                            housenumber TEXT,
219                            postcode TEXT,
220                            centroid GEOMETRY(Geometry, 4326))""")
221     temp_db_conn.commit()
222
223
224 @pytest.fixture
225 def osmline_table(temp_db_with_extensions, temp_db_conn):
226     with temp_db_conn.cursor() as cur:
227         cur.execute("""CREATE TABLE location_property_osmline (
228                            place_id BIGINT,
229                            osm_id BIGINT,
230                            parent_place_id BIGINT,
231                            geometry_sector INTEGER,
232                            indexed_date TIMESTAMP,
233                            startnumber INTEGER,
234                            endnumber INTEGER,
235                            partition SMALLINT,
236                            indexed_status SMALLINT,
237                            linegeo GEOMETRY,
238                            interpolationtype TEXT,
239                            address HSTORE,
240                            postcode TEXT,
241                            country_code VARCHAR(2))""")
242     temp_db_conn.commit()
243
244
245 @pytest.fixture
246 def word_table(temp_db, temp_db_conn):
247     with temp_db_conn.cursor() as cur:
248         cur.execute("""CREATE TABLE word (
249                            word_id INTEGER,
250                            word_token text,
251                            word text,
252                            class text,
253                            type text,
254                            country_code varchar(2),
255                            search_name_count INTEGER,
256                            operator TEXT)""")
257     temp_db_conn.commit()
258
259
260 @pytest.fixture
261 def osm2pgsql_options(temp_db):
262     return dict(osm2pgsql='echo',
263                 osm2pgsql_cache=10,
264                 osm2pgsql_style='style.file',
265                 threads=1,
266                 dsn='dbname=' + temp_db,
267                 flatnode_file='',
268                 tablespaces=dict(slim_data='', slim_index='',
269                                  main_data='', main_index=''))
270
271 @pytest.fixture
272 def sql_preprocessor(temp_db_conn, tmp_path, def_config, monkeypatch, table_factory):
273     monkeypatch.setenv('NOMINATIM_DATABASE_MODULE_PATH', '.')
274     table_factory('country_name', 'partition INT', (0, 1, 2))
275     return SQLPreprocessor(temp_db_conn, def_config, tmp_path)