test/python/test_tools_import_special_phrases.py

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