test/python/tokenizer/test_legacy.py

   1 """
   2 Test for legacy tokenizer.
   3 """
   4 import shutil
   5 import re
   6
   7 import pytest
   8
   9 from nominatim.indexer.place_info import PlaceInfo
  10 from nominatim.tokenizer import legacy_tokenizer
  11 from nominatim.db import properties
  12 from nominatim.errors import UsageError
  13
  14 from mock_legacy_word_table import MockLegacyWordTable
  15
  16 # Force use of legacy word table
  17 @pytest.fixture
  18 def word_table(temp_db_conn):
  19     return MockLegacyWordTable(temp_db_conn)
  20
  21
  22 @pytest.fixture
  23 def test_config(project_env, tmp_path):
  24     module_dir = tmp_path / 'module_src'
  25     module_dir.mkdir()
  26     (module_dir / 'nominatim.so').write_text('TEST nomiantim.so')
  27
  28     project_env.lib_dir.module = module_dir
  29
  30     sqldir = tmp_path / 'sql'
  31     sqldir.mkdir()
  32     (sqldir / 'tokenizer').mkdir()
  33
  34     # Get the original SQL but replace make_standard_name to avoid module use.
  35     init_sql = (project_env.lib_dir.sql / 'tokenizer' / 'legacy_tokenizer.sql').read_text()
  36     for fn in ('transliteration', 'gettokenstring'):
  37         init_sql = re.sub(f'CREATE OR REPLACE FUNCTION {fn}[^;]*;',
  38                           '', init_sql, re.DOTALL)
  39     init_sql += """
  40                    CREATE OR REPLACE FUNCTION make_standard_name(name TEXT)
  41                    RETURNS TEXT AS $$ SELECT lower(name); $$ LANGUAGE SQL;
  42
  43                 """
  44     # Also load util functions. Some are needed by the tokenizer.
  45     init_sql += (project_env.lib_dir.sql / 'functions' / 'utils.sql').read_text()
  46     (sqldir / 'tokenizer' / 'legacy_tokenizer.sql').write_text(init_sql)
  47
  48     (sqldir / 'words.sql').write_text("SELECT 'a'")
  49
  50     shutil.copy(str(project_env.lib_dir.sql / 'tokenizer' / 'legacy_tokenizer_tables.sql'),
  51                 str(sqldir / 'tokenizer' / 'legacy_tokenizer_tables.sql'))
  52
  53     project_env.lib_dir.sql = sqldir
  54     project_env.lib_dir.data = sqldir
  55
  56     return project_env
  57
  58
  59 @pytest.fixture
  60 def tokenizer_factory(dsn, tmp_path, property_table):
  61     (tmp_path / 'tokenizer').mkdir()
  62
  63     def _maker():
  64         return legacy_tokenizer.create(dsn, tmp_path / 'tokenizer')
  65
  66     return _maker
  67
  68
  69 @pytest.fixture
  70 def tokenizer_setup(tokenizer_factory, test_config, monkeypatch, sql_preprocessor):
  71     monkeypatch.setattr(legacy_tokenizer, '_check_module', lambda m, c: None)
  72     tok = tokenizer_factory()
  73     tok.init_new_db(test_config)
  74
  75
  76 @pytest.fixture
  77 def analyzer(tokenizer_factory, test_config, monkeypatch, sql_preprocessor,
  78              word_table, temp_db_with_extensions, tmp_path):
  79     monkeypatch.setattr(legacy_tokenizer, '_check_module', lambda m, c: None)
  80     monkeypatch.setenv('NOMINATIM_TERM_NORMALIZATION', ':: lower();')
  81     tok = tokenizer_factory()
  82     tok.init_new_db(test_config)
  83     monkeypatch.undo()
  84
  85     with tok.name_analyzer() as analyzer:
  86         yield analyzer
  87
  88
  89 @pytest.fixture
  90 def make_standard_name(temp_db_cursor):
  91     temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION make_standard_name(name TEXT)
  92                               RETURNS TEXT AS $$ SELECT '#' || lower(name) || '#'; $$ LANGUAGE SQL""")
  93
  94
  95 @pytest.fixture
  96 def create_postcode_id(temp_db_cursor):
  97     temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION create_postcode_id(postcode TEXT)
  98                               RETURNS BOOLEAN AS $$
  99                               INSERT INTO word (word_token, word, class, type)
 100                                 VALUES (' ' || postcode, postcode, 'place', 'postcode')
 101                               RETURNING True;
 102                               $$ LANGUAGE SQL""")
 103
 104
 105 def test_init_new(tokenizer_factory, test_config, monkeypatch,
 106                   temp_db_conn, sql_preprocessor):
 107     monkeypatch.setenv('NOMINATIM_TERM_NORMALIZATION', 'xxvv')
 108     monkeypatch.setattr(legacy_tokenizer, '_check_module', lambda m, c: None)
 109
 110     tok = tokenizer_factory()
 111     tok.init_new_db(test_config)
 112
 113     assert properties.get_property(temp_db_conn, legacy_tokenizer.DBCFG_NORMALIZATION) == 'xxvv'
 114
 115     outfile = test_config.project_dir / 'module' / 'nominatim.so'
 116
 117     assert outfile.exists()
 118     assert outfile.read_text() == 'TEST nomiantim.so'
 119     assert outfile.stat().st_mode == 33261
 120
 121
 122 def test_init_module_load_failed(tokenizer_factory, test_config):
 123     tok = tokenizer_factory()
 124
 125     with pytest.raises(UsageError):
 126         tok.init_new_db(test_config)
 127
 128
 129 def test_init_module_custom(tokenizer_factory, test_config,
 130                             monkeypatch, tmp_path, sql_preprocessor):
 131     module_dir = (tmp_path / 'custom').resolve()
 132     module_dir.mkdir()
 133     (module_dir/ 'nominatim.so').write_text('CUSTOM nomiantim.so')
 134
 135     monkeypatch.setenv('NOMINATIM_DATABASE_MODULE_PATH', str(module_dir))
 136     monkeypatch.setattr(legacy_tokenizer, '_check_module', lambda m, c: None)
 137
 138     tok = tokenizer_factory()
 139     tok.init_new_db(test_config)
 140
 141     assert not (test_config.project_dir / 'module').exists()
 142
 143
 144 def test_init_from_project(tokenizer_setup, tokenizer_factory, test_config):
 145     tok = tokenizer_factory()
 146
 147     tok.init_from_project(test_config)
 148
 149     assert tok.normalization is not None
 150
 151
 152 def test_update_sql_functions(sql_preprocessor, temp_db_conn,
 153                               tokenizer_factory, test_config, table_factory,
 154                               monkeypatch, temp_db_cursor):
 155     monkeypatch.setenv('NOMINATIM_MAX_WORD_FREQUENCY', '1133')
 156     monkeypatch.setattr(legacy_tokenizer, '_check_module', lambda m, c: None)
 157     tok = tokenizer_factory()
 158     tok.init_new_db(test_config)
 159     monkeypatch.undo()
 160
 161     assert properties.get_property(temp_db_conn, legacy_tokenizer.DBCFG_MAXWORDFREQ) == '1133'
 162
 163     table_factory('test', 'txt TEXT')
 164
 165     func_file = test_config.lib_dir.sql / 'tokenizer' / 'legacy_tokenizer.sql'
 166     func_file.write_text("""INSERT INTO test VALUES ('{{max_word_freq}}'),
 167                                                    ('{{modulepath}}')""")
 168
 169     tok.update_sql_functions(test_config)
 170
 171     test_content = temp_db_cursor.row_set('SELECT * FROM test')
 172     assert test_content == set((('1133', ), (str(test_config.project_dir / 'module'), )))
 173
 174
 175 def test_finalize_import(tokenizer_factory, temp_db_conn,
 176                          temp_db_cursor, test_config, monkeypatch,
 177                          sql_preprocessor_cfg):
 178     monkeypatch.setattr(legacy_tokenizer, '_check_module', lambda m, c: None)
 179
 180     func_file = test_config.lib_dir.sql / 'tokenizer' / 'legacy_tokenizer_indices.sql'
 181     func_file.write_text("""CREATE FUNCTION test() RETURNS TEXT
 182                             AS $$ SELECT 'b' $$ LANGUAGE SQL""")
 183
 184     tok = tokenizer_factory()
 185     tok.init_new_db(test_config)
 186
 187     tok.finalize_import(test_config)
 188
 189     temp_db_cursor.scalar('SELECT test()') == 'b'
 190
 191
 192 def test_migrate_database(tokenizer_factory, test_config, temp_db_conn, monkeypatch):
 193     monkeypatch.setattr(legacy_tokenizer, '_check_module', lambda m, c: None)
 194     tok = tokenizer_factory()
 195     tok.migrate_database(test_config)
 196
 197     assert properties.get_property(temp_db_conn, legacy_tokenizer.DBCFG_MAXWORDFREQ) is not None
 198     assert properties.get_property(temp_db_conn, legacy_tokenizer.DBCFG_NORMALIZATION) is not None
 199
 200     outfile = test_config.project_dir / 'module' / 'nominatim.so'
 201
 202     assert outfile.exists()
 203     assert outfile.read_text() == 'TEST nomiantim.so'
 204     assert outfile.stat().st_mode == 33261
 205
 206
 207 def test_check_database(test_config, tokenizer_factory, monkeypatch,
 208                         temp_db_cursor, sql_preprocessor_cfg):
 209     monkeypatch.setattr(legacy_tokenizer, '_check_module', lambda m, c: None)
 210     tok = tokenizer_factory()
 211     tok.init_new_db(test_config)
 212
 213     assert tok.check_database(False) is None
 214
 215
 216 def test_check_database_no_tokenizer(test_config, tokenizer_factory):
 217     tok = tokenizer_factory()
 218
 219     assert tok.check_database(False) is not None
 220
 221
 222 def test_check_database_bad_setup(test_config, tokenizer_factory, monkeypatch,
 223                                   temp_db_cursor, sql_preprocessor_cfg):
 224     monkeypatch.setattr(legacy_tokenizer, '_check_module', lambda m, c: None)
 225     tok = tokenizer_factory()
 226     tok.init_new_db(test_config)
 227
 228     # Inject a bad transliteration.
 229     temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION make_standard_name(name TEXT)
 230                               RETURNS TEXT AS $$ SELECT 'garbage'; $$ LANGUAGE SQL""")
 231
 232     assert tok.check_database(False) is not None
 233
 234
 235 def test_update_statistics_reverse_only(word_table, tokenizer_factory):
 236     tok = tokenizer_factory()
 237     tok.update_statistics()
 238
 239
 240 def test_update_statistics(word_table, table_factory, temp_db_cursor, tokenizer_factory):
 241     word_table.add_full_word(1000, 'hello')
 242     table_factory('search_name',
 243                   'place_id BIGINT, name_vector INT[]',
 244                   [(12, [1000])])
 245     tok = tokenizer_factory()
 246
 247     tok.update_statistics()
 248
 249     assert temp_db_cursor.scalar("""SELECT count(*) FROM word
 250                                     WHERE word_token like ' %' and
 251                                           search_name_count > 0""") > 0
 252
 253
 254 def test_normalize(analyzer):
 255     assert analyzer.normalize('TEsT') == 'test'
 256
 257
 258 def test_update_postcodes_from_db_empty(analyzer, table_factory, word_table,
 259                                         create_postcode_id):
 260     table_factory('location_postcode', 'postcode TEXT',
 261                   content=(('1234',), ('12 34',), ('AB23',), ('1234',)))
 262
 263     analyzer.update_postcodes_from_db()
 264
 265     assert word_table.get_postcodes() == {'1234', '12 34', 'AB23'}
 266
 267
 268 def test_update_postcodes_from_db_add_and_remove(analyzer, table_factory, word_table,
 269                                                  create_postcode_id):
 270     table_factory('location_postcode', 'postcode TEXT',
 271                   content=(('1234',), ('45BC', ), ('XX45', )))
 272     word_table.add_postcode(' 1234', '1234')
 273     word_table.add_postcode(' 5678', '5678')
 274
 275     analyzer.update_postcodes_from_db()
 276
 277     assert word_table.get_postcodes() == {'1234', '45BC', 'XX45'}
 278
 279
 280 def test_update_special_phrase_empty_table(analyzer, word_table, make_standard_name):
 281     analyzer.update_special_phrases([
 282         ("König bei", "amenity", "royal", "near"),
 283         ("Könige", "amenity", "royal", "-"),
 284         ("könige", "amenity", "royal", "-"),
 285         ("strasse", "highway", "primary", "in")
 286     ], True)
 287
 288     assert word_table.get_special() \
 289                == set(((' #könig bei#', 'könig bei', 'amenity', 'royal', 'near'),
 290                        (' #könige#', 'könige', 'amenity', 'royal', None),
 291                        (' #strasse#', 'strasse', 'highway', 'primary', 'in')))
 292
 293
 294 def test_update_special_phrase_delete_all(analyzer, word_table, make_standard_name):
 295     word_table.add_special(' #foo#', 'foo', 'amenity', 'prison', 'in')
 296     word_table.add_special(' #bar#', 'bar', 'highway', 'road', None)
 297
 298     assert word_table.count_special() == 2
 299
 300     analyzer.update_special_phrases([], True)
 301
 302     assert word_table.count_special() == 0
 303
 304
 305 def test_update_special_phrases_no_replace(analyzer, word_table, make_standard_name):
 306     word_table.add_special(' #foo#', 'foo', 'amenity', 'prison', 'in')
 307     word_table.add_special(' #bar#', 'bar', 'highway', 'road', None)
 308
 309     assert word_table.count_special() == 2
 310
 311     analyzer.update_special_phrases([], False)
 312
 313     assert word_table.count_special() == 2
 314
 315
 316 def test_update_special_phrase_modify(analyzer, word_table, make_standard_name):
 317     word_table.add_special(' #foo#', 'foo', 'amenity', 'prison', 'in')
 318     word_table.add_special(' #bar#', 'bar', 'highway', 'road', None)
 319
 320     assert word_table.count_special() == 2
 321
 322     analyzer.update_special_phrases([
 323         ('prison', 'amenity', 'prison', 'in'),
 324         ('bar', 'highway', 'road', '-'),
 325         ('garden', 'leisure', 'garden', 'near')
 326     ], True)
 327
 328     assert word_table.get_special() \
 329                == set(((' #prison#', 'prison', 'amenity', 'prison', 'in'),
 330                        (' #bar#', 'bar', 'highway', 'road', None),
 331                        (' #garden#', 'garden', 'leisure', 'garden', 'near')))
 332
 333
 334 def test_add_country_names(analyzer, word_table, make_standard_name):
 335     analyzer.add_country_names('de', {'name': 'Germany',
 336                                       'name:de': 'Deutschland',
 337                                       'short_name': 'germany'})
 338
 339     assert word_table.get_country() \
 340                == {('de', ' #germany#'),
 341                    ('de', ' #deutschland#')}
 342
 343
 344 def test_add_more_country_names(analyzer, word_table, make_standard_name):
 345     word_table.add_country('fr', ' #france#')
 346     word_table.add_country('it', ' #italy#')
 347     word_table.add_country('it', ' #itala#')
 348
 349     analyzer.add_country_names('it', {'name': 'Italy', 'ref': 'IT'})
 350
 351     assert word_table.get_country() \
 352                == {('fr', ' #france#'),
 353                    ('it', ' #italy#'),
 354                    ('it', ' #itala#'),
 355                    ('it', ' #it#')}
 356
 357
 358 @pytest.mark.parametrize('pcode', ['12345', 'AB 123', '34-345'])
 359 def test_process_place_postcode(analyzer, create_postcode_id, word_table, pcode):
 360     analyzer.process_place(PlaceInfo({'address': {'postcode' : pcode}}))
 361
 362     assert word_table.get_postcodes() == {pcode, }
 363
 364
 365 @pytest.mark.parametrize('pcode', ['12:23', 'ab;cd;f', '123;836'])
 366 def test_process_place_bad_postcode(analyzer, create_postcode_id, word_table, pcode):
 367     analyzer.process_place(PlaceInfo({'address': {'postcode' : pcode}}))
 368
 369     assert not word_table.get_postcodes()
 370
 371
 372 class TestHousenumberName:
 373
 374     @staticmethod
 375     @pytest.fixture(autouse=True)
 376     def setup_create_housenumbers(temp_db_cursor):
 377         temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION create_housenumbers(
 378                                       housenumbers TEXT[],
 379                                       OUT tokens TEXT, OUT normtext TEXT)
 380                                   AS $$
 381                                   SELECT housenumbers::TEXT, array_to_string(housenumbers, ';')
 382                                   $$ LANGUAGE SQL""")
 383
 384
 385     @staticmethod
 386     @pytest.mark.parametrize('hnr', ['123a', '1', '101'])
 387     def test_process_place_housenumbers_simple(analyzer, hnr):
 388         info = analyzer.process_place(PlaceInfo({'address': {'housenumber' : hnr}}))
 389
 390         assert info['hnr'] == hnr
 391         assert info['hnr_tokens'].startswith("{")
 392
 393
 394     @staticmethod
 395     def test_process_place_housenumbers_lists(analyzer):
 396         info = analyzer.process_place(PlaceInfo({'address': {'conscriptionnumber' : '1; 2;3'}}))
 397
 398         assert set(info['hnr'].split(';')) == set(('1', '2', '3'))
 399
 400
 401     @staticmethod
 402     def test_process_place_housenumbers_duplicates(analyzer):
 403         info = analyzer.process_place(PlaceInfo({'address': {'housenumber' : '134',
 404                                                    'conscriptionnumber' : '134',
 405                                                    'streetnumber' : '99a'}}))
 406
 407         assert set(info['hnr'].split(';')) == set(('134', '99a'))
 408
 409
 410 class TestPlaceNames:
 411
 412     @pytest.fixture(autouse=True)
 413     def setup(self, analyzer):
 414         self.analyzer = analyzer
 415
 416
 417     def expect_name_terms(self, info, *expected_terms):
 418         tokens = self.analyzer.get_word_token_info(list(expected_terms))
 419         for token in tokens:
 420             assert token[2] is not None, "No token for {0}".format(token)
 421
 422         assert eval(info['names']) == set((t[2] for t in tokens)),\
 423                f"Expected: {tokens}\nGot: {info['names']}"
 424
 425
 426     def process_named_place(self, names):
 427         return self.analyzer.process_place(PlaceInfo({'name': names}))
 428
 429
 430     def test_simple_names(self):
 431         info = self.process_named_place({'name': 'Soft bAr', 'ref': '34'})
 432
 433         self.expect_name_terms(info, '#Soft bAr', '#34', 'Soft', 'bAr', '34')
 434
 435
 436     @pytest.mark.parametrize('sep', [',' , ';'])
 437     def test_names_with_separator(self, sep):
 438         info = self.process_named_place({'name': sep.join(('New York', 'Big Apple'))})
 439
 440         self.expect_name_terms(info, '#New York', '#Big Apple',
 441                                      'new', 'york', 'big', 'apple')
 442
 443
 444     def test_full_names_with_bracket(self):
 445         info = self.process_named_place({'name': 'Houseboat (left)'})
 446
 447         self.expect_name_terms(info, '#Houseboat (left)', '#Houseboat',
 448                                      'houseboat', '(left)')
 449
 450
 451     def test_country_name(self, word_table):
 452         place = PlaceInfo({'name' : {'name': 'Norge'},
 453                            'country_code': 'no',
 454                            'rank_address': 4,
 455                            'class': 'boundary',
 456                            'type': 'administrative'})
 457
 458         info = self.analyzer.process_place(place)
 459
 460         self.expect_name_terms(info, '#norge', 'norge')
 461         assert word_table.get_country() == {('no', ' norge')}
 462
 463
 464 class TestPlaceAddress:
 465
 466     @pytest.fixture(autouse=True)
 467     def setup(self, analyzer):
 468         self.analyzer = analyzer
 469
 470
 471     @pytest.fixture
 472     def getorcreate_hnr_id(self, temp_db_cursor):
 473         temp_db_cursor.execute("""CREATE SEQUENCE seq_hnr start 1;
 474                                   CREATE OR REPLACE FUNCTION getorcreate_housenumber_id(lookup_word TEXT)
 475                                   RETURNS INTEGER AS $$
 476                                   SELECT -nextval('seq_hnr')::INTEGER; $$ LANGUAGE SQL""")
 477
 478     def process_address(self, **kwargs):
 479         return self.analyzer.process_place(PlaceInfo({'address': kwargs}))
 480
 481
 482     def name_token_set(self, *expected_terms):
 483         tokens = self.analyzer.get_word_token_info(list(expected_terms))
 484         for token in tokens:
 485             assert token[2] is not None, "No token for {0}".format(token)
 486
 487         return set((t[2] for t in tokens))
 488
 489
 490     @pytest.mark.parametrize('pcode', ['12345', 'AB 123', '34-345'])
 491     def test_process_place_postcode(self, word_table, pcode):
 492         self.process_address(postcode=pcode)
 493
 494         assert word_table.get_postcodes() == {pcode, }
 495
 496
 497     @pytest.mark.parametrize('pcode', ['12:23', 'ab;cd;f', '123;836'])
 498     def test_process_place_bad_postcode(self, word_table, pcode):
 499         self.process_address(postcode=pcode)
 500
 501         assert not word_table.get_postcodes()
 502
 503
 504     @pytest.mark.parametrize('hnr', ['123a', '0', '101'])
 505     def test_process_place_housenumbers_simple(self, hnr, getorcreate_hnr_id):
 506         info = self.process_address(housenumber=hnr)
 507
 508         assert info['hnr'] == hnr.lower()
 509         assert info['hnr_tokens'] == "{-1}"
 510
 511
 512     def test_process_place_housenumbers_lists(self, getorcreate_hnr_id):
 513         info = self.process_address(conscriptionnumber='1; 2;3')
 514
 515         assert set(info['hnr'].split(';')) == set(('1', '2', '3'))
 516         assert info['hnr_tokens'] == "{-1,-2,-3}"
 517
 518
 519     def test_process_place_housenumbers_duplicates(self, getorcreate_hnr_id):
 520         info = self.process_address(housenumber='134',
 521                                     conscriptionnumber='134',
 522                                     streetnumber='99A')
 523
 524         assert set(info['hnr'].split(';')) == set(('134', '99a'))
 525         assert info['hnr_tokens'] == "{-1,-2}"
 526
 527
 528     def test_process_place_street(self):
 529         # legacy tokenizer only indexes known names
 530         self.analyzer.process_place(PlaceInfo({'name': {'name' : 'Grand Road'}}))
 531         info = self.process_address(street='Grand Road')
 532
 533         assert eval(info['street']) == self.name_token_set('#Grand Road')
 534
 535
 536     def test_process_place_street_empty(self):
 537         info = self.process_address(street='🜵')
 538
 539         assert 'street' not in info
 540
 541
 542     def test_process_place_place(self):
 543         self.analyzer.process_place(PlaceInfo({'name': {'name' : 'Honu Lulu'}}))
 544         info = self.process_address(place='Honu Lulu')
 545
 546         assert eval(info['place_search']) == self.name_token_set('#Honu Lulu',
 547                                                                  'Honu', 'Lulu')
 548         assert eval(info['place_match']) == self.name_token_set('#Honu Lulu')
 549
 550
 551     def test_process_place_place_empty(self):
 552         info = self.process_address(place='🜵')
 553
 554         assert 'place' not in info
 555
 556
 557     def test_process_place_address_terms(self):
 558         for name in ('Zwickau', 'Haupstraße', 'Sachsen'):
 559             self.analyzer.process_place(PlaceInfo({'name': {'name' : name}}))
 560         info = self.process_address(country='de', city='Zwickau', state='Sachsen',
 561                                     suburb='Zwickau', street='Hauptstr',
 562                                     full='right behind the church')
 563
 564         city = self.name_token_set('ZWICKAU')
 565         state = self.name_token_set('SACHSEN')
 566
 567         print(info)
 568         result = {k: eval(v[0]) for k,v in info['addr'].items()}
 569
 570         assert result == {'city': city, 'suburb': city, 'state': state}
 571
 572
 573     def test_process_place_address_terms_empty(self):
 574         info = self.process_address(country='de', city=' ', street='Hauptstr',
 575                                     full='right behind the church')
 576
 577         assert 'addr' not in info
 578