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     conn = connect('dbname=' + temp_db)
  11     yield conn
  12     conn.close()
  13
  14
  15 def test_connection_table_exists(db, temp_db_cursor):
  16     assert db.table_exists('foobar') == False
  17
  18     temp_db_cursor.execute('CREATE TABLE foobar (id INT)')
  19
  20     assert db.table_exists('foobar') == True
  21
  22
  23 def test_connection_index_exists(db, temp_db_cursor):
  24     assert db.index_exists('some_index') == False
  25
  26     temp_db_cursor.execute('CREATE TABLE foobar (id INT)')
  27     temp_db_cursor.execute('CREATE INDEX some_index ON foobar(id)')
  28
  29     assert db.index_exists('some_index') == True
  30     assert db.index_exists('some_index', table='foobar') == True
  31     assert db.index_exists('some_index', table='bar') == False
  32
  33
  34 def test_connection_server_version_tuple(db):
  35     ver = db.server_version_tuple()
  36
  37     assert isinstance(ver, tuple)
  38     assert len(ver) == 2
  39     assert ver[0] > 8
  40
  41 def test_cursor_scalar(db, temp_db_cursor):
  42     temp_db_cursor.execute('CREATE TABLE dummy (id INT)')
  43
  44     with db.cursor() as cur:
  45         assert cur.scalar('SELECT count(*) FROM dummy') == 0
  46
  47
  48 def test_cursor_scalar_many_rows(db):
  49     with db.cursor() as cur:
  50         with pytest.raises(RuntimeError):
  51             cur.scalar('SELECT * FROM pg_tables')