test/python/conftest.py

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