test/python/test_db_sql_preprocessor.py

   1 """
   2 Tests for SQL preprocessing.
   3 """
   4 import pytest
   5
   6 from nominatim.db.sql_preprocessor import SQLPreprocessor
   7
   8 @pytest.fixture
   9 def sql_factory(tmp_path):
  10     def _mk_sql(sql_body):
  11         (tmp_path / 'test.sql').write_text("""
  12           CREATE OR REPLACE FUNCTION test() RETURNS TEXT
  13           AS $$
  14           BEGIN
  15             {}
  16           END;
  17           $$ LANGUAGE plpgsql IMMUTABLE;""".format(sql_body))
  18         return 'test.sql'
  19
  20     return _mk_sql
  21
  22 @pytest.mark.parametrize("expr,ret", [
  23     ("'a'", 'a'),
  24     ("'{{db.partitions|join}}'", '012'),
  25     ("{% if 'country_name' in db.tables %}'yes'{% else %}'no'{% endif %}", "yes"),
  26     ("{% if 'xxx' in db.tables %}'yes'{% else %}'no'{% endif %}", "no"),
  27     ("'{{db.tablespace.address_data}}'", ""),
  28     ("'{{db.tablespace.search_data}}'", 'TABLESPACE "dsearch"'),
  29     ("'{{db.tablespace.address_index}}'", 'TABLESPACE "iaddress"'),
  30     ("'{{db.tablespace.aux_data}}'", 'TABLESPACE "daux"')
  31     ])
  32 def test_load_file_simple(sql_preprocessor_cfg, sql_factory,
  33                           temp_db_conn, temp_db_cursor, monkeypatch,
  34                           expr, ret):
  35     monkeypatch.setenv('NOMINATIM_TABLESPACE_SEARCH_DATA', 'dsearch')
  36     monkeypatch.setenv('NOMINATIM_TABLESPACE_ADDRESS_INDEX', 'iaddress')
  37     monkeypatch.setenv('NOMINATIM_TABLESPACE_AUX_DATA', 'daux')
  38     sqlfile = sql_factory("RETURN {};".format(expr))
  39
  40     SQLPreprocessor(temp_db_conn, sql_preprocessor_cfg).run_sql_file(temp_db_conn, sqlfile)
  41
  42     assert temp_db_cursor.scalar('SELECT test()') == ret
  43
  44
  45 def test_load_file_with_params(sql_preprocessor, sql_factory, temp_db_conn, temp_db_cursor):
  46     sqlfile = sql_factory("RETURN '{{ foo }} {{ bar }}';")
  47
  48     sql_preprocessor.run_sql_file(temp_db_conn, sqlfile, bar='XX', foo='ZZ')
  49
  50     assert temp_db_cursor.scalar('SELECT test()') == 'ZZ XX'