4 from pathlib import Path
9 SRC_DIR = Path(__file__) / '..' / '..' / '..'
11 # always test against the source
12 sys.path.insert(0, str(SRC_DIR.resolve()))
14 from nominatim.config import Configuration
15 from nominatim.db import connection
16 from nominatim.db.sql_preprocessor import SQLPreprocessor
17 from nominatim.db import properties
18 import nominatim.tokenizer.factory
21 import dummy_tokenizer
23 from cursor import CursorForTesting
27 def temp_db(monkeypatch):
28 """ Create an empty database for the test. The database name is also
29 exported into NOMINATIM_DATABASE_DSN.
31 name = 'test_nominatim_python_unittest'
32 conn = psycopg2.connect(database='postgres')
34 conn.set_isolation_level(0)
35 with conn.cursor() as cur:
36 cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
37 cur.execute('CREATE DATABASE {}'.format(name))
41 monkeypatch.setenv('NOMINATIM_DATABASE_DSN', 'dbname=' + name)
45 conn = psycopg2.connect(database='postgres')
47 conn.set_isolation_level(0)
48 with conn.cursor() as cur:
49 cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
56 return 'dbname=' + temp_db
60 def temp_db_with_extensions(temp_db):
61 conn = psycopg2.connect(database=temp_db)
62 with conn.cursor() as cur:
63 cur.execute('CREATE EXTENSION hstore; CREATE EXTENSION postgis;')
70 def temp_db_conn(temp_db):
71 """ Connection to the test database.
73 with connection.connect('dbname=' + temp_db) as conn:
78 def temp_db_cursor(temp_db):
79 """ Connection and cursor towards the test database. The connection will
80 be in auto-commit mode.
82 conn = psycopg2.connect('dbname=' + temp_db)
83 conn.set_isolation_level(0)
84 with conn.cursor(cursor_factory=CursorForTesting) as cur:
90 def table_factory(temp_db_cursor):
91 def mk_table(name, definition='id INT', content=None):
92 temp_db_cursor.execute('CREATE TABLE {} ({})'.format(name, definition))
93 if content is not None:
94 temp_db_cursor.execute_values("INSERT INTO {} VALUES %s".format(name), content)
101 cfg = Configuration(None, SRC_DIR.resolve() / 'settings')
102 cfg.set_libdirs(module='.', osm2pgsql='.',
103 php=SRC_DIR / 'lib-php',
104 sql=SRC_DIR / 'lib-sql',
105 data=SRC_DIR / 'data')
111 return SRC_DIR.resolve()
116 def _call_nominatim(*args):
117 return nominatim.cli.nominatim(
118 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 tmp_phplib_dir():
132 with tempfile.TemporaryDirectory() as phpdir:
133 (Path(phpdir) / 'admin').mkdir()
139 def property_table(table_factory):
140 table_factory('nominatim_properties', 'property TEXT, value TEXT')
143 def status_table(temp_db_conn):
144 """ Create an empty version of the status table and
145 the status logging table.
147 with temp_db_conn.cursor() as cur:
148 cur.execute("""CREATE TABLE import_status (
149 lastimportdate timestamp with time zone NOT NULL,
153 cur.execute("""CREATE TABLE import_osmosis_log (
161 temp_db_conn.commit()
165 def place_table(temp_db_with_extensions, temp_db_conn):
166 """ Create an empty version of the place table.
168 with temp_db_conn.cursor() as cur:
169 cur.execute("""CREATE TABLE place (
170 osm_id int8 NOT NULL,
171 osm_type char(1) NOT NULL,
175 admin_level smallint,
178 geometry Geometry(Geometry,4326) NOT NULL)""")
179 temp_db_conn.commit()
183 def place_row(place_table, temp_db_cursor):
184 """ A factory for rows in the place table. The table is created as a
185 prerequisite to the fixture.
187 idseq = itertools.count(1001)
188 def _insert(osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
189 admin_level=None, address=None, extratags=None, geom=None):
190 temp_db_cursor.execute("INSERT INTO place VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
191 (osm_id or next(idseq), osm_type, cls, typ, names,
192 admin_level, address, extratags,
193 geom or 'SRID=4326;POINT(0 0)'))
198 def placex_table(temp_db_with_extensions, temp_db_conn):
199 """ Create an empty version of the place table.
201 return mocks.MockPlacexTable(temp_db_conn)
205 def osmline_table(temp_db_with_extensions, temp_db_conn):
206 with temp_db_conn.cursor() as cur:
207 cur.execute("""CREATE TABLE location_property_osmline (
210 parent_place_id BIGINT,
211 geometry_sector INTEGER,
212 indexed_date TIMESTAMP,
216 indexed_status SMALLINT,
218 interpolationtype TEXT,
221 country_code VARCHAR(2))""")
222 temp_db_conn.commit()
226 def word_table(temp_db_conn):
227 return mocks.MockWordTable(temp_db_conn)
231 def osm2pgsql_options(temp_db):
232 return dict(osm2pgsql='echo',
234 osm2pgsql_style='style.file',
236 dsn='dbname=' + temp_db,
238 tablespaces=dict(slim_data='', slim_index='',
239 main_data='', main_index=''))
242 def sql_preprocessor(temp_db_conn, tmp_path, monkeypatch, table_factory):
243 table_factory('country_name', 'partition INT', ((0, ), (1, ), (2, )))
244 cfg = Configuration(None, SRC_DIR.resolve() / 'settings')
245 cfg.set_libdirs(module='.', osm2pgsql='.', php=SRC_DIR / 'lib-php',
246 sql=tmp_path, data=SRC_DIR / 'data')
248 return SQLPreprocessor(temp_db_conn, cfg)
252 def tokenizer_mock(monkeypatch, property_table, temp_db_conn, tmp_path):
253 """ Sets up the configuration so that the test dummy tokenizer will be
254 loaded when the tokenizer factory is used. Also returns a factory
255 with which a new dummy tokenizer may be created.
257 monkeypatch.setenv('NOMINATIM_TOKENIZER', 'dummy')
259 def _import_dummy(module, *args, **kwargs):
260 return dummy_tokenizer
262 monkeypatch.setattr(nominatim.tokenizer.factory, "_import_tokenizer", _import_dummy)
263 properties.set_property(temp_db_conn, 'tokenizer', 'dummy')
265 def _create_tokenizer():
266 return dummy_tokenizer.DummyTokenizer(None, None)
268 return _create_tokenizer