test/python/conftest.py

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