test/python/conftest.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) 2026 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 import itertools
   8 import sys
   9 from pathlib import Path
  10
  11 import psycopg
  12 from psycopg import sql as pysql
  13 import pytest
  14
  15 # always test against the source
  16 SRC_DIR = (Path(__file__) / '..' / '..' / '..').resolve()
  17 sys.path.insert(0, str(SRC_DIR / 'src'))
  18
  19 from nominatim_db.config import Configuration
  20 from nominatim_db.db import connection, properties
  21 from nominatim_db.db.sql_preprocessor import SQLPreprocessor
  22 import nominatim_db.tokenizer.factory
  23
  24 import dummy_tokenizer
  25 from cursor import CursorForTesting
  26
  27
  28 def _with_srid(geom, default=None):
  29     if geom is None:
  30         return None if default is None else f"SRID=4326;{default}"
  31
  32     return f"SRID=4326;{geom}"
  33
  34
  35 @pytest.fixture
  36 def src_dir():
  37     return SRC_DIR
  38
  39
  40 @pytest.fixture
  41 def temp_db(monkeypatch):
  42     """ Create an empty database for the test. The database name is also
  43         exported into NOMINATIM_DATABASE_DSN.
  44     """
  45     name = 'test_nominatim_python_unittest'
  46
  47     with psycopg.connect(dbname='postgres', autocommit=True) as conn:
  48         with conn.cursor() as cur:
  49             cur.execute(pysql.SQL('DROP DATABASE IF EXISTS') + pysql.Identifier(name))
  50             cur.execute(pysql.SQL('CREATE DATABASE') + pysql.Identifier(name))
  51
  52     monkeypatch.setenv('NOMINATIM_DATABASE_DSN', 'dbname=' + name)
  53
  54     with psycopg.connect(dbname=name) as conn:
  55         with conn.cursor() as cur:
  56             cur.execute('CREATE EXTENSION hstore')
  57
  58     yield name
  59
  60     with psycopg.connect(dbname='postgres', autocommit=True) as conn:
  61         with conn.cursor() as cur:
  62             cur.execute(pysql.SQL('DROP DATABASE IF EXISTS') + pysql.Identifier(name))
  63
  64
  65 @pytest.fixture
  66 def dsn(temp_db):
  67     return 'dbname=' + temp_db
  68
  69
  70 @pytest.fixture
  71 def temp_db_with_extensions(temp_db):
  72     with psycopg.connect(dbname=temp_db) as conn:
  73         with conn.cursor() as cur:
  74             cur.execute('CREATE EXTENSION postgis')
  75
  76     return temp_db
  77
  78
  79 @pytest.fixture
  80 def temp_db_conn(temp_db):
  81     """ Connection to the test database.
  82     """
  83     with connection.connect('', autocommit=True, dbname=temp_db) as conn:
  84         connection.register_hstore(conn)
  85         yield conn
  86
  87
  88 @pytest.fixture
  89 def temp_db_cursor(temp_db):
  90     """ Connection and cursor towards the test database. The connection will
  91         be in auto-commit mode.
  92     """
  93     with psycopg.connect(dbname=temp_db, autocommit=True, cursor_factory=CursorForTesting) as conn:
  94         connection.register_hstore(conn)
  95         with conn.cursor() as cur:
  96             yield cur
  97
  98
  99 @pytest.fixture
 100 def table_factory(temp_db_conn):
 101     """ A fixture that creates new SQL tables, potentially filled with
 102         content.
 103     """
 104     def mk_table(name, definition='id INT', content=None):
 105         with psycopg.ClientCursor(temp_db_conn) as cur:
 106             cur.execute(pysql.SQL("CREATE TABLE {} ({})")
 107                              .format(pysql.Identifier(name),
 108                                      pysql.SQL(definition)))
 109             if content:
 110                 sql = pysql.SQL("INSERT INTO {} VALUES ({})")\
 111                            .format(pysql.Identifier(name),
 112                                    pysql.SQL(',').join([pysql.Placeholder()
 113                                                         for _ in range(len(content[0]))]))
 114                 cur.executemany(sql, content)
 115
 116     return mk_table
 117
 118
 119 @pytest.fixture
 120 def def_config():
 121     cfg = Configuration(None)
 122     return cfg
 123
 124
 125 @pytest.fixture
 126 def project_env(tmp_path):
 127     projdir = tmp_path / 'project'
 128     projdir.mkdir()
 129     cfg = Configuration(projdir)
 130     return cfg
 131
 132
 133 @pytest.fixture
 134 def country_table(table_factory):
 135     table_factory('country_name', 'partition INT, country_code varchar(2), name hstore')
 136
 137
 138 @pytest.fixture
 139 def country_row(country_table, temp_db_cursor):
 140     def _add(partition=None, country=None, names=None):
 141         temp_db_cursor.insert_row('country_name', partition=partition,
 142                                   country_code=country, name=names)
 143
 144     return _add
 145
 146
 147 @pytest.fixture
 148 def load_sql(temp_db_conn, country_row):
 149     proc = SQLPreprocessor(temp_db_conn, Configuration(None))
 150
 151     def _run(filename, **kwargs):
 152         proc.run_sql_file(temp_db_conn, filename, **kwargs)
 153
 154     return _run
 155
 156
 157 @pytest.fixture
 158 def property_table(load_sql, temp_db_conn):
 159     load_sql('tables/nominatim_properties.sql')
 160
 161     class _PropTable:
 162         def set(self, name, value):
 163             properties.set_property(temp_db_conn, name, value)
 164
 165         def get(self, name):
 166             return properties.get_property(temp_db_conn, name)
 167
 168     return _PropTable()
 169
 170
 171 @pytest.fixture
 172 def status_table(load_sql):
 173     """ Create an empty version of the status table and
 174         the status logging table.
 175     """
 176     load_sql('tables/status.sql')
 177
 178
 179 @pytest.fixture
 180 def place_table(temp_db_with_extensions, table_factory):
 181     """ Create an empty version of the place table.
 182     """
 183     table_factory('place',
 184                   """osm_id int8 NOT NULL,
 185                      osm_type char(1) NOT NULL,
 186                      class text NOT NULL,
 187                      type text NOT NULL,
 188                      name hstore,
 189                      admin_level smallint,
 190                      address hstore,
 191                      extratags hstore,
 192                      geometry Geometry(Geometry,4326) NOT NULL""")
 193
 194
 195 @pytest.fixture
 196 def place_row(place_table, temp_db_cursor):
 197     """ A factory for rows in the place table. The table is created as a
 198         prerequisite to the fixture.
 199     """
 200     idseq = itertools.count(1001)
 201
 202     def _insert(osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
 203                 admin_level=None, address=None, extratags=None, geom='POINT(0 0)'):
 204         args = {'osm_type': osm_type, 'osm_id': osm_id or next(idseq),
 205                 'class': cls, 'type': typ, 'name': names, 'admin_level': admin_level,
 206                 'address': address, 'extratags': extratags,
 207                 'geometry': _with_srid(geom)}
 208         temp_db_cursor.insert_row('place', **args)
 209
 210     return _insert
 211
 212
 213 @pytest.fixture
 214 def place_postcode_table(temp_db_with_extensions, table_factory):
 215     """ Create an empty version of the place_postcode table.
 216     """
 217     table_factory('place_postcode',
 218                   """osm_type char(1) NOT NULL,
 219                      osm_id bigint NOT NULL,
 220                      postcode text NOT NULL,
 221                      country_code text,
 222                      centroid Geometry(Point, 4326) NOT NULL,
 223                      geometry Geometry(Geometry, 4326)""")
 224
 225
 226 @pytest.fixture
 227 def place_postcode_row(place_postcode_table, temp_db_cursor):
 228     """ A factory for rows in the place_postcode table. The table is created as a
 229         prerequisite to the fixture.
 230     """
 231     idseq = itertools.count(5001)
 232
 233     def _insert(osm_type='N', osm_id=None, postcode=None, country=None,
 234                 centroid='POINT(12.0 4.0)', geom=None):
 235         temp_db_cursor.insert_row('place_postcode',
 236                                   osm_type=osm_type, osm_id=osm_id or next(idseq),
 237                                   postcode=postcode, country_code=country,
 238                                   centroid=_with_srid(centroid),
 239                                   geometry=_with_srid(geom))
 240
 241     return _insert
 242
 243
 244 @pytest.fixture
 245 def placex_table(temp_db_with_extensions, temp_db_conn, load_sql, place_table):
 246     """ Create an empty version of the placex table.
 247     """
 248     load_sql('tables/placex.sql')
 249     temp_db_conn.execute("CREATE SEQUENCE IF NOT EXISTS seq_place START 1")
 250
 251
 252 @pytest.fixture
 253 def placex_row(placex_table, temp_db_cursor):
 254     """ A factory for rows in the placex table. The table is created as a
 255         prerequisite to the fixture.
 256     """
 257     idseq = itertools.count(1001)
 258
 259     def _add(osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
 260              admin_level=None, address=None, extratags=None, geom='POINT(10 4)',
 261              country=None, housenumber=None, rank_search=30, rank_address=30,
 262              centroid='POINT(10 4)', indexed_status=0, indexed_date=None):
 263         args = {'place_id': pysql.SQL("nextval('seq_place')"),
 264                 'osm_type': osm_type, 'osm_id': osm_id or next(idseq),
 265                 'class': cls, 'type': typ, 'name': names, 'admin_level': admin_level,
 266                 'address': address, 'housenumber': housenumber,
 267                 'rank_search': rank_search, 'rank_address': rank_address,
 268                 'extratags': extratags,
 269                 'centroid': _with_srid(centroid), 'geometry': _with_srid(geom),
 270                 'country_code': country,
 271                 'indexed_status': indexed_status, 'indexed_date': indexed_date,
 272                 'partition': pysql.Literal(0), 'geometry_sector': pysql.Literal(1)}
 273         return temp_db_cursor.insert_row('placex', **args)
 274
 275     return _add
 276
 277
 278 @pytest.fixture
 279 def osmline_table(temp_db_with_extensions, load_sql):
 280     load_sql('tables/interpolation.sql')
 281
 282
 283 @pytest.fixture
 284 def sql_preprocessor_cfg(tmp_path, table_factory, temp_db_with_extensions, country_row):
 285     for part in range(3):
 286         country_row(partition=part)
 287
 288     cfg = Configuration(None)
 289     cfg.set_libdirs(sql=tmp_path)
 290     return cfg
 291
 292
 293 @pytest.fixture
 294 def sql_preprocessor(sql_preprocessor_cfg, temp_db_conn):
 295     return SQLPreprocessor(temp_db_conn, sql_preprocessor_cfg)
 296
 297
 298 @pytest.fixture
 299 def tokenizer_mock(monkeypatch, property_table):
 300     """ Sets up the configuration so that the test dummy tokenizer will be
 301         loaded when the tokenizer factory is used. Also returns a factory
 302         with which a new dummy tokenizer may be created.
 303     """
 304     monkeypatch.setenv('NOMINATIM_TOKENIZER', 'dummy')
 305
 306     def _import_dummy(*args, **kwargs):
 307         return dummy_tokenizer
 308
 309     monkeypatch.setattr(nominatim_db.tokenizer.factory,
 310                         "_import_tokenizer", _import_dummy)
 311     property_table.set('tokenizer', 'dummy')
 312
 313     def _create_tokenizer():
 314         return dummy_tokenizer.DummyTokenizer(None)
 315
 316     return _create_tokenizer