test/python/db/test_sql_preprocessor.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 SQL preprocessing.
   9 """
  10 import pytest
  11 import pytest_asyncio  # noqa
  12
  13 from nominatim_db.db.sql_preprocessor import SQLPreprocessor
  14
  15
  16 @pytest.fixture
  17 def sql_factory(tmp_path):
  18     def _mk_sql(sql_body):
  19         (tmp_path / 'test.sql').write_text("""
  20           CREATE OR REPLACE FUNCTION test() RETURNS TEXT
  21           AS $$
  22           BEGIN
  23             {}
  24           END;
  25           $$ LANGUAGE plpgsql IMMUTABLE;""".format(sql_body))
  26         return 'test.sql'
  27
  28     return _mk_sql
  29
  30
  31 @pytest.mark.parametrize("expr,ret", [
  32     ("'a'", 'a'),
  33     ("'{{db.partitions|join}}'", '012'),
  34     ("{% if 'country_name' in db.tables %}'yes'{% else %}'no'{% endif %}", "yes"),
  35     ("{% if 'xxx' in db.tables %}'yes'{% else %}'no'{% endif %}", "no"),
  36     ("'{{db.tablespace.address_data}}'", ""),
  37     ("'{{db.tablespace.search_data}}'", 'TABLESPACE "dsearch"'),
  38     ("'{{db.tablespace.address_index}}'", 'TABLESPACE "iaddress"'),
  39     ("'{{db.tablespace.aux_data}}'", 'TABLESPACE "daux"')
  40     ])
  41 def test_load_file_simple(sql_preprocessor_cfg, sql_factory,
  42                           temp_db_conn, temp_db_cursor, monkeypatch,
  43                           expr, ret):
  44     monkeypatch.setenv('NOMINATIM_TABLESPACE_SEARCH_DATA', 'dsearch')
  45     monkeypatch.setenv('NOMINATIM_TABLESPACE_ADDRESS_INDEX', 'iaddress')
  46     monkeypatch.setenv('NOMINATIM_TABLESPACE_AUX_DATA', 'daux')
  47     sqlfile = sql_factory("RETURN {};".format(expr))
  48
  49     SQLPreprocessor(temp_db_conn, sql_preprocessor_cfg).run_sql_file(temp_db_conn, sqlfile)
  50
  51     assert temp_db_cursor.scalar('SELECT test()') == ret
  52
  53
  54 def test_load_file_with_params(sql_preprocessor, sql_factory, temp_db_conn, temp_db_cursor):
  55     sqlfile = sql_factory("RETURN '{{ foo }} {{ bar }}';")
  56
  57     sql_preprocessor.run_sql_file(temp_db_conn, sqlfile, bar='XX', foo='ZZ')
  58
  59     assert temp_db_cursor.scalar('SELECT test()') == 'ZZ XX'
  60
  61
  62 @pytest.mark.asyncio
  63 async def test_load_parallel_file(dsn, sql_preprocessor, tmp_path, temp_db_cursor):
  64     (tmp_path / 'test.sql').write_text("""
  65         CREATE TABLE foo (a TEXT);
  66         CREATE TABLE foo2(a TEXT);""" + "\n---\nCREATE TABLE bar (b INT);")
  67
  68     await sql_preprocessor.run_parallel_sql_file(dsn, 'test.sql', num_threads=4)
  69
  70     assert temp_db_cursor.table_exists('foo')
  71     assert temp_db_cursor.table_exists('foo2')
  72     assert temp_db_cursor.table_exists('bar')