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) 2024 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 json
  11
  12 import pytest
  13
  14 import nominatim_db.db.utils as db_utils
  15 from nominatim_db.errors import UsageError
  16
  17 def test_execute_file_success(dsn, temp_db_cursor, tmp_path):
  18     tmpfile = tmp_path / 'test.sql'
  19     tmpfile.write_text('CREATE TABLE test (id INT);\nINSERT INTO test VALUES(56);')
  20
  21     db_utils.execute_file(dsn, tmpfile)
  22
  23     assert temp_db_cursor.row_set('SELECT * FROM test') == {(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)
  43
  44
  45 def test_execute_file_with_pre_code(dsn, tmp_path, temp_db_cursor):
  46     tmpfile = tmp_path / 'test.sql'
  47     tmpfile.write_text('INSERT INTO test VALUES(4)')
  48
  49     db_utils.execute_file(dsn, tmpfile, pre_code='CREATE TABLE test (id INT)')
  50
  51     assert temp_db_cursor.row_set('SELECT * FROM test') == {(4, )}
  52
  53
  54 def test_execute_file_with_post_code(dsn, tmp_path, temp_db_cursor):
  55     tmpfile = tmp_path / 'test.sql'
  56     tmpfile.write_text('CREATE TABLE test (id INT)')
  57
  58     db_utils.execute_file(dsn, tmpfile, post_code='INSERT INTO test VALUES(23)')
  59
  60     assert temp_db_cursor.row_set('SELECT * FROM test') == {(23, )}