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 def test_execute_file_success(dsn, temp_db_cursor, tmp_path):
  11     tmpfile = tmp_path / 'test.sql'
  12     tmpfile.write_text('CREATE TABLE test (id INT);\nINSERT INTO test VALUES(56);')
  13
  14     db_utils.execute_file(dsn, tmpfile)
  15
  16     temp_db_cursor.execute('SELECT * FROM test')
  17
  18     assert temp_db_cursor.rowcount == 1
  19     assert temp_db_cursor.fetchone()[0] == 56
  20
  21 def test_execute_file_bad_file(dsn, tmp_path):
  22     with pytest.raises(FileNotFoundError):
  23         db_utils.execute_file(dsn, tmp_path / 'test2.sql')
  24
  25
  26 def test_execute_file_bad_sql(dsn, tmp_path):
  27     tmpfile = tmp_path / 'test.sql'
  28     tmpfile.write_text('CREATE STABLE test (id INT)')
  29
  30     with pytest.raises(UsageError):
  31         db_utils.execute_file(dsn, tmpfile)
  32
  33
  34 def test_execute_file_bad_sql_ignore_errors(dsn, tmp_path):
  35     tmpfile = tmp_path / 'test.sql'
  36     tmpfile.write_text('CREATE STABLE test (id INT)')
  37
  38     db_utils.execute_file(dsn, tmpfile, ignore_errors=True)
  39
  40
  41 def test_execute_file_with_pre_code(dsn, tmp_path, temp_db_cursor):
  42     tmpfile = tmp_path / 'test.sql'
  43     tmpfile.write_text('INSERT INTO test VALUES(4)')
  44
  45     db_utils.execute_file(dsn, tmpfile, pre_code='CREATE TABLE test (id INT)')
  46
  47     temp_db_cursor.execute('SELECT * FROM test')
  48
  49     assert temp_db_cursor.rowcount == 1
  50     assert temp_db_cursor.fetchone()[0] == 4
  51
  52
  53 def test_execute_file_with_post_code(dsn, tmp_path, temp_db_cursor):
  54     tmpfile = tmp_path / 'test.sql'
  55     tmpfile.write_text('CREATE TABLE test (id INT)')
  56
  57     db_utils.execute_file(dsn, tmpfile, post_code='INSERT INTO test VALUES(23)')
  58
  59     temp_db_cursor.execute('SELECT * FROM test')
  60
  61     assert temp_db_cursor.rowcount == 1
  62     assert temp_db_cursor.fetchone()[0] == 23