enable flake for Python tests

[nominatim.git] / test / python / db / test_sql_preprocessor.py

diff --git a/test/python/db/test_sql_preprocessor.py b/test/python/db/test_sql_preprocessor.py
index 2ca0c66112e8b8f094a2045e681cb3fa6bc10eae..f2fbbb2af52cad7ead0c62ece5412bdc8b5ddae0 100644 (file)
--- a/test/python/db/test_sql_preprocessor.py
+++ b/test/python/db/test_sql_preprocessor.py
@@ -1,9 +1,17 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2025 by the Nominatim developer community.
+# For a full list of authors see the git log.

 """
 Tests for SQL preprocessing.
 """
 import pytest
 """
 Tests for SQL preprocessing.
 """
 import pytest

+import pytest_asyncio  # noqa
+
+from nominatim_db.db.sql_preprocessor import SQLPreprocessor

 
 

-from nominatim.db.sql_preprocessor import SQLPreprocessor

 
 @pytest.fixture
 def sql_factory(tmp_path):
 
 @pytest.fixture
 def sql_factory(tmp_path):


@@ -19,6 +27,7 @@ def sql_factory(tmp_path):
 
     return _mk_sql
 
 
     return _mk_sql
 

+

 @pytest.mark.parametrize("expr,ret", [
     ("'a'", 'a'),
     ("'{{db.partitions|join}}'", '012'),
 @pytest.mark.parametrize("expr,ret", [
     ("'a'", 'a'),
     ("'{{db.partitions|join}}'", '012'),


@@ -48,3 +57,16 @@ def test_load_file_with_params(sql_preprocessor, sql_factory, temp_db_conn, temp
     sql_preprocessor.run_sql_file(temp_db_conn, sqlfile, bar='XX', foo='ZZ')
 
     assert temp_db_cursor.scalar('SELECT test()') == 'ZZ XX'
     sql_preprocessor.run_sql_file(temp_db_conn, sqlfile, bar='XX', foo='ZZ')
 
     assert temp_db_cursor.scalar('SELECT test()') == 'ZZ XX'

+
+
+@pytest.mark.asyncio
+async def test_load_parallel_file(dsn, sql_preprocessor, tmp_path, temp_db_cursor):
+    (tmp_path / 'test.sql').write_text("""
+        CREATE TABLE foo (a TEXT);
+        CREATE TABLE foo2(a TEXT);""" + "\n---\nCREATE TABLE bar (b INT);")
+
+    await sql_preprocessor.run_parallel_sql_file(dsn, 'test.sql', num_threads=4)
+
+    assert temp_db_cursor.table_exists('foo')
+    assert temp_db_cursor.table_exists('foo2')
+    assert temp_db_cursor.table_exists('bar')