test/python/conftest.py

   1 import importlib
   2 import itertools
   3 import sys
   4 from pathlib import Path
   5
   6 import psycopg2
   7 import psycopg2.extras
   8 import pytest
   9 import tempfile
  10
  11 SRC_DIR = Path(__file__) / '..' / '..' / '..'
  12
  13 # always test against the source
  14 sys.path.insert(0, str(SRC_DIR.resolve()))
  15
  16 from nominatim.config import Configuration
  17 from nominatim.db import connection
  18 from nominatim.db.sql_preprocessor import SQLPreprocessor
  19 from nominatim.db import properties
  20
  21 import dummy_tokenizer
  22 import mocks
  23
  24 class _TestingCursor(psycopg2.extras.DictCursor):
  25     """ Extension to the DictCursor class that provides execution
  26         short-cuts that simplify writing assertions.
  27     """
  28
  29     def scalar(self, sql, params=None):
  30         """ Execute a query with a single return value and return this value.
  31             Raises an assertion when not exactly one row is returned.
  32         """
  33         self.execute(sql, params)
  34         assert self.rowcount == 1
  35         return self.fetchone()[0]
  36
  37     def row_set(self, sql, params=None):
  38         """ Execute a query and return the result as a set of tuples.
  39         """
  40         self.execute(sql, params)
  41
  42         return set((tuple(row) for row in self))
  43
  44     def table_exists(self, table):
  45         """ Check that a table with the given name exists in the database.
  46         """
  47         num = self.scalar("""SELECT count(*) FROM pg_tables
  48                              WHERE tablename = %s""", (table, ))
  49         return num == 1
  50
  51     def table_rows(self, table):
  52         """ Return the number of rows in the given table.
  53         """
  54         return self.scalar('SELECT count(*) FROM ' + table)
  55
  56
  57 @pytest.fixture
  58 def temp_db(monkeypatch):
  59     """ Create an empty database for the test. The database name is also
  60         exported into NOMINATIM_DATABASE_DSN.
  61     """
  62     name = 'test_nominatim_python_unittest'
  63     conn = psycopg2.connect(database='postgres')
  64
  65     conn.set_isolation_level(0)
  66     with conn.cursor() as cur:
  67         cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
  68         cur.execute('CREATE DATABASE {}'.format(name))
  69
  70     conn.close()
  71
  72     monkeypatch.setenv('NOMINATIM_DATABASE_DSN' , 'dbname=' + name)
  73
  74     yield name
  75
  76     conn = psycopg2.connect(database='postgres')
  77
  78     conn.set_isolation_level(0)
  79     with conn.cursor() as cur:
  80         cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
  81
  82     conn.close()
  83
  84
  85 @pytest.fixture
  86 def dsn(temp_db):
  87     return 'dbname=' + temp_db
  88
  89
  90 @pytest.fixture
  91 def temp_db_with_extensions(temp_db):
  92     conn = psycopg2.connect(database=temp_db)
  93     with conn.cursor() as cur:
  94         cur.execute('CREATE EXTENSION hstore; CREATE EXTENSION postgis;')
  95     conn.commit()
  96     conn.close()
  97
  98     return temp_db
  99
 100 @pytest.fixture
 101 def temp_db_conn(temp_db):
 102     """ Connection to the test database.
 103     """
 104     with connection.connect('dbname=' + temp_db) as conn:
 105         yield conn
 106
 107
 108 @pytest.fixture
 109 def temp_db_cursor(temp_db):
 110     """ Connection and cursor towards the test database. The connection will
 111         be in auto-commit mode.
 112     """
 113     conn = psycopg2.connect('dbname=' + temp_db)
 114     conn.set_isolation_level(0)
 115     with conn.cursor(cursor_factory=_TestingCursor) as cur:
 116         yield cur
 117     conn.close()
 118
 119
 120 @pytest.fixture
 121 def table_factory(temp_db_cursor):
 122     def mk_table(name, definition='id INT', content=None):
 123         temp_db_cursor.execute('CREATE TABLE {} ({})'.format(name, definition))
 124         if content is not None:
 125             psycopg2.extras.execute_values(
 126                 temp_db_cursor, "INSERT INTO {} VALUES %s".format(name), content)
 127
 128     return mk_table
 129
 130
 131 @pytest.fixture
 132 def def_config():
 133     cfg = Configuration(None, SRC_DIR.resolve() / 'settings')
 134     cfg.set_libdirs(module='.', osm2pgsql='.',
 135                     php=SRC_DIR / 'lib-php',
 136                     sql=SRC_DIR / 'lib-sql',
 137                     data=SRC_DIR / 'data')
 138     return cfg
 139
 140 @pytest.fixture
 141 def src_dir():
 142     return SRC_DIR.resolve()
 143
 144 @pytest.fixture
 145 def tmp_phplib_dir():
 146     with tempfile.TemporaryDirectory() as phpdir:
 147         (Path(phpdir) / 'admin').mkdir()
 148
 149         yield Path(phpdir)
 150
 151
 152 @pytest.fixture
 153 def property_table(table_factory):
 154     table_factory('nominatim_properties', 'property TEXT, value TEXT')
 155
 156 @pytest.fixture
 157 def status_table(temp_db_conn):
 158     """ Create an empty version of the status table and
 159         the status logging table.
 160     """
 161     with temp_db_conn.cursor() as cur:
 162         cur.execute("""CREATE TABLE import_status (
 163                            lastimportdate timestamp with time zone NOT NULL,
 164                            sequence_id integer,
 165                            indexed boolean
 166                        )""")
 167         cur.execute("""CREATE TABLE import_osmosis_log (
 168                            batchend timestamp,
 169                            batchseq integer,
 170                            batchsize bigint,
 171                            starttime timestamp,
 172                            endtime timestamp,
 173                            event text
 174                            )""")
 175     temp_db_conn.commit()
 176
 177
 178 @pytest.fixture
 179 def place_table(temp_db_with_extensions, temp_db_conn):
 180     """ Create an empty version of the place table.
 181     """
 182     with temp_db_conn.cursor() as cur:
 183         cur.execute("""CREATE TABLE 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     temp_db_conn.commit()
 194
 195
 196 @pytest.fixture
 197 def place_row(place_table, temp_db_cursor):
 198     """ A factory for rows in the place table. The table is created as a
 199         prerequisite to the fixture.
 200     """
 201     idseq = itertools.count(1001)
 202     def _insert(osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
 203                 admin_level=None, address=None, extratags=None, geom=None):
 204         temp_db_cursor.execute("INSERT INTO place VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
 205                                (osm_id or next(idseq), osm_type, cls, typ, names,
 206                                 admin_level, address, extratags,
 207                                 geom or 'SRID=4326;POINT(0 0)'))
 208
 209     return _insert
 210
 211 @pytest.fixture
 212 def placex_table(temp_db_with_extensions, temp_db_conn):
 213     """ Create an empty version of the place table.
 214     """
 215     return mocks.MockPlacexTable(temp_db_conn)
 216
 217
 218 @pytest.fixture
 219 def osmline_table(temp_db_with_extensions, temp_db_conn):
 220     with temp_db_conn.cursor() as cur:
 221         cur.execute("""CREATE TABLE location_property_osmline (
 222                            place_id BIGINT,
 223                            osm_id BIGINT,
 224                            parent_place_id BIGINT,
 225                            geometry_sector INTEGER,
 226                            indexed_date TIMESTAMP,
 227                            startnumber INTEGER,
 228                            endnumber INTEGER,
 229                            partition SMALLINT,
 230                            indexed_status SMALLINT,
 231                            linegeo GEOMETRY,
 232                            interpolationtype TEXT,
 233                            address HSTORE,
 234                            postcode TEXT,
 235                            country_code VARCHAR(2))""")
 236     temp_db_conn.commit()
 237
 238
 239 @pytest.fixture
 240 def word_table(temp_db_conn):
 241     return mocks.MockWordTable(temp_db_conn)
 242
 243
 244 @pytest.fixture
 245 def osm2pgsql_options(temp_db):
 246     return dict(osm2pgsql='echo',
 247                 osm2pgsql_cache=10,
 248                 osm2pgsql_style='style.file',
 249                 threads=1,
 250                 dsn='dbname=' + temp_db,
 251                 flatnode_file='',
 252                 tablespaces=dict(slim_data='', slim_index='',
 253                                  main_data='', main_index=''))
 254
 255 @pytest.fixture
 256 def sql_preprocessor(temp_db_conn, tmp_path, monkeypatch, table_factory):
 257     table_factory('country_name', 'partition INT', ((0, ), (1, ), (2, )))
 258     cfg = Configuration(None, SRC_DIR.resolve() / 'settings')
 259     cfg.set_libdirs(module='.', osm2pgsql='.', php=SRC_DIR / 'lib-php',
 260                     sql=tmp_path, data=SRC_DIR / 'data')
 261
 262     return SQLPreprocessor(temp_db_conn, cfg)
 263
 264
 265 @pytest.fixture
 266 def tokenizer_mock(monkeypatch, property_table, temp_db_conn, tmp_path):
 267     """ Sets up the configuration so that the test dummy tokenizer will be
 268         loaded when the tokenizer factory is used. Also returns a factory
 269         with which a new dummy tokenizer may be created.
 270     """
 271     monkeypatch.setenv('NOMINATIM_TOKENIZER', 'dummy')
 272
 273     def _import_dummy(module, *args, **kwargs):
 274         return dummy_tokenizer
 275
 276     monkeypatch.setattr(importlib, "import_module", _import_dummy)
 277     properties.set_property(temp_db_conn, 'tokenizer', 'dummy')
 278
 279     def _create_tokenizer():
 280         return dummy_tokenizer.DummyTokenizer(None, None)
 281
 282     return _create_tokenizer