test/python/test_tools_import_special_phrases.py

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