test/python/conftest.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2022 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 psycopg2
  12 import pytest
  13
  14 # always test against the source
  15 SRC_DIR = (Path(__file__) / '..' / '..' / '..').resolve()
  16 sys.path.insert(0, str(SRC_DIR))
  17
  18 from nominatim.config import Configuration
  19 from nominatim.db import connection
  20 from nominatim.db.sql_preprocessor import SQLPreprocessor
  21 import nominatim.tokenizer.factory
  22
  23 import dummy_tokenizer
  24 import mocks
  25 from cursor import CursorForTesting
  26
  27
  28 @pytest.fixture
  29 def src_dir():
  30     return SRC_DIR
  31
  32
  33 @pytest.fixture
  34 def temp_db(monkeypatch):
  35     """ Create an empty database for the test. The database name is also
  36         exported into NOMINATIM_DATABASE_DSN.
  37     """
  38     name = 'test_nominatim_python_unittest'
  39     conn = psycopg2.connect(database='postgres')
  40
  41     conn.set_isolation_level(0)
  42     with conn.cursor() as cur:
  43         cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
  44         cur.execute('CREATE DATABASE {}'.format(name))
  45
  46     conn.close()
  47
  48     monkeypatch.setenv('NOMINATIM_DATABASE_DSN', 'dbname=' + name)
  49
  50     yield name
  51
  52     conn = psycopg2.connect(database='postgres')
  53
  54     conn.set_isolation_level(0)
  55     with conn.cursor() as cur:
  56         cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
  57
  58     conn.close()
  59
  60
  61 @pytest.fixture
  62 def dsn(temp_db):
  63     return 'dbname=' + temp_db
  64
  65
  66 @pytest.fixture
  67 def temp_db_with_extensions(temp_db):
  68     conn = psycopg2.connect(database=temp_db)
  69     with conn.cursor() as cur:
  70         cur.execute('CREATE EXTENSION hstore; CREATE EXTENSION postgis;')
  71     conn.commit()
  72     conn.close()
  73
  74     return temp_db
  75
  76 @pytest.fixture
  77 def temp_db_conn(temp_db):
  78     """ Connection to the test database.
  79     """
  80     with connection.connect('dbname=' + temp_db) as conn:
  81         yield conn
  82
  83
  84 @pytest.fixture
  85 def temp_db_cursor(temp_db):
  86     """ Connection and cursor towards the test database. The connection will
  87         be in auto-commit mode.
  88     """
  89     conn = psycopg2.connect('dbname=' + temp_db)
  90     conn.set_isolation_level(0)
  91     with conn.cursor(cursor_factory=CursorForTesting) as cur:
  92         yield cur
  93     conn.close()
  94
  95
  96 @pytest.fixture
  97 def table_factory(temp_db_cursor):
  98     """ A fixture that creates new SQL tables, potentially filled with
  99         content.
 100     """
 101     def mk_table(name, definition='id INT', content=None):
 102         temp_db_cursor.execute('CREATE TABLE {} ({})'.format(name, definition))
 103         if content is not None:
 104             temp_db_cursor.execute_values("INSERT INTO {} VALUES %s".format(name), content)
 105
 106     return mk_table
 107
 108
 109 @pytest.fixture
 110 def def_config(src_dir):
 111     cfg = Configuration(None, src_dir / 'settings')
 112     cfg.set_libdirs(module='.', osm2pgsql='.',
 113                     php=src_dir / 'lib-php',
 114                     sql=src_dir / 'lib-sql',
 115                     data=src_dir / 'data')
 116     return cfg
 117
 118
 119 @pytest.fixture
 120 def project_env(src_dir, tmp_path):
 121     projdir = tmp_path / 'project'
 122     projdir.mkdir()
 123     cfg = Configuration(projdir, src_dir / 'settings')
 124     cfg.set_libdirs(module='.', osm2pgsql='.',
 125                     php=src_dir / 'lib-php',
 126                     sql=src_dir / 'lib-sql',
 127                     data=src_dir / 'data')
 128     return cfg
 129
 130
 131 @pytest.fixture
 132 def property_table(table_factory, temp_db_conn):
 133     table_factory('nominatim_properties', 'property TEXT, value TEXT')
 134
 135     return mocks.MockPropertyTable(temp_db_conn)
 136
 137
 138 @pytest.fixture
 139 def status_table(table_factory):
 140     """ Create an empty version of the status table and
 141         the status logging table.
 142     """
 143     table_factory('import_status',
 144                   """lastimportdate timestamp with time zone NOT NULL,
 145                      sequence_id integer,
 146                      indexed boolean""")
 147     table_factory('import_osmosis_log',
 148                   """batchend timestamp,
 149                      batchseq integer,
 150                      batchsize bigint,
 151                      starttime timestamp,
 152                      endtime timestamp,
 153                      event text""")
 154
 155
 156 @pytest.fixture
 157 def place_table(temp_db_with_extensions, table_factory):
 158     """ Create an empty version of the place table.
 159     """
 160     table_factory('place',
 161                   """osm_id int8 NOT NULL,
 162                      osm_type char(1) NOT NULL,
 163                      class text NOT NULL,
 164                      type text NOT NULL,
 165                      name hstore,
 166                      admin_level smallint,
 167                      address hstore,
 168                      extratags hstore,
 169                      geometry Geometry(Geometry,4326) NOT NULL""")
 170
 171
 172 @pytest.fixture
 173 def place_row(place_table, temp_db_cursor):
 174     """ A factory for rows in the place table. The table is created as a
 175         prerequisite to the fixture.
 176     """
 177     psycopg2.extras.register_hstore(temp_db_cursor)
 178     idseq = itertools.count(1001)
 179     def _insert(osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
 180                 admin_level=None, address=None, extratags=None, geom=None):
 181         temp_db_cursor.execute("INSERT INTO place VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
 182                                (osm_id or next(idseq), osm_type, cls, typ, names,
 183                                 admin_level, address, extratags,
 184                                 geom or 'SRID=4326;POINT(0 0)'))
 185
 186     return _insert
 187
 188 @pytest.fixture
 189 def placex_table(temp_db_with_extensions, temp_db_conn):
 190     """ Create an empty version of the place table.
 191     """
 192     return mocks.MockPlacexTable(temp_db_conn)
 193
 194
 195 @pytest.fixture
 196 def osmline_table(temp_db_with_extensions, table_factory):
 197     table_factory('location_property_osmline',
 198                   """place_id BIGINT,
 199                      osm_id BIGINT,
 200                      parent_place_id BIGINT,
 201                      geometry_sector INTEGER,
 202                      indexed_date TIMESTAMP,
 203                      startnumber INTEGER,
 204                      endnumber INTEGER,
 205                      partition SMALLINT,
 206                      indexed_status SMALLINT,
 207                      linegeo GEOMETRY,
 208                      interpolationtype TEXT,
 209                      address HSTORE,
 210                      postcode TEXT,
 211                      country_code VARCHAR(2)""")
 212
 213
 214 @pytest.fixture
 215 def sql_preprocessor_cfg(tmp_path, table_factory, temp_db_with_extensions):
 216     table_factory('country_name', 'partition INT', ((0, ), (1, ), (2, )))
 217     cfg = Configuration(None, SRC_DIR.resolve() / 'settings')
 218     cfg.set_libdirs(module='.', osm2pgsql='.', php=SRC_DIR / 'lib-php',
 219                     sql=tmp_path, data=SRC_DIR / 'data')
 220     return cfg
 221
 222
 223 @pytest.fixture
 224 def sql_preprocessor(sql_preprocessor_cfg, temp_db_conn):
 225     return SQLPreprocessor(temp_db_conn, sql_preprocessor_cfg)
 226
 227
 228 @pytest.fixture
 229 def tokenizer_mock(monkeypatch, property_table):
 230     """ Sets up the configuration so that the test dummy tokenizer will be
 231         loaded when the tokenizer factory is used. Also returns a factory
 232         with which a new dummy tokenizer may be created.
 233     """
 234     monkeypatch.setenv('NOMINATIM_TOKENIZER', 'dummy')
 235
 236     def _import_dummy(*args, **kwargs):
 237         return dummy_tokenizer
 238
 239     monkeypatch.setattr(nominatim.tokenizer.factory, "_import_tokenizer", _import_dummy)
 240     property_table.set('tokenizer', 'dummy')
 241
 242     def _create_tokenizer():
 243         return dummy_tokenizer.DummyTokenizer(None, None)
 244
 245     return _create_tokenizer