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
  17 class _TestingCursor(psycopg2.extras.DictCursor):
  18     """ Extension to the DictCursor class that provides execution
  19         short-cuts that simplify writing assertions.
  20     """
  21
  22     def scalar(self, sql, params=None):
  23         """ Execute a query with a single return value and return this value.
  24             Raises an assertion when not exactly one row is returned.
  25         """
  26         self.execute(sql, params)
  27         assert self.rowcount == 1
  28         return self.fetchone()[0]
  29
  30     def row_set(self, sql, params=None):
  31         """ Execute a query and return the result as a set of tuples.
  32         """
  33         self.execute(sql, params)
  34         if self.rowcount == 1:
  35             return set(tuple(self.fetchone()))
  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
  47 @pytest.fixture
  48 def temp_db(monkeypatch):
  49     """ Create an empty database for the test. The database name is also
  50         exported into NOMINATIM_DATABASE_DSN.
  51     """
  52     name = 'test_nominatim_python_unittest'
  53     conn = psycopg2.connect(database='postgres')
  54
  55     conn.set_isolation_level(0)
  56     with conn.cursor() as cur:
  57         cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
  58         cur.execute('CREATE DATABASE {}'.format(name))
  59
  60     conn.close()
  61
  62     monkeypatch.setenv('NOMINATIM_DATABASE_DSN' , 'dbname=' + name)
  63
  64     yield name
  65
  66     conn = psycopg2.connect(database='postgres')
  67
  68     conn.set_isolation_level(0)
  69     with conn.cursor() as cur:
  70         cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
  71
  72     conn.close()
  73
  74
  75 @pytest.fixture
  76 def dsn(temp_db):
  77     return 'dbname=' + temp_db
  78
  79
  80 @pytest.fixture
  81 def temp_db_with_extensions(temp_db):
  82     conn = psycopg2.connect(database=temp_db)
  83     with conn.cursor() as cur:
  84         cur.execute('CREATE EXTENSION hstore; CREATE EXTENSION postgis;')
  85     conn.commit()
  86     conn.close()
  87
  88     return temp_db
  89
  90 @pytest.fixture
  91 def temp_db_conn(temp_db):
  92     """ Connection to the test database.
  93     """
  94     with connection.connect('dbname=' + temp_db) as conn:
  95         yield conn
  96
  97
  98 @pytest.fixture
  99 def temp_db_cursor(temp_db):
 100     """ Connection and cursor towards the test database. The connection will
 101         be in auto-commit mode.
 102     """
 103     conn = psycopg2.connect('dbname=' + temp_db)
 104     conn.set_isolation_level(0)
 105     with conn.cursor(cursor_factory=_TestingCursor) as cur:
 106         yield cur
 107     conn.close()
 108
 109
 110 @pytest.fixture
 111 def table_factory(temp_db_cursor):
 112     def mk_table(name, definition='id INT'):
 113         temp_db_cursor.execute('CREATE TABLE {} ({})'.format(name, definition))
 114
 115     return mk_table
 116
 117
 118 @pytest.fixture
 119 def def_config():
 120     return Configuration(None, SRC_DIR.resolve() / 'settings')
 121
 122 @pytest.fixture
 123 def src_dir():
 124     return SRC_DIR.resolve()
 125
 126 @pytest.fixture
 127 def status_table(temp_db_conn):
 128     """ Create an empty version of the status table and
 129         the status logging table.
 130     """
 131     with temp_db_conn.cursor() as cur:
 132         cur.execute("""CREATE TABLE import_status (
 133                            lastimportdate timestamp with time zone NOT NULL,
 134                            sequence_id integer,
 135                            indexed boolean
 136                        )""")
 137         cur.execute("""CREATE TABLE import_osmosis_log (
 138                            batchend timestamp,
 139                            batchseq integer,
 140                            batchsize bigint,
 141                            starttime timestamp,
 142                            endtime timestamp,
 143                            event text
 144                            )""")
 145     temp_db_conn.commit()
 146
 147
 148 @pytest.fixture
 149 def place_table(temp_db_with_extensions, temp_db_conn):
 150     """ Create an empty version of the place table.
 151     """
 152     with temp_db_conn.cursor() as cur:
 153         cur.execute("""CREATE TABLE place (
 154                            osm_id int8 NOT NULL,
 155                            osm_type char(1) NOT NULL,
 156                            class text NOT NULL,
 157                            type text NOT NULL,
 158                            name hstore,
 159                            admin_level smallint,
 160                            address hstore,
 161                            extratags hstore,
 162                            geometry Geometry(Geometry,4326) NOT NULL)""")
 163     temp_db_conn.commit()
 164
 165
 166 @pytest.fixture
 167 def place_row(place_table, temp_db_cursor):
 168     """ A factory for rows in the place table. The table is created as a
 169         prerequisite to the fixture.
 170     """
 171     idseq = itertools.count(1001)
 172     def _insert(osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
 173                 admin_level=None, address=None, extratags=None, geom=None):
 174         temp_db_cursor.execute("INSERT INTO place VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
 175                                (osm_id or next(idseq), osm_type, cls, typ, names,
 176                                 admin_level, address, extratags,
 177                                 geom or 'SRID=4326;POINT(0 0 )'))
 178
 179     return _insert
 180
 181 @pytest.fixture
 182 def placex_table(temp_db_with_extensions, temp_db_conn):
 183     """ Create an empty version of the place table.
 184     """
 185     with temp_db_conn.cursor() as cur:
 186         cur.execute("""CREATE TABLE placex (
 187                            place_id BIGINT NOT NULL,
 188                            parent_place_id BIGINT,
 189                            linked_place_id BIGINT,
 190                            importance FLOAT,
 191                            indexed_date TIMESTAMP,
 192                            geometry_sector INTEGER,
 193                            rank_address SMALLINT,
 194                            rank_search SMALLINT,
 195                            partition SMALLINT,
 196                            indexed_status SMALLINT,
 197                            osm_id int8,
 198                            osm_type char(1),
 199                            class text,
 200                            type text,
 201                            name hstore,
 202                            admin_level smallint,
 203                            address hstore,
 204                            extratags hstore,
 205                            geometry Geometry(Geometry,4326),
 206                            wikipedia TEXT,
 207                            country_code varchar(2),
 208                            housenumber TEXT,
 209                            postcode TEXT,
 210                            centroid GEOMETRY(Geometry, 4326))
 211                            """)
 212     temp_db_conn.commit()
 213
 214
 215 @pytest.fixture
 216 def osm2pgsql_options(temp_db):
 217     return dict(osm2pgsql='echo',
 218                 osm2pgsql_cache=10,
 219                 osm2pgsql_style='style.file',
 220                 threads=1,
 221                 dsn='dbname=' + temp_db,
 222                 flatnode_file='',
 223                 tablespaces=dict(slim_data='', slim_index='',
 224                                  main_data='', main_index=''))