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