+ assert check_grant_access(temp_db_conn, def_config.DATABASE_WEBUSER, phrase_class, phrase_type)
+
+def test_create_place_classtype_table_and_indexes(
+ temp_db_conn, def_config, placex_table,
+ sp_importer):
+ """
+ Test that _create_place_classtype_table_and_indexes()
+ create the right place_classtype tables and place_id indexes
+ and centroid indexes and grant access to the web user
+ for the given set of pairs.
+ """
+ pairs = set([('class1', 'type1'), ('class2', 'type2')])
+
+ sp_importer._create_place_classtype_table_and_indexes(pairs)
+
+ for pair in pairs:
+ assert check_table_exist(temp_db_conn, pair[0], pair[1])
+ assert check_placeid_and_centroid_indexes(temp_db_conn, pair[0], pair[1])
+ assert check_grant_access(temp_db_conn, def_config.DATABASE_WEBUSER, pair[0], pair[1])
+
+def test_remove_non_existent_tables_from_db(sp_importer, default_phrases,
+ temp_db_conn):
+ """
+ Check for the remove_non_existent_phrases_from_db() method.
+
+ It should removed entries from the word table which are contained
+ in the words_phrases_to_delete set and not those also contained
+ in the words_phrases_still_exist set.
+
+ place_classtype tables contained in table_phrases_to_delete should
+ be deleted.
+ """
+ sp_importer.table_phrases_to_delete = {
+ 'place_classtype_testclasstypetable_to_delete'
+ }
+
+ query_tables = """
+ SELECT table_name
+ FROM information_schema.tables
+ WHERE table_schema='public'
+ AND table_name like 'place_classtype_%';
+ """
+
+ sp_importer._remove_non_existent_tables_from_db()
+
+ # Changes are not committed yet. Use temp_db_conn for checking results.
+ with temp_db_conn.cursor(cursor_factory=CursorForTesting) as cur:
+ assert cur.row_set(query_tables) \
+ == {('place_classtype_testclasstypetable_to_keep', )}
+
+
+@pytest.mark.parametrize("should_replace", [(True), (False)])
+def test_import_phrases(monkeypatch, temp_db_conn, def_config, sp_importer,
+ placex_table, table_factory, tokenizer_mock,
+ xml_wiki_content, should_replace):
+ """
+ Check that the main import_phrases() method is well executed.
+ It should create the place_classtype table, the place_id and centroid indexes,
+ grand access to the web user and executing the SQL functions for amenities.
+ It should also update the database well by deleting or preserving existing entries
+ of the database.
+ """
+ #Add some data to the database before execution in order to test
+ #what is deleted and what is preserved.
+ table_factory('place_classtype_amenity_animal_shelter')
+ table_factory('place_classtype_wrongclass_wrongtype')
+
+ monkeypatch.setattr('nominatim.tools.special_phrases.sp_wiki_loader.SPWikiLoader._get_wiki_content',
+ lambda self, lang: xml_wiki_content)
+
+ tokenizer = tokenizer_mock()
+ sp_importer.import_phrases(tokenizer, should_replace)
+
+ assert len(tokenizer.analyser_cache['special_phrases']) == 18
+
+ class_test = 'aerialway'
+ type_test = 'zip_line'
+
+ assert check_table_exist(temp_db_conn, class_test, type_test)
+ assert check_placeid_and_centroid_indexes(temp_db_conn, class_test, type_test)
+ assert check_grant_access(temp_db_conn, def_config.DATABASE_WEBUSER, class_test, type_test)
+ assert check_table_exist(temp_db_conn, 'amenity', 'animal_shelter')
+ if should_replace:
+ assert not check_table_exist(temp_db_conn, 'wrong_class', 'wrong_type')
+
+ assert temp_db_conn.table_exists('place_classtype_amenity_animal_shelter')
+ if should_replace:
+ assert not temp_db_conn.table_exists('place_classtype_wrongclass_wrongtype')
+
+def check_table_exist(temp_db_conn, phrase_class, phrase_type):
+ """
+ Verify that the place_classtype table exists for the given
+ phrase_class and phrase_type.
+ """
+ return temp_db_conn.table_exists('place_classtype_{}_{}'.format(phrase_class, phrase_type))
+
+
+def check_grant_access(temp_db_conn, user, phrase_class, phrase_type):
+ """
+ Check that the web user has been granted right access to the
+ place_classtype table of the given phrase_class and phrase_type.
+ """
+ table_name = 'place_classtype_{}_{}'.format(phrase_class, phrase_type)