test/python/conftest.py

   1 import sys
   2 from pathlib import Path
   3
   4 import psycopg2
   5 import psycopg2.extras
   6 import pytest
   7
   8 SRC_DIR = Path(__file__) / '..' / '..' / '..'
   9
  10 # always test against the source
  11 sys.path.insert(0, str(SRC_DIR.resolve()))
  12
  13 from nominatim.config import Configuration
  14
  15 class _TestingCursor(psycopg2.extras.DictCursor):
  16     """ Extension to the DictCursor class that provides execution
  17         short-cuts that simplify writing assertions.
  18     """
  19
  20     def scalar(self, sql, params=None):
  21         """ Execute a query with a single return value and return this value.
  22             Raises an assertion when not exactly one row is returned.
  23         """
  24         self.execute(sql, params)
  25         assert self.rowcount == 1
  26         return self.fetchone()[0]
  27
  28     def row_set(self, sql, params=None):
  29         """ Execute a query and return the result as a set of tuples.
  30         """
  31         self.execute(sql, params)
  32         if self.rowcount == 1:
  33             return set(tuple(self.fetchone()))
  34
  35         return set((tuple(row) for row in self))
  36
  37 @pytest.fixture
  38 def temp_db(monkeypatch):
  39     """ Create an empty database for the test. The database name is also
  40         exported into NOMINATIM_DATABASE_DSN.
  41     """
  42     name = 'test_nominatim_python_unittest'
  43     with psycopg2.connect(database='postgres') as conn:
  44         conn.set_isolation_level(0)
  45         with conn.cursor() as cur:
  46             cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
  47             cur.execute('CREATE DATABASE {}'.format(name))
  48
  49     monkeypatch.setenv('NOMINATIM_DATABASE_DSN' , 'dbname=' + name)
  50
  51     yield name
  52
  53     with psycopg2.connect(database='postgres') as conn:
  54         conn.set_isolation_level(0)
  55         with conn.cursor() as cur:
  56             cur.execute('DROP DATABASE IF EXISTS {}'.format(name))
  57
  58
  59 @pytest.fixture
  60 def temp_db_conn(temp_db):
  61     """ Connection to the test database.
  62     """
  63     conn = psycopg2.connect(database=temp_db)
  64     yield conn
  65     conn.close()
  66
  67
  68 @pytest.fixture
  69 def temp_db_cursor(temp_db):
  70     """ Connection and cursor towards the test database. The connection will
  71         be in auto-commit mode.
  72     """
  73     conn = psycopg2.connect('dbname=' + temp_db)
  74     conn.set_isolation_level(0)
  75     with conn.cursor(cursor_factory=_TestingCursor) as cur:
  76         yield cur
  77     conn.close()
  78
  79
  80 @pytest.fixture
  81 def def_config():
  82     return Configuration(None, SRC_DIR.resolve() / 'settings')