test/python/db/test_utils.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2025 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Tests for DB utility functions in db.utils
   9 """
  10 import pytest
  11
  12 import nominatim_db.db.utils as db_utils
  13 from nominatim_db.errors import UsageError
  14
  15
  16 def test_execute_file_success(dsn, temp_db_cursor, tmp_path):
  17     tmpfile = tmp_path / 'test.sql'
  18     tmpfile.write_text(
  19         'CREATE TABLE test (id INT);\nINSERT INTO test VALUES(56);', encoding='utf-8')
  20
  21     db_utils.execute_file(dsn, tmpfile)
  22
  23     assert temp_db_cursor.row_set('SELECT * FROM test') == {(56, )}
  24
  25
  26 def test_execute_file_bad_file(dsn, tmp_path):
  27     with pytest.raises(FileNotFoundError):
  28         db_utils.execute_file(dsn, tmp_path / 'test2.sql')
  29
  30
  31 def test_execute_file_bad_sql(dsn, tmp_path):
  32     tmpfile = tmp_path / 'test.sql'
  33     tmpfile.write_text('CREATE STABLE test (id INT)', encoding='utf-8')
  34
  35     with pytest.raises(UsageError):
  36         db_utils.execute_file(dsn, tmpfile)
  37
  38
  39 def test_execute_file_bad_sql_ignore_errors(dsn, tmp_path):
  40     tmpfile = tmp_path / 'test.sql'
  41     tmpfile.write_text('CREATE STABLE test (id INT)', encoding='utf-8')
  42
  43     db_utils.execute_file(dsn, tmpfile, ignore_errors=True)
  44
  45
  46 def test_execute_file_with_pre_code(dsn, tmp_path, temp_db_cursor):
  47     tmpfile = tmp_path / 'test.sql'
  48     tmpfile.write_text('INSERT INTO test VALUES(4)', encoding='utf-8')
  49
  50     db_utils.execute_file(dsn, tmpfile, pre_code='CREATE TABLE test (id INT)')
  51
  52     assert temp_db_cursor.row_set('SELECT * FROM test') == {(4, )}
  53
  54
  55 def test_execute_file_with_post_code(dsn, tmp_path, temp_db_cursor):
  56     tmpfile = tmp_path / 'test.sql'
  57     tmpfile.write_text('CREATE TABLE test (id INT)', encoding='utf-8')
  58
  59     db_utils.execute_file(dsn, tmpfile, post_code='INSERT INTO test VALUES(23)')
  60
  61     assert temp_db_cursor.row_set('SELECT * FROM test') == {(23, )}