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, get_pg_env
   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
  41 def test_connection_postgis_version_tuple(db, temp_db_cursor):
  42     temp_db_cursor.execute('CREATE EXTENSION postgis')
  43
  44     ver = db.postgis_version_tuple()
  45
  46     assert isinstance(ver, tuple)
  47     assert len(ver) == 2
  48     assert ver[0] >= 2
  49
  50
  51 def test_cursor_scalar(db, temp_db_cursor):
  52     temp_db_cursor.execute('CREATE TABLE dummy (id INT)')
  53
  54     with db.cursor() as cur:
  55         assert cur.scalar('SELECT count(*) FROM dummy') == 0
  56
  57
  58 def test_cursor_scalar_many_rows(db):
  59     with db.cursor() as cur:
  60         with pytest.raises(RuntimeError):
  61             cur.scalar('SELECT * FROM pg_tables')
  62
  63
  64 def test_get_pg_env_add_variable(monkeypatch):
  65     monkeypatch.delenv('PGPASSWORD', raising=False)
  66     env = get_pg_env('user=fooF')
  67
  68     assert env['PGUSER'] == 'fooF'
  69     assert 'PGPASSWORD' not in env
  70
  71
  72 def test_get_pg_env_overwrite_variable(monkeypatch):
  73     monkeypatch.setenv('PGUSER', 'some default')
  74     env = get_pg_env('user=overwriter')
  75
  76     assert env['PGUSER'] == 'overwriter'
  77
  78
  79 def test_get_pg_env_ignore_unknown():
  80     env = get_pg_env('tty=stuff', base_env={})
  81
  82     assert env == {}