3 from pathlib import Path
8 SRC_DIR = Path(__file__) / '..' / '..' / '..'
10 # always test against the source
11 sys.path.insert(0, str(SRC_DIR.resolve()))
13 from nominatim.config import Configuration
14 from nominatim.db import connection
15 from nominatim.db.sql_preprocessor import SQLPreprocessor
16 import nominatim.tokenizer.factory
19 import dummy_tokenizer
21 from cursor import CursorForTesting
25 def temp_db(monkeypatch):
26 """ Create an empty database for the test. The database name is also
27 exported into NOMINATIM_DATABASE_DSN.
29 name = 'test_nominatim_python_unittest'
30 conn = psycopg2.connect(database='postgres')
32 conn.set_isolation_level(0)
33 with conn.cursor() as cur:
34 cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
35 cur.execute('CREATE DATABASE {}'.format(name))
39 monkeypatch.setenv('NOMINATIM_DATABASE_DSN', 'dbname=' + name)
43 conn = psycopg2.connect(database='postgres')
45 conn.set_isolation_level(0)
46 with conn.cursor() as cur:
47 cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
54 return 'dbname=' + temp_db
58 def temp_db_with_extensions(temp_db):
59 conn = psycopg2.connect(database=temp_db)
60 with conn.cursor() as cur:
61 cur.execute('CREATE EXTENSION hstore; CREATE EXTENSION postgis;')
68 def temp_db_conn(temp_db):
69 """ Connection to the test database.
71 with connection.connect('dbname=' + temp_db) as conn:
76 def temp_db_cursor(temp_db):
77 """ Connection and cursor towards the test database. The connection will
78 be in auto-commit mode.
80 conn = psycopg2.connect('dbname=' + temp_db)
81 conn.set_isolation_level(0)
82 with conn.cursor(cursor_factory=CursorForTesting) as cur:
88 def table_factory(temp_db_cursor):
89 """ A fixture that creates new SQL tables, potentially filled with
92 def mk_table(name, definition='id INT', content=None):
93 temp_db_cursor.execute('CREATE TABLE {} ({})'.format(name, definition))
94 if content is not None:
95 temp_db_cursor.execute_values("INSERT INTO {} VALUES %s".format(name), content)
102 cfg = Configuration(None, SRC_DIR.resolve() / 'settings')
103 cfg.set_libdirs(module='.', osm2pgsql='.',
104 php=SRC_DIR / 'lib-php',
105 sql=SRC_DIR / 'lib-sql',
106 data=SRC_DIR / 'data')
112 return SRC_DIR.resolve()
117 def _call_nominatim(*args):
118 return nominatim.cli.nominatim(module_dir='MODULE NOT AVAILABLE',
119 osm2pgsql_path='OSM2PGSQL NOT AVAILABLE',
120 phplib_dir=str(SRC_DIR / 'lib-php'),
121 data_dir=str(SRC_DIR / 'data'),
122 phpcgi_path='/usr/bin/php-cgi',
123 sqllib_dir=str(SRC_DIR / 'lib-sql'),
124 config_dir=str(SRC_DIR / 'settings'),
127 return _call_nominatim
131 def property_table(table_factory, temp_db_conn):
132 table_factory('nominatim_properties', 'property TEXT, value TEXT')
134 return mocks.MockPropertyTable(temp_db_conn)
138 def status_table(table_factory):
139 """ Create an empty version of the status table and
140 the status logging table.
142 table_factory('import_status',
143 """lastimportdate timestamp with time zone NOT NULL,
146 table_factory('import_osmosis_log',
147 """batchend timestamp,
156 def place_table(temp_db_with_extensions, table_factory):
157 """ Create an empty version of the place table.
159 table_factory('place',
160 """osm_id int8 NOT NULL,
161 osm_type char(1) NOT NULL,
165 admin_level smallint,
168 geometry Geometry(Geometry,4326) NOT NULL""")
172 def place_row(place_table, temp_db_cursor):
173 """ A factory for rows in the place table. The table is created as a
174 prerequisite to the fixture.
176 idseq = itertools.count(1001)
177 def _insert(osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
178 admin_level=None, address=None, extratags=None, geom=None):
179 temp_db_cursor.execute("INSERT INTO place VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
180 (osm_id or next(idseq), osm_type, cls, typ, names,
181 admin_level, address, extratags,
182 geom or 'SRID=4326;POINT(0 0)'))
187 def placex_table(temp_db_with_extensions, temp_db_conn):
188 """ Create an empty version of the place table.
190 return mocks.MockPlacexTable(temp_db_conn)
194 def osmline_table(temp_db_with_extensions, table_factory):
195 table_factory('location_property_osmline',
198 parent_place_id BIGINT,
199 geometry_sector INTEGER,
200 indexed_date TIMESTAMP,
204 indexed_status SMALLINT,
206 interpolationtype TEXT,
209 country_code VARCHAR(2)""")
213 def word_table(temp_db_conn):
214 return mocks.MockWordTable(temp_db_conn)
218 def osm2pgsql_options(temp_db):
219 return dict(osm2pgsql='echo',
221 osm2pgsql_style='style.file',
223 dsn='dbname=' + temp_db,
225 tablespaces=dict(slim_data='', slim_index='',
226 main_data='', main_index=''))
229 def sql_preprocessor(temp_db_conn, tmp_path, table_factory):
230 table_factory('country_name', 'partition INT', ((0, ), (1, ), (2, )))
231 cfg = Configuration(None, SRC_DIR.resolve() / 'settings')
232 cfg.set_libdirs(module='.', osm2pgsql='.', php=SRC_DIR / 'lib-php',
233 sql=tmp_path, data=SRC_DIR / 'data')
235 return SQLPreprocessor(temp_db_conn, cfg)
239 def tokenizer_mock(monkeypatch, property_table):
240 """ Sets up the configuration so that the test dummy tokenizer will be
241 loaded when the tokenizer factory is used. Also returns a factory
242 with which a new dummy tokenizer may be created.
244 monkeypatch.setenv('NOMINATIM_TOKENIZER', 'dummy')
246 def _import_dummy(*args, **kwargs):
247 return dummy_tokenizer
249 monkeypatch.setattr(nominatim.tokenizer.factory, "_import_tokenizer", _import_dummy)
250 property_table.set('tokenizer', 'dummy')
252 def _create_tokenizer():
253 return dummy_tokenizer.DummyTokenizer(None, None)
255 return _create_tokenizer