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 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')
  51
  52
  53 def test_get_pg_env_add_variable(monkeypatch):
  54     monkeypatch.delenv('PGPASSWORD', raising=False)
  55     env = get_pg_env('user=fooF')
  56
  57     assert env['PGUSER'] == 'fooF'
  58     assert 'PGPASSWORD' not in env
  59
  60
  61 def test_get_pg_env_overwrite_variable(monkeypatch):
  62     monkeypatch.setenv('PGUSER', 'some default')
  63     env = get_pg_env('user=overwriter')
  64
  65     assert env['PGUSER'] == 'overwriter'
  66
  67
  68 def test_get_pg_env_ignore_unknown():
  69     env = get_pg_env('tty=stuff', base_env={})
  70
  71     assert env == {}