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