test/python/test_tools_refresh_create_functions.py

   1 """
   2 Tests for creating PL/pgSQL functions for Nominatim.
   3 """
   4 import pytest
   5
   6 from nominatim.tools.refresh import create_functions
   7
   8 @pytest.fixture
   9 def sql_tmp_path(tmp_path, def_config):
  10     def_config.lib_dir.sql = tmp_path
  11     return tmp_path
  12
  13 @pytest.fixture
  14 def conn(temp_db_conn, table_factory, monkeypatch):
  15     monkeypatch.setenv('NOMINATIM_DATABASE_MODULE_PATH', '.')
  16     table_factory('country_name', 'partition INT', (0, 1, 2))
  17     return temp_db_conn
  18
  19
  20 def test_create_functions(temp_db_cursor, conn, def_config, sql_tmp_path):
  21     sqlfile = sql_tmp_path / 'functions.sql'
  22     sqlfile.write_text("""CREATE OR REPLACE FUNCTION test() RETURNS INTEGER
  23                           AS $$
  24                           BEGIN
  25                             RETURN 43;
  26                           END;
  27                           $$ LANGUAGE plpgsql IMMUTABLE;
  28                        """)
  29
  30     create_functions(conn, def_config, sql_tmp_path)
  31
  32     assert temp_db_cursor.scalar('SELECT test()') == 43
  33
  34
  35 @pytest.mark.parametrize("dbg,ret", ((True, 43), (False, 22)))
  36 def test_create_functions_with_template(temp_db_cursor, conn, def_config, sql_tmp_path, dbg, ret):
  37     sqlfile = sql_tmp_path / 'functions.sql'
  38     sqlfile.write_text("""CREATE OR REPLACE FUNCTION test() RETURNS INTEGER
  39                           AS $$
  40                           BEGIN
  41                             {% if debug %}
  42                             RETURN 43;
  43                             {% else %}
  44                             RETURN 22;
  45                             {% endif %}
  46                           END;
  47                           $$ LANGUAGE plpgsql IMMUTABLE;
  48                        """)
  49
  50     create_functions(conn, def_config, sql_tmp_path, enable_debug=dbg)
  51
  52     assert temp_db_cursor.scalar('SELECT test()') == ret