test/python/tools/test_import_special_phrases.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2022 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8     Tests for import special phrases methods
   9     of the class SPImporter.
  10 """
  11 from shutil import copyfile
  12 import pytest
  13 from nominatim.tools.special_phrases.sp_importer import SPImporter
  14 from nominatim.tools.special_phrases.sp_wiki_loader import SPWikiLoader
  15 from nominatim.tools.special_phrases.special_phrase import SpecialPhrase
  16 from nominatim.errors import UsageError
  17
  18 from cursor import CursorForTesting
  19
  20 @pytest.fixture
  21 def testfile_dir(src_dir):
  22     return src_dir / 'test' / 'testfiles'
  23
  24
  25 @pytest.fixture
  26 def sp_importer(temp_db_conn, def_config, monkeypatch):
  27     """
  28         Return an instance of SPImporter.
  29     """
  30     monkeypatch.setenv('NOMINATIM_LANGUAGES', 'en')
  31     loader = SPWikiLoader(def_config)
  32     return SPImporter(def_config, temp_db_conn, loader)
  33
  34
  35 @pytest.fixture
  36 def xml_wiki_content(src_dir):
  37     """
  38         return the content of the static xml test file.
  39     """
  40     xml_test_content = src_dir / 'test' / 'testdata' / 'special_phrases_test_content.txt'
  41     return xml_test_content.read_text()
  42
  43
  44 @pytest.fixture
  45 def default_phrases(table_factory):
  46     table_factory('place_classtype_testclasstypetable_to_delete')
  47     table_factory('place_classtype_testclasstypetable_to_keep')
  48
  49
  50 def test_fetch_existing_place_classtype_tables(sp_importer, table_factory):
  51     """
  52         Check for the fetch_existing_place_classtype_tables() method.
  53         It should return the table just created.
  54     """
  55     table_factory('place_classtype_testclasstypetable')
  56
  57     sp_importer._fetch_existing_place_classtype_tables()
  58     contained_table = sp_importer.table_phrases_to_delete.pop()
  59     assert contained_table == 'place_classtype_testclasstypetable'
  60
  61 def test_check_sanity_class(sp_importer):
  62     """
  63         Check for _check_sanity() method.
  64         If a wrong class or type is given, an UsageError should raise.
  65         If a good class and type are given, nothing special happens.
  66     """
  67
  68     assert not sp_importer._check_sanity(SpecialPhrase('en', '', 'type', ''))
  69     assert not sp_importer._check_sanity(SpecialPhrase('en', 'class', '', ''))
  70
  71     assert sp_importer._check_sanity(SpecialPhrase('en', 'class', 'type', ''))
  72
  73 def test_load_white_and_black_lists(sp_importer):
  74     """
  75         Test that _load_white_and_black_lists() well return
  76         black list and white list and that they are of dict type.
  77     """
  78     black_list, white_list = sp_importer._load_white_and_black_lists()
  79
  80     assert isinstance(black_list, dict) and isinstance(white_list, dict)
  81
  82
  83 def test_create_place_classtype_indexes(temp_db_with_extensions, temp_db_conn,
  84                                         table_factory, sp_importer):
  85     """
  86         Test that _create_place_classtype_indexes() create the
  87         place_id index and centroid index on the right place_class_type table.
  88     """
  89     phrase_class = 'class'
  90     phrase_type = 'type'
  91     table_name = 'place_classtype_{}_{}'.format(phrase_class, phrase_type)
  92
  93     table_factory(table_name, 'place_id BIGINT, centroid GEOMETRY')
  94
  95     sp_importer._create_place_classtype_indexes('', phrase_class, phrase_type)
  96
  97     assert check_placeid_and_centroid_indexes(temp_db_conn, phrase_class, phrase_type)
  98
  99 def test_create_place_classtype_table(temp_db_conn, placex_table, sp_importer):
 100     """
 101         Test that _create_place_classtype_table() create
 102         the right place_classtype table.
 103     """
 104     phrase_class = 'class'
 105     phrase_type = 'type'
 106     sp_importer._create_place_classtype_table('', phrase_class, phrase_type)
 107
 108     assert check_table_exist(temp_db_conn, phrase_class, phrase_type)
 109
 110 def test_grant_access_to_web_user(temp_db_conn, table_factory, def_config, sp_importer):
 111     """
 112         Test that _grant_access_to_webuser() give
 113         right access to the web user.
 114     """
 115     phrase_class = 'class'
 116     phrase_type = 'type'
 117     table_name = 'place_classtype_{}_{}'.format(phrase_class, phrase_type)
 118
 119     table_factory(table_name)
 120
 121     sp_importer._grant_access_to_webuser(phrase_class, phrase_type)
 122
 123     assert check_grant_access(temp_db_conn, def_config.DATABASE_WEBUSER, phrase_class, phrase_type)
 124
 125 def test_create_place_classtype_table_and_indexes(
 126         temp_db_conn, def_config, placex_table,
 127         sp_importer):
 128     """
 129         Test that _create_place_classtype_table_and_indexes()
 130         create the right place_classtype tables and place_id indexes
 131         and centroid indexes and grant access to the web user
 132         for the given set of pairs.
 133     """
 134     pairs = set([('class1', 'type1'), ('class2', 'type2')])
 135
 136     sp_importer._create_place_classtype_table_and_indexes(pairs)
 137
 138     for pair in pairs:
 139         assert check_table_exist(temp_db_conn, pair[0], pair[1])
 140         assert check_placeid_and_centroid_indexes(temp_db_conn, pair[0], pair[1])
 141         assert check_grant_access(temp_db_conn, def_config.DATABASE_WEBUSER, pair[0], pair[1])
 142
 143 def test_remove_non_existent_tables_from_db(sp_importer, default_phrases,
 144                                             temp_db_conn):
 145     """
 146         Check for the remove_non_existent_phrases_from_db() method.
 147
 148         It should removed entries from the word table which are contained
 149         in the words_phrases_to_delete set and not those also contained
 150         in the words_phrases_still_exist set.
 151
 152         place_classtype tables contained in table_phrases_to_delete should
 153         be deleted.
 154     """
 155     sp_importer.table_phrases_to_delete = {
 156         'place_classtype_testclasstypetable_to_delete'
 157     }
 158
 159     query_tables = """
 160         SELECT table_name
 161         FROM information_schema.tables
 162         WHERE table_schema='public'
 163         AND table_name like 'place_classtype_%';
 164     """
 165
 166     sp_importer._remove_non_existent_tables_from_db()
 167
 168     # Changes are not committed yet. Use temp_db_conn for checking results.
 169     with temp_db_conn.cursor(cursor_factory=CursorForTesting) as cur:
 170         assert cur.row_set(query_tables) \
 171                  == {('place_classtype_testclasstypetable_to_keep', )}
 172
 173
 174 @pytest.mark.parametrize("should_replace", [(True), (False)])
 175 def test_import_phrases(monkeypatch, temp_db_conn, def_config, sp_importer,
 176                         placex_table, table_factory, tokenizer_mock,
 177                         xml_wiki_content, should_replace):
 178     """
 179         Check that the main import_phrases() method is well executed.
 180         It should create the place_classtype table, the place_id and centroid indexes,
 181         grand access to the web user and executing the SQL functions for amenities.
 182         It should also update the database well by deleting or preserving existing entries
 183         of the database.
 184     """
 185     #Add some data to the database before execution in order to test
 186     #what is deleted and what is preserved.
 187     table_factory('place_classtype_amenity_animal_shelter')
 188     table_factory('place_classtype_wrongclass_wrongtype')
 189
 190     monkeypatch.setattr('nominatim.tools.special_phrases.sp_wiki_loader.SPWikiLoader._get_wiki_content',
 191                         lambda self, lang: xml_wiki_content)
 192
 193     tokenizer = tokenizer_mock()
 194     sp_importer.import_phrases(tokenizer, should_replace)
 195
 196     assert len(tokenizer.analyser_cache['special_phrases']) == 18
 197
 198     class_test = 'aerialway'
 199     type_test = 'zip_line'
 200
 201     assert check_table_exist(temp_db_conn, class_test, type_test)
 202     assert check_placeid_and_centroid_indexes(temp_db_conn, class_test, type_test)
 203     assert check_grant_access(temp_db_conn, def_config.DATABASE_WEBUSER, class_test, type_test)
 204     assert check_table_exist(temp_db_conn, 'amenity', 'animal_shelter')
 205     if should_replace:
 206         assert not check_table_exist(temp_db_conn, 'wrong_class', 'wrong_type')
 207
 208     assert temp_db_conn.table_exists('place_classtype_amenity_animal_shelter')
 209     if should_replace:
 210         assert not temp_db_conn.table_exists('place_classtype_wrongclass_wrongtype')
 211
 212 def check_table_exist(temp_db_conn, phrase_class, phrase_type):
 213     """
 214         Verify that the place_classtype table exists for the given
 215         phrase_class and phrase_type.
 216     """
 217     return temp_db_conn.table_exists('place_classtype_{}_{}'.format(phrase_class, phrase_type))
 218
 219
 220 def check_grant_access(temp_db_conn, user, phrase_class, phrase_type):
 221     """
 222         Check that the web user has been granted right access to the
 223         place_classtype table of the given phrase_class and phrase_type.
 224     """
 225     table_name = 'place_classtype_{}_{}'.format(phrase_class, phrase_type)
 226
 227     with temp_db_conn.cursor() as temp_db_cursor:
 228         temp_db_cursor.execute("""
 229                 SELECT * FROM information_schema.role_table_grants
 230                 WHERE table_name='{}'
 231                 AND grantee='{}'
 232                 AND privilege_type='SELECT'""".format(table_name, user))
 233         return temp_db_cursor.fetchone()
 234
 235 def check_placeid_and_centroid_indexes(temp_db_conn, phrase_class, phrase_type):
 236     """
 237         Check that the place_id index and centroid index exist for the
 238         place_classtype table of the given phrase_class and phrase_type.
 239     """
 240     index_prefix = 'idx_place_classtype_{}_{}_'.format(phrase_class, phrase_type)
 241
 242     return (
 243         temp_db_conn.index_exists(index_prefix + 'centroid')
 244         and
 245         temp_db_conn.index_exists(index_prefix + 'place_id')
 246     )