test/python/test_db_connection.py

   1 """
   2 Tests for specialised conenction and cursor classes.
   3 """
   4 import pytest
   5
   6 from nominatim.db.connection import connect
   7
   8 @pytest.fixture
   9 def db(temp_db):
  10     with connect('dbname=' + temp_db) as conn:
  11         yield conn
  12
  13
  14 def test_connection_table_exists(db, temp_db_cursor):
  15     assert db.table_exists('foobar') == False
  16
  17     temp_db_cursor.execute('CREATE TABLE foobar (id INT)')
  18
  19     assert db.table_exists('foobar') == True
  20
  21
  22 def test_connection_index_exists(db, temp_db_cursor):
  23     assert db.index_exists('some_index') == False
  24
  25     temp_db_cursor.execute('CREATE TABLE foobar (id INT)')
  26     temp_db_cursor.execute('CREATE INDEX some_index ON foobar(id)')
  27
  28     assert db.index_exists('some_index') == True
  29     assert db.index_exists('some_index', table='foobar') == True
  30     assert db.index_exists('some_index', table='bar') == False
  31
  32
  33 def test_connection_server_version_tuple(db):
  34     ver = db.server_version_tuple()
  35
  36     assert isinstance(ver, tuple)
  37     assert len(ver) == 2
  38     assert ver[0] > 8
  39
  40 def test_cursor_scalar(db, temp_db_cursor):
  41     temp_db_cursor.execute('CREATE TABLE dummy (id INT)')
  42
  43     with db.cursor() as cur:
  44         assert cur.scalar('SELECT count(*) FROM dummy') == 0
  45
  46
  47 def test_cursor_scalar_many_rows(db):
  48     with db.cursor() as cur:
  49         with pytest.raises(RuntimeError):
  50             cur.scalar('SELECT * FROM pg_tables')