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
   9 def test_execute_file_success(temp_db, tmp_path):
  10     tmpfile = tmp_path / 'test.sql'
  11     tmpfile.write_text('CREATE TABLE test (id INT);\nINSERT INTO test VALUES(56);')
  12
  13     with psycopg2.connect('dbname=' + temp_db) as conn:
  14         db_utils.execute_file(conn, tmpfile)
  15
  16         with conn.cursor() as cur:
  17             cur.execute('SELECT * FROM test')
  18
  19             assert cur.rowcount == 1
  20             assert cur.fetchone()[0] == 56
  21
  22 def test_execute_file_bad_file(temp_db, tmp_path):
  23     with psycopg2.connect('dbname=' + temp_db) as conn:
  24         with pytest.raises(FileNotFoundError):
  25             db_utils.execute_file(conn, tmp_path / 'test2.sql')
  26
  27 def test_execute_file_bad_sql(temp_db, tmp_path):
  28     tmpfile = tmp_path / 'test.sql'
  29     tmpfile.write_text('CREATE STABLE test (id INT)')
  30
  31     with psycopg2.connect('dbname=' + temp_db) as conn:
  32         with pytest.raises(psycopg2.ProgrammingError):
  33             db_utils.execute_file(conn, tmpfile)