test/python/test_db_utils.py

   1 """
   2 Tests for DB utility functions in db.utils
   3 """
   4 import psycopg2
   5 import pytest
   6
   7 import nominatim.db.utils as db_utils
   8 from nominatim.errors import UsageError
   9
  10 @pytest.fixture
  11 def dsn(temp_db):
  12     return 'dbname=' + temp_db
  13
  14 def test_execute_file_success(dsn, temp_db_cursor, tmp_path):
  15     tmpfile = tmp_path / 'test.sql'
  16     tmpfile.write_text('CREATE TABLE test (id INT);\nINSERT INTO test VALUES(56);')
  17
  18     db_utils.execute_file(dsn, tmpfile)
  19
  20     temp_db_cursor.execute('SELECT * FROM test')
  21
  22     assert temp_db_cursor.rowcount == 1
  23     assert temp_db_cursor.fetchone()[0] == 56
  24
  25 def test_execute_file_bad_file(dsn, tmp_path):
  26     with pytest.raises(FileNotFoundError):
  27         db_utils.execute_file(dsn, tmp_path / 'test2.sql')
  28
  29
  30 def test_execute_file_bad_sql(dsn, tmp_path):
  31     tmpfile = tmp_path / 'test.sql'
  32     tmpfile.write_text('CREATE STABLE test (id INT)')
  33
  34     with pytest.raises(UsageError):
  35         db_utils.execute_file(dsn, tmpfile)
  36
  37
  38 def test_execute_file_bad_sql_ignore_errors(dsn, tmp_path):
  39     tmpfile = tmp_path / 'test.sql'
  40     tmpfile.write_text('CREATE STABLE test (id INT)')
  41
  42     db_utils.execute_file(dsn, tmpfile, ignore_errors=True)