adapt tests for ICU tokenizer

[nominatim.git] / test / python / test_db_utils.py

diff --git a/test/python/test_db_utils.py b/test/python/test_db_utils.py
index e756f2c4ea2602216d00bb29c56d86b576facdd4..d549b70f803ec8f7e873e21e8c79f350e73f7af6 100644 (file)
--- a/test/python/test_db_utils.py
+++ b/test/python/test_db_utils.py
@@ -1,30 +1,52 @@
 """
 Tests for DB utility functions in db.utils
 """
-import psycopg2
 import pytest
 
 import nominatim.db.utils as db_utils
+from nominatim.errors import UsageError
 
-def test_execute_file_success(temp_db_conn, tmp_path):
+def test_execute_file_success(dsn, temp_db_cursor, tmp_path):
     tmpfile = tmp_path / 'test.sql'
     tmpfile.write_text('CREATE TABLE test (id INT);\nINSERT INTO test VALUES(56);')
 
-    db_utils.execute_file(temp_db_conn, tmpfile)
+    db_utils.execute_file(dsn, tmpfile)
 
-    with temp_db_conn.cursor() as cur:
-        cur.execute('SELECT * FROM test')
+    assert temp_db_cursor.row_set('SELECT * FROM test') == {(56, )}
 
-        assert cur.rowcount == 1
-        assert cur.fetchone()[0] == 56
-
-def test_execute_file_bad_file(temp_db_conn, tmp_path):
+def test_execute_file_bad_file(dsn, tmp_path):
     with pytest.raises(FileNotFoundError):
-        db_utils.execute_file(temp_db_conn, tmp_path / 'test2.sql')
+        db_utils.execute_file(dsn, tmp_path / 'test2.sql')
+
 
-def test_execute_file_bad_sql(temp_db_conn, tmp_path):
+def test_execute_file_bad_sql(dsn, tmp_path):
     tmpfile = tmp_path / 'test.sql'
     tmpfile.write_text('CREATE STABLE test (id INT)')
 
-    with pytest.raises(psycopg2.ProgrammingError):
-        db_utils.execute_file(temp_db_conn, tmpfile)
+    with pytest.raises(UsageError):
+        db_utils.execute_file(dsn, tmpfile)
+
+
+def test_execute_file_bad_sql_ignore_errors(dsn, tmp_path):
+    tmpfile = tmp_path / 'test.sql'
+    tmpfile.write_text('CREATE STABLE test (id INT)')
+
+    db_utils.execute_file(dsn, tmpfile, ignore_errors=True)
+
+
+def test_execute_file_with_pre_code(dsn, tmp_path, temp_db_cursor):
+    tmpfile = tmp_path / 'test.sql'
+    tmpfile.write_text('INSERT INTO test VALUES(4)')
+
+    db_utils.execute_file(dsn, tmpfile, pre_code='CREATE TABLE test (id INT)')
+
+    assert temp_db_cursor.row_set('SELECT * FROM test') == {(4, )}
+
+
+def test_execute_file_with_post_code(dsn, tmp_path, temp_db_cursor):
+    tmpfile = tmp_path / 'test.sql'
+    tmpfile.write_text('CREATE TABLE test (id INT)')
+
+    db_utils.execute_file(dsn, tmpfile, post_code='INSERT INTO test VALUES(23)')
+
+    assert temp_db_cursor.row_set('SELECT * FROM test') == {(23, )}