test/python/test_tools_import_special_phrases.py

   1 """
   2     Tests for import special phrases functions
   3 """
   4 import pytest
   5 from nominatim.tools.special_phrases import _create_place_classtype_indexes, _create_place_classtype_table, _get_wiki_content, _grant_access_to_webuser, _process_amenity
   6
   7 def test_get_wiki_content():
   8     assert _get_wiki_content('fr')
   9
  10 def execute_and_verify_add_word(temp_db_conn, phrase_label, phrase_class, phrase_type):
  11     _process_amenity(temp_db_conn, phrase_label, phrase_class, phrase_type, '')
  12
  13     with temp_db_conn.cursor() as temp_db_cursor:
  14         temp_db_cursor.execute(f"""
  15             SELECT * FROM word
  16             WHERE word_token=' {phrase_label}'
  17             AND word='{phrase_label}'
  18             AND class='{phrase_class}'
  19             AND type='{phrase_type}'""")
  20         return temp_db_cursor.fetchone()
  21
  22 def execute_and_verify_add_word_with_operator(temp_db_conn, phrase_label, phrase_class, phrase_type, phrase_operator):
  23     _process_amenity(temp_db_conn, phrase_label, phrase_class, phrase_type, phrase_operator)
  24
  25     with temp_db_conn.cursor() as temp_db_cursor:
  26         temp_db_cursor.execute(f"""
  27             SELECT * FROM word
  28             WHERE word_token=' {phrase_label}'
  29             AND word='{phrase_label}'
  30             AND class='{phrase_class}'
  31             AND type='{phrase_type}'
  32             AND operator='{phrase_operator}'""")
  33         return temp_db_cursor.fetchone()
  34
  35 def test_process_amenity_with_near_operator(temp_db_conn, word_table, amenity_operator_funcs):
  36     phrase_label = 'label'
  37     phrase_class = 'class'
  38     phrase_type = 'type'
  39
  40     assert execute_and_verify_add_word(temp_db_conn, phrase_label, phrase_class, phrase_type)
  41     assert execute_and_verify_add_word_with_operator(temp_db_conn, phrase_label, phrase_class, phrase_type, 'near')
  42     assert execute_and_verify_add_word_with_operator(temp_db_conn, phrase_label, phrase_class, phrase_type, 'in')
  43
  44 def index_exists(db_connect, index):
  45         """ Check that an index with the given name exists in the database.
  46         """
  47         with db_connect.cursor() as cur:
  48             cur.execute("""SELECT tablename FROM pg_indexes
  49                            WHERE indexname = %s and schemaname = 'public'""", (index, ))
  50             if cur.rowcount == 0:
  51                 return False
  52         return True
  53
  54 def test_create_place_classtype_indexes(temp_db_conn):
  55     phrase_class = 'class'
  56     phrase_type = 'type'
  57     table_name = f'place_classtype_{phrase_class}_{phrase_type}'
  58
  59     with temp_db_conn.cursor() as temp_db_cursor:
  60         temp_db_cursor.execute("CREATE EXTENSION postgis;")
  61         temp_db_cursor.execute(f'CREATE TABLE {table_name}(place_id BIGINT, centroid GEOMETRY)')
  62
  63     _create_place_classtype_indexes(temp_db_conn, '', phrase_class, phrase_type)
  64
  65     centroid_index_exists = index_exists(temp_db_conn, f'idx_place_classtype_{phrase_class}_{phrase_type}_centroid')
  66     place_id_index_exists = index_exists(temp_db_conn, f'idx_place_classtype_{phrase_class}_{phrase_type}_place_id')
  67
  68     assert centroid_index_exists and place_id_index_exists
  69
  70 def test_create_place_classtype_table(temp_db_conn, placex_table):
  71     phrase_class = 'class'
  72     phrase_type = 'type'
  73     _create_place_classtype_table(temp_db_conn, '', phrase_class, phrase_type)
  74
  75     with temp_db_conn.cursor() as temp_db_cursor:
  76         temp_db_cursor.execute(f"""
  77             SELECT *
  78             FROM information_schema.tables
  79             WHERE table_type='BASE TABLE'
  80             AND table_name='place_classtype_{phrase_class}_{phrase_type}'""")
  81         result = temp_db_cursor.fetchone()
  82     assert result
  83
  84 def test_grant_access_to_web_user(temp_db_conn, def_config):
  85     phrase_class = 'class'
  86     phrase_type = 'type'
  87     table_name = f'place_classtype_{phrase_class}_{phrase_type}'
  88
  89     with temp_db_conn.cursor() as temp_db_cursor:
  90         temp_db_cursor.execute(f'CREATE TABLE {table_name}()')
  91
  92     _grant_access_to_webuser(temp_db_conn, def_config, phrase_class, phrase_type)
  93
  94     with temp_db_conn.cursor() as temp_db_cursor:
  95         temp_db_cursor.execute(f"""
  96                 SELECT * FROM information_schema.role_table_grants
  97                 WHERE table_name='{table_name}'
  98                 AND grantee='{def_config.DATABASE_WEBUSER}'
  99                 AND privilege_type='SELECT'""")
 100         result = temp_db_cursor.fetchone()
 101     assert result
 102
 103 @pytest.fixture
 104 def amenity_operator_funcs(temp_db_cursor):
 105     temp_db_cursor.execute(f"""
 106         CREATE OR REPLACE FUNCTION make_standard_name(name TEXT) RETURNS TEXT
 107         AS $$
 108         DECLARE
 109         o TEXT;
 110         BEGIN
 111         RETURN name; --Basically return the same name for the tests
 112         END;
 113         $$
 114         LANGUAGE plpgsql IMMUTABLE;
 115
 116         CREATE SEQUENCE seq_word start 1;
 117
 118         CREATE OR REPLACE FUNCTION getorcreate_amenity(lookup_word TEXT,
 119                                                     lookup_class text, lookup_type text)
 120         RETURNS INTEGER
 121         AS $$
 122         DECLARE
 123         lookup_token TEXT;
 124         return_word_id INTEGER;
 125         BEGIN
 126         lookup_token := ' '||trim(lookup_word);
 127         SELECT min(word_id) FROM word
 128         WHERE word_token = lookup_token and word = lookup_word
 129                 and class = lookup_class and type = lookup_type
 130         INTO return_word_id;
 131         IF return_word_id IS NULL THEN
 132             return_word_id := nextval('seq_word');
 133             INSERT INTO word VALUES (return_word_id, lookup_token, lookup_word,
 134                                     lookup_class, lookup_type, null, 0);
 135         END IF;
 136         RETURN return_word_id;
 137         END;
 138         $$
 139         LANGUAGE plpgsql;
 140
 141         CREATE OR REPLACE FUNCTION getorcreate_amenityoperator(lookup_word TEXT,
 142                                                         lookup_class text,
 143                                                         lookup_type text,
 144                                                         op text)
 145         RETURNS INTEGER
 146         AS $$
 147         DECLARE
 148         lookup_token TEXT;
 149         return_word_id INTEGER;
 150         BEGIN
 151         lookup_token := ' '||trim(lookup_word);
 152         SELECT min(word_id) FROM word
 153         WHERE word_token = lookup_token and word = lookup_word
 154                 and class = lookup_class and type = lookup_type and operator = op
 155         INTO return_word_id;
 156         IF return_word_id IS NULL THEN
 157             return_word_id := nextval('seq_word');
 158             INSERT INTO word VALUES (return_word_id, lookup_token, lookup_word,
 159                                     lookup_class, lookup_type, null, 0, op);
 160         END IF;
 161         RETURN return_word_id;
 162         END;
 163         $$
 164         LANGUAGE plpgsql;""")