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