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     assert temp_db_cursor.row_set('SELECT * FROM test') == {(56, )}
  17
  18 def test_execute_file_bad_file(dsn, tmp_path):
  19     with pytest.raises(FileNotFoundError):
  20         db_utils.execute_file(dsn, tmp_path / 'test2.sql')
  21
  22
  23 def test_execute_file_bad_sql(dsn, tmp_path):
  24     tmpfile = tmp_path / 'test.sql'
  25     tmpfile.write_text('CREATE STABLE test (id INT)')
  26
  27     with pytest.raises(UsageError):
  28         db_utils.execute_file(dsn, tmpfile)
  29
  30
  31 def test_execute_file_bad_sql_ignore_errors(dsn, tmp_path):
  32     tmpfile = tmp_path / 'test.sql'
  33     tmpfile.write_text('CREATE STABLE test (id INT)')
  34
  35     db_utils.execute_file(dsn, tmpfile, ignore_errors=True)
  36
  37
  38 def test_execute_file_with_pre_code(dsn, tmp_path, temp_db_cursor):
  39     tmpfile = tmp_path / 'test.sql'
  40     tmpfile.write_text('INSERT INTO test VALUES(4)')
  41
  42     db_utils.execute_file(dsn, tmpfile, pre_code='CREATE TABLE test (id INT)')
  43
  44     assert temp_db_cursor.row_set('SELECT * FROM test') == {(4, )}
  45
  46
  47 def test_execute_file_with_post_code(dsn, tmp_path, temp_db_cursor):
  48     tmpfile = tmp_path / 'test.sql'
  49     tmpfile.write_text('CREATE TABLE test (id INT)')
  50
  51     db_utils.execute_file(dsn, tmpfile, post_code='INSERT INTO test VALUES(23)')
  52
  53     assert temp_db_cursor.row_set('SELECT * FROM test') == {(23, )}