test/python/tokenizer/test_icu.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 ICU tokenizer.
   9 """
  10 import shutil
  11 import yaml
  12 import itertools
  13
  14 import pytest
  15
  16 from nominatim.tokenizer import icu_tokenizer
  17 import nominatim.tokenizer.icu_rule_loader
  18 from nominatim.db import properties
  19 from nominatim.db.sql_preprocessor import SQLPreprocessor
  20 from nominatim.data.place_info import PlaceInfo
  21
  22 from mock_icu_word_table import MockIcuWordTable
  23
  24 @pytest.fixture
  25 def word_table(temp_db_conn):
  26     return MockIcuWordTable(temp_db_conn)
  27
  28
  29 @pytest.fixture
  30 def test_config(project_env, tmp_path):
  31     sqldir = tmp_path / 'sql'
  32     sqldir.mkdir()
  33     (sqldir / 'tokenizer').mkdir()
  34     (sqldir / 'tokenizer' / 'icu_tokenizer.sql').write_text("SELECT 'a'")
  35     shutil.copy(str(project_env.lib_dir.sql / 'tokenizer' / 'icu_tokenizer_tables.sql'),
  36                 str(sqldir / 'tokenizer' / 'icu_tokenizer_tables.sql'))
  37
  38     project_env.lib_dir.sql = sqldir
  39
  40     return project_env
  41
  42
  43 @pytest.fixture
  44 def tokenizer_factory(dsn, tmp_path, property_table,
  45                       sql_preprocessor, place_table, word_table):
  46     (tmp_path / 'tokenizer').mkdir()
  47
  48     def _maker():
  49         return icu_tokenizer.create(dsn, tmp_path / 'tokenizer')
  50
  51     return _maker
  52
  53
  54 @pytest.fixture
  55 def db_prop(temp_db_conn):
  56     def _get_db_property(name):
  57         return properties.get_property(temp_db_conn, name)
  58
  59     return _get_db_property
  60
  61
  62 @pytest.fixture
  63 def analyzer(tokenizer_factory, test_config, monkeypatch,
  64              temp_db_with_extensions, tmp_path):
  65     sql = tmp_path / 'sql' / 'tokenizer' / 'icu_tokenizer.sql'
  66     sql.write_text("SELECT 'a';")
  67
  68     monkeypatch.setenv('NOMINATIM_TERM_NORMALIZATION', ':: lower();')
  69     tok = tokenizer_factory()
  70     tok.init_new_db(test_config)
  71     monkeypatch.undo()
  72
  73     def _mk_analyser(norm=("[[:Punctuation:][:Space:]]+ > ' '",), trans=(':: upper()',),
  74                      variants=('~gasse -> gasse', 'street => st', ),
  75                      sanitizers=[], with_housenumber=False,
  76                      with_postcode=False):
  77         cfgstr = {'normalization': list(norm),
  78                   'sanitizers': sanitizers,
  79                   'transliteration': list(trans),
  80                   'token-analysis': [{'analyzer': 'generic',
  81                                       'variants': [{'words': list(variants)}]}]}
  82         if with_housenumber:
  83             cfgstr['token-analysis'].append({'id': '@housenumber',
  84                                              'analyzer': 'housenumbers'})
  85         if with_postcode:
  86             cfgstr['token-analysis'].append({'id': '@postcode',
  87                                              'analyzer': 'postcodes'})
  88         (test_config.project_dir / 'icu_tokenizer.yaml').write_text(yaml.dump(cfgstr))
  89         tok.loader = nominatim.tokenizer.icu_rule_loader.ICURuleLoader(test_config)
  90
  91         return tok.name_analyzer()
  92
  93     return _mk_analyser
  94
  95 @pytest.fixture
  96 def sql_functions(temp_db_conn, def_config, src_dir):
  97     orig_sql = def_config.lib_dir.sql
  98     def_config.lib_dir.sql = src_dir / 'lib-sql'
  99     sqlproc = SQLPreprocessor(temp_db_conn, def_config)
 100     sqlproc.run_sql_file(temp_db_conn, 'functions/utils.sql')
 101     sqlproc.run_sql_file(temp_db_conn, 'tokenizer/icu_tokenizer.sql')
 102     def_config.lib_dir.sql = orig_sql
 103
 104
 105 @pytest.fixture
 106 def getorcreate_full_word(temp_db_cursor):
 107     temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION getorcreate_full_word(
 108                                                  norm_term TEXT, lookup_terms TEXT[],
 109                                                  OUT full_token INT,
 110                                                  OUT partial_tokens INT[])
 111   AS $$
 112 DECLARE
 113   partial_terms TEXT[] = '{}'::TEXT[];
 114   term TEXT;
 115   term_id INTEGER;
 116   term_count INTEGER;
 117 BEGIN
 118   SELECT min(word_id) INTO full_token
 119     FROM word WHERE info->>'word' = norm_term and type = 'W';
 120
 121   IF full_token IS NULL THEN
 122     full_token := nextval('seq_word');
 123     INSERT INTO word (word_id, word_token, type, info)
 124       SELECT full_token, lookup_term, 'W',
 125              json_build_object('word', norm_term, 'count', 0)
 126         FROM unnest(lookup_terms) as lookup_term;
 127   END IF;
 128
 129   FOR term IN SELECT unnest(string_to_array(unnest(lookup_terms), ' ')) LOOP
 130     term := trim(term);
 131     IF NOT (ARRAY[term] <@ partial_terms) THEN
 132       partial_terms := partial_terms || term;
 133     END IF;
 134   END LOOP;
 135
 136   partial_tokens := '{}'::INT[];
 137   FOR term IN SELECT unnest(partial_terms) LOOP
 138     SELECT min(word_id), max(info->>'count') INTO term_id, term_count
 139       FROM word WHERE word_token = term and type = 'w';
 140
 141     IF term_id IS NULL THEN
 142       term_id := nextval('seq_word');
 143       term_count := 0;
 144       INSERT INTO word (word_id, word_token, type, info)
 145         VALUES (term_id, term, 'w', json_build_object('count', term_count));
 146     END IF;
 147
 148     IF NOT (ARRAY[term_id] <@ partial_tokens) THEN
 149       partial_tokens := partial_tokens || term_id;
 150     END IF;
 151   END LOOP;
 152 END;
 153 $$
 154 LANGUAGE plpgsql;
 155                               """)
 156
 157
 158
 159 def test_init_new(tokenizer_factory, test_config, db_prop):
 160     tok = tokenizer_factory()
 161     tok.init_new_db(test_config)
 162
 163     assert db_prop(nominatim.tokenizer.icu_rule_loader.DBCFG_IMPORT_NORM_RULES) \
 164             .startswith(':: lower ();')
 165
 166
 167 def test_init_word_table(tokenizer_factory, test_config, place_row, temp_db_cursor):
 168     place_row(names={'name' : 'Test Area', 'ref' : '52'})
 169     place_row(names={'name' : 'No Area'})
 170     place_row(names={'name' : 'Holzstrasse'})
 171
 172     tok = tokenizer_factory()
 173     tok.init_new_db(test_config)
 174
 175     assert temp_db_cursor.table_exists('word')
 176
 177
 178 def test_init_from_project(test_config, tokenizer_factory):
 179     tok = tokenizer_factory()
 180     tok.init_new_db(test_config)
 181
 182     tok = tokenizer_factory()
 183     tok.init_from_project(test_config)
 184
 185     assert tok.loader is not None
 186
 187
 188 def test_update_sql_functions(db_prop, temp_db_cursor,
 189                               tokenizer_factory, test_config, table_factory,
 190                               monkeypatch):
 191     tok = tokenizer_factory()
 192     tok.init_new_db(test_config)
 193
 194     table_factory('test', 'txt TEXT')
 195
 196     func_file = test_config.lib_dir.sql / 'tokenizer' / 'icu_tokenizer.sql'
 197     func_file.write_text("""INSERT INTO test VALUES (1133)""")
 198
 199     tok.update_sql_functions(test_config)
 200
 201     test_content = temp_db_cursor.row_set('SELECT * FROM test')
 202     assert test_content == set((('1133', ), ))
 203
 204
 205 def test_finalize_import(tokenizer_factory, temp_db_conn,
 206                          temp_db_cursor, test_config, sql_preprocessor_cfg):
 207     func_file = test_config.lib_dir.sql / 'tokenizer' / 'legacy_tokenizer_indices.sql'
 208     func_file.write_text("""CREATE FUNCTION test() RETURNS TEXT
 209                             AS $$ SELECT 'b'::text $$ LANGUAGE SQL""")
 210
 211     tok = tokenizer_factory()
 212     tok.init_new_db(test_config)
 213
 214     tok.finalize_import(test_config)
 215
 216     temp_db_cursor.scalar('SELECT test()') == 'b'
 217
 218
 219 def test_check_database(test_config, tokenizer_factory,
 220                         temp_db_cursor, sql_preprocessor_cfg):
 221     tok = tokenizer_factory()
 222     tok.init_new_db(test_config)
 223
 224     assert tok.check_database(test_config) is None
 225
 226
 227 def test_update_statistics_reverse_only(word_table, tokenizer_factory):
 228     tok = tokenizer_factory()
 229     tok.update_statistics()
 230
 231
 232 def test_update_statistics(word_table, table_factory, temp_db_cursor, tokenizer_factory):
 233     word_table.add_full_word(1000, 'hello')
 234     table_factory('search_name',
 235                   'place_id BIGINT, name_vector INT[]',
 236                   [(12, [1000])])
 237     tok = tokenizer_factory()
 238
 239     tok.update_statistics()
 240
 241     assert temp_db_cursor.scalar("""SELECT count(*) FROM word
 242                                     WHERE type = 'W' and
 243                                           (info->>'count')::int > 0""") > 0
 244
 245
 246 def test_normalize_postcode(analyzer):
 247     with analyzer() as anl:
 248         anl.normalize_postcode('123') == '123'
 249         anl.normalize_postcode('ab-34 ') == 'AB-34'
 250         anl.normalize_postcode('38 Б') == '38 Б'
 251
 252
 253 class TestPostcodes:
 254
 255     @pytest.fixture(autouse=True)
 256     def setup(self, analyzer, sql_functions):
 257         sanitizers = [{'step': 'clean-postcodes'}]
 258         with analyzer(sanitizers=sanitizers, with_postcode=True) as anl:
 259             self.analyzer = anl
 260             yield anl
 261
 262
 263     def process_postcode(self, cc, postcode):
 264         return self.analyzer.process_place(PlaceInfo({'country_code': cc,
 265                                                       'address': {'postcode': postcode}}))
 266
 267
 268     def test_update_postcodes_from_db_empty(self, table_factory, word_table):
 269         table_factory('location_postcode', 'country_code TEXT, postcode TEXT',
 270                       content=(('de', '12345'), ('se', '132 34'),
 271                                ('bm', 'AB23'), ('fr', '12345')))
 272
 273         self.analyzer.update_postcodes_from_db()
 274
 275         assert word_table.count() == 5
 276         assert word_table.get_postcodes() == {'12345', '132 34@132 34', 'AB 23@AB 23'}
 277
 278
 279     def test_update_postcodes_from_db_ambigious(self, table_factory, word_table):
 280         table_factory('location_postcode', 'country_code TEXT, postcode TEXT',
 281                       content=(('in', '123456'), ('sg', '123456')))
 282
 283         self.analyzer.update_postcodes_from_db()
 284
 285         assert word_table.count() == 3
 286         assert word_table.get_postcodes() == {'123456', '123456@123 456'}
 287
 288
 289     def test_update_postcodes_from_db_add_and_remove(self, table_factory, word_table):
 290         table_factory('location_postcode', 'country_code TEXT, postcode TEXT',
 291                       content=(('ch', '1234'), ('bm', 'BC 45'), ('bm', 'XX45')))
 292         word_table.add_postcode(' 1234', '1234')
 293         word_table.add_postcode(' 5678', '5678')
 294
 295         self.analyzer.update_postcodes_from_db()
 296
 297         assert word_table.count() == 5
 298         assert word_table.get_postcodes() == {'1234', 'BC 45@BC 45', 'XX 45@XX 45'}
 299
 300
 301     def test_process_place_postcode_simple(self, word_table):
 302         info = self.process_postcode('de', '12345')
 303
 304         assert info['postcode'] == '12345'
 305
 306         assert word_table.get_postcodes() == {'12345', }
 307
 308
 309     def test_process_place_postcode_with_space(self, word_table):
 310         info = self.process_postcode('in', '123 567')
 311
 312         assert info['postcode'] == '123567'
 313
 314         assert word_table.get_postcodes() == {'123567@123 567', }
 315
 316
 317
 318 def test_update_special_phrase_empty_table(analyzer, word_table):
 319     with analyzer() as anl:
 320         anl.update_special_phrases([
 321             ("König  bei", "amenity", "royal", "near"),
 322             ("Könige ", "amenity", "royal", "-"),
 323             ("street", "highway", "primary", "in")
 324         ], True)
 325
 326     assert word_table.get_special() \
 327                == {('KÖNIG BEI', 'König bei', 'amenity', 'royal', 'near'),
 328                    ('KÖNIGE', 'Könige', 'amenity', 'royal', None),
 329                    ('STREET', 'street', 'highway', 'primary', 'in')}
 330
 331
 332 def test_update_special_phrase_delete_all(analyzer, word_table):
 333     word_table.add_special('FOO', 'foo', 'amenity', 'prison', 'in')
 334     word_table.add_special('BAR', 'bar', 'highway', 'road', None)
 335
 336     assert word_table.count_special() == 2
 337
 338     with analyzer() as anl:
 339         anl.update_special_phrases([], True)
 340
 341     assert word_table.count_special() == 0
 342
 343
 344 def test_update_special_phrases_no_replace(analyzer, word_table):
 345     word_table.add_special('FOO', 'foo', 'amenity', 'prison', 'in')
 346     word_table.add_special('BAR', 'bar', 'highway', 'road', None)
 347
 348     assert word_table.count_special() == 2
 349
 350     with analyzer() as anl:
 351         anl.update_special_phrases([], False)
 352
 353     assert word_table.count_special() == 2
 354
 355
 356 def test_update_special_phrase_modify(analyzer, word_table):
 357     word_table.add_special('FOO', 'foo', 'amenity', 'prison', 'in')
 358     word_table.add_special('BAR', 'bar', 'highway', 'road', None)
 359
 360     assert word_table.count_special() == 2
 361
 362     with analyzer() as anl:
 363         anl.update_special_phrases([
 364             ('prison', 'amenity', 'prison', 'in'),
 365             ('bar', 'highway', 'road', '-'),
 366             ('garden', 'leisure', 'garden', 'near')
 367         ], True)
 368
 369     assert word_table.get_special() \
 370                == {('PRISON', 'prison', 'amenity', 'prison', 'in'),
 371                    ('BAR', 'bar', 'highway', 'road', None),
 372                    ('GARDEN', 'garden', 'leisure', 'garden', 'near')}
 373
 374
 375 def test_add_country_names_new(analyzer, word_table):
 376     with analyzer() as anl:
 377         anl.add_country_names('es', {'name': 'Espagña', 'name:en': 'Spain'})
 378
 379     assert word_table.get_country() == {('es', 'ESPAGÑA'), ('es', 'SPAIN')}
 380
 381
 382 def test_add_country_names_extend(analyzer, word_table):
 383     word_table.add_country('ch', 'SCHWEIZ')
 384
 385     with analyzer() as anl:
 386         anl.add_country_names('ch', {'name': 'Schweiz', 'name:fr': 'Suisse'})
 387
 388     assert word_table.get_country() == {('ch', 'SCHWEIZ'), ('ch', 'SUISSE')}
 389
 390
 391 class TestPlaceNames:
 392
 393     @pytest.fixture(autouse=True)
 394     def setup(self, analyzer, sql_functions):
 395         sanitizers = [{'step': 'split-name-list'},
 396                       {'step': 'strip-brace-terms'}]
 397         with analyzer(sanitizers=sanitizers) as anl:
 398             self.analyzer = anl
 399             yield anl
 400
 401
 402     def expect_name_terms(self, info, *expected_terms):
 403         tokens = self.analyzer.get_word_token_info(expected_terms)
 404         for token in tokens:
 405             assert token[2] is not None, "No token for {0}".format(token)
 406
 407         assert eval(info['names']) == set((t[2] for t in tokens))
 408
 409
 410     def process_named_place(self, names):
 411         return self.analyzer.process_place(PlaceInfo({'name': names}))
 412
 413
 414     def test_simple_names(self):
 415         info = self.process_named_place({'name': 'Soft bAr', 'ref': '34'})
 416
 417         self.expect_name_terms(info, '#Soft bAr', '#34', 'Soft', 'bAr', '34')
 418
 419
 420     @pytest.mark.parametrize('sep', [',' , ';'])
 421     def test_names_with_separator(self, sep):
 422         info = self.process_named_place({'name': sep.join(('New York', 'Big Apple'))})
 423
 424         self.expect_name_terms(info, '#New York', '#Big Apple',
 425                                      'new', 'york', 'big', 'apple')
 426
 427
 428     def test_full_names_with_bracket(self):
 429         info = self.process_named_place({'name': 'Houseboat (left)'})
 430
 431         self.expect_name_terms(info, '#Houseboat (left)', '#Houseboat',
 432                                      'houseboat', 'left')
 433
 434
 435     def test_country_name(self, word_table):
 436         place = PlaceInfo({'name' : {'name': 'Norge'},
 437                            'country_code': 'no',
 438                            'rank_address': 4,
 439                            'class': 'boundary',
 440                            'type': 'administrative'})
 441
 442         info = self.analyzer.process_place(place)
 443
 444         self.expect_name_terms(info, '#norge', 'norge')
 445         assert word_table.get_country() == {('no', 'NORGE')}
 446
 447
 448 class TestPlaceAddress:
 449
 450     @pytest.fixture(autouse=True)
 451     def setup(self, analyzer, sql_functions):
 452         hnr = {'step': 'clean-housenumbers',
 453                'filter-kind': ['housenumber', 'conscriptionnumber', 'streetnumber']}
 454         with analyzer(trans=(":: upper()", "'🜵' > ' '"), sanitizers=[hnr]) as anl:
 455             self.analyzer = anl
 456             yield anl
 457
 458
 459     @pytest.fixture
 460     def getorcreate_hnr_id(self, temp_db_cursor):
 461         temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION getorcreate_hnr_id(lookup_term TEXT)
 462                                   RETURNS INTEGER AS $$
 463                                     SELECT -nextval('seq_word')::INTEGER; $$ LANGUAGE SQL""")
 464
 465
 466     def process_address(self, **kwargs):
 467         return self.analyzer.process_place(PlaceInfo({'address': kwargs}))
 468
 469
 470     def name_token_set(self, *expected_terms):
 471         tokens = self.analyzer.get_word_token_info(expected_terms)
 472         for token in tokens:
 473             assert token[2] is not None, "No token for {0}".format(token)
 474
 475         return set((t[2] for t in tokens))
 476
 477
 478     @pytest.mark.parametrize('pcode', ['12345', 'AB 123', '34-345'])
 479     def test_process_place_postcode(self, word_table, pcode):
 480         self.process_address(postcode=pcode)
 481
 482         assert word_table.get_postcodes() == {pcode, }
 483
 484
 485     @pytest.mark.parametrize('hnr', ['123a', '1', '101'])
 486     def test_process_place_housenumbers_simple(self, hnr, getorcreate_hnr_id):
 487         info = self.process_address(housenumber=hnr)
 488
 489         assert info['hnr'] == hnr.upper()
 490         assert info['hnr_tokens'] == "{-1}"
 491
 492
 493     def test_process_place_housenumbers_duplicates(self, getorcreate_hnr_id):
 494         info = self.process_address(housenumber='134',
 495                                     conscriptionnumber='134',
 496                                     streetnumber='99a')
 497
 498         assert set(info['hnr'].split(';')) == set(('134', '99A'))
 499         assert info['hnr_tokens'] == "{-1,-2}"
 500
 501
 502     def test_process_place_housenumbers_cached(self, getorcreate_hnr_id):
 503         info = self.process_address(housenumber="45")
 504         assert info['hnr_tokens'] == "{-1}"
 505
 506         info = self.process_address(housenumber="46")
 507         assert info['hnr_tokens'] == "{-2}"
 508
 509         info = self.process_address(housenumber="41;45")
 510         assert eval(info['hnr_tokens']) == {-1, -3}
 511
 512         info = self.process_address(housenumber="41")
 513         assert eval(info['hnr_tokens']) == {-3}
 514
 515
 516     def test_process_place_street(self):
 517         self.analyzer.process_place(PlaceInfo({'name': {'name' : 'Grand Road'}}))
 518         info = self.process_address(street='Grand Road')
 519
 520         assert eval(info['street']) == self.name_token_set('#Grand Road')
 521
 522
 523     def test_process_place_nonexisting_street(self):
 524         info = self.process_address(street='Grand Road')
 525
 526         assert 'street' not in info
 527
 528
 529     def test_process_place_multiple_street_tags(self):
 530         self.analyzer.process_place(PlaceInfo({'name': {'name' : 'Grand Road',
 531                                                         'ref': '05989'}}))
 532         info = self.process_address(**{'street': 'Grand Road',
 533                                       'street:sym_ul': '05989'})
 534
 535         assert eval(info['street']) == self.name_token_set('#Grand Road', '#05989')
 536
 537
 538     def test_process_place_street_empty(self):
 539         info = self.process_address(street='🜵')
 540
 541         assert 'street' not in info
 542
 543
 544     def test_process_place_street_from_cache(self):
 545         self.analyzer.process_place(PlaceInfo({'name': {'name' : 'Grand Road'}}))
 546         self.process_address(street='Grand Road')
 547
 548         # request address again
 549         info = self.process_address(street='Grand Road')
 550
 551         assert eval(info['street']) == self.name_token_set('#Grand Road')
 552
 553
 554     def test_process_place_place(self):
 555         info = self.process_address(place='Honu Lulu')
 556
 557         assert eval(info['place']) == self.name_token_set('HONU', 'LULU')
 558
 559
 560     def test_process_place_place_extra(self):
 561         info = self.process_address(**{'place:en': 'Honu Lulu'})
 562
 563         assert 'place' not in info
 564
 565
 566     def test_process_place_place_empty(self):
 567         info = self.process_address(place='🜵')
 568
 569         assert 'place' not in info
 570
 571
 572     def test_process_place_address_terms(self):
 573         info = self.process_address(country='de', city='Zwickau', state='Sachsen',
 574                                     suburb='Zwickau', street='Hauptstr',
 575                                     full='right behind the church')
 576
 577         city = self.name_token_set('ZWICKAU')
 578         state = self.name_token_set('SACHSEN')
 579
 580         result = {k: eval(v) for k,v in info['addr'].items()}
 581
 582         assert result == {'city': city, 'suburb': city, 'state': state}
 583
 584
 585     def test_process_place_multiple_address_terms(self):
 586         info = self.process_address(**{'city': 'Bruxelles', 'city:de': 'Brüssel'})
 587
 588         result = {k: eval(v) for k,v in info['addr'].items()}
 589
 590         assert result == {'city': self.name_token_set('Bruxelles')}
 591
 592
 593     def test_process_place_address_terms_empty(self):
 594         info = self.process_address(country='de', city=' ', street='Hauptstr',
 595                                     full='right behind the church')
 596
 597         assert 'addr' not in info
 598
 599
 600 class TestPlaceHousenumberWithAnalyser:
 601
 602     @pytest.fixture(autouse=True)
 603     def setup(self, analyzer, sql_functions):
 604         hnr = {'step': 'clean-housenumbers',
 605                'filter-kind': ['housenumber', 'conscriptionnumber', 'streetnumber']}
 606         with analyzer(trans=(":: upper()", "'🜵' > ' '"), sanitizers=[hnr], with_housenumber=True) as anl:
 607             self.analyzer = anl
 608             yield anl
 609
 610
 611     @pytest.fixture
 612     def getorcreate_hnr_id(self, temp_db_cursor):
 613         temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION create_analyzed_hnr_id(norm_term TEXT, lookup_terms TEXT[])
 614                                   RETURNS INTEGER AS $$
 615                                     SELECT -nextval('seq_word')::INTEGER; $$ LANGUAGE SQL""")
 616
 617
 618     def process_address(self, **kwargs):
 619         return self.analyzer.process_place(PlaceInfo({'address': kwargs}))
 620
 621
 622     def name_token_set(self, *expected_terms):
 623         tokens = self.analyzer.get_word_token_info(expected_terms)
 624         for token in tokens:
 625             assert token[2] is not None, "No token for {0}".format(token)
 626
 627         return set((t[2] for t in tokens))
 628
 629
 630     @pytest.mark.parametrize('hnr', ['123 a', '1', '101'])
 631     def test_process_place_housenumbers_simple(self, hnr, getorcreate_hnr_id):
 632         info = self.process_address(housenumber=hnr)
 633
 634         assert info['hnr'] == hnr.upper()
 635         assert info['hnr_tokens'] == "{-1}"
 636
 637
 638     def test_process_place_housenumbers_duplicates(self, getorcreate_hnr_id):
 639         info = self.process_address(housenumber='134',
 640                                     conscriptionnumber='134',
 641                                     streetnumber='99a')
 642
 643         assert set(info['hnr'].split(';')) == set(('134', '99 A'))
 644         assert info['hnr_tokens'] == "{-1,-2}"
 645
 646
 647     def test_process_place_housenumbers_cached(self, getorcreate_hnr_id):
 648         info = self.process_address(housenumber="45")
 649         assert info['hnr_tokens'] == "{-1}"
 650
 651         info = self.process_address(housenumber="46")
 652         assert info['hnr_tokens'] == "{-2}"
 653
 654         info = self.process_address(housenumber="41;45")
 655         assert eval(info['hnr_tokens']) == {-1, -3}
 656
 657         info = self.process_address(housenumber="41")
 658         assert eval(info['hnr_tokens']) == {-3}
 659
 660
 661 class TestUpdateWordTokens:
 662
 663     @pytest.fixture(autouse=True)
 664     def setup(self, tokenizer_factory, table_factory, placex_table, word_table):
 665         table_factory('search_name', 'place_id BIGINT, name_vector INT[]')
 666         self.tok = tokenizer_factory()
 667
 668
 669     @pytest.fixture
 670     def search_entry(self, temp_db_cursor):
 671         place_id = itertools.count(1000)
 672
 673         def _insert(*args):
 674             temp_db_cursor.execute("INSERT INTO search_name VALUES (%s, %s)",
 675                                    (next(place_id), list(args)))
 676
 677         return _insert
 678
 679
 680     @pytest.fixture(params=['simple', 'analyzed'])
 681     def add_housenumber(self, request, word_table):
 682         if request.param == 'simple':
 683             def _make(hid, hnr):
 684                 word_table.add_housenumber(hid, hnr)
 685         elif request.param == 'analyzed':
 686             def _make(hid, hnr):
 687                 word_table.add_housenumber(hid, [hnr])
 688
 689         return _make
 690
 691
 692     @pytest.mark.parametrize('hnr', ('1a', '1234567', '34 5'))
 693     def test_remove_unused_housenumbers(self, add_housenumber, word_table, hnr):
 694         word_table.add_housenumber(1000, hnr)
 695
 696         assert word_table.count_housenumbers() == 1
 697         self.tok.update_word_tokens()
 698         assert word_table.count_housenumbers() == 0
 699
 700
 701     def test_keep_unused_numeral_housenumbers(self, add_housenumber, word_table):
 702         add_housenumber(1000, '5432')
 703
 704         assert word_table.count_housenumbers() == 1
 705         self.tok.update_word_tokens()
 706         assert word_table.count_housenumbers() == 1
 707
 708
 709     def test_keep_housenumbers_from_search_name_table(self, add_housenumber, word_table, search_entry):
 710         add_housenumber(9999, '5432a')
 711         add_housenumber(9991, '9 a')
 712         search_entry(123, 9999, 34)
 713
 714         assert word_table.count_housenumbers() == 2
 715         self.tok.update_word_tokens()
 716         assert word_table.count_housenumbers() == 1
 717
 718
 719     def test_keep_housenumbers_from_placex_table(self, add_housenumber, word_table, placex_table):
 720         add_housenumber(9999, '5432a')
 721         add_housenumber(9990, '34z')
 722         placex_table.add(housenumber='34z')
 723         placex_table.add(housenumber='25432a')
 724
 725         assert word_table.count_housenumbers() == 2
 726         self.tok.update_word_tokens()
 727         assert word_table.count_housenumbers() == 1
 728
 729
 730     def test_keep_housenumbers_from_placex_table_hnr_list(self, add_housenumber, word_table, placex_table):
 731         add_housenumber(9991, '9 b')
 732         add_housenumber(9990, '34z')
 733         placex_table.add(housenumber='9 a;9 b;9 c')
 734
 735         assert word_table.count_housenumbers() == 2
 736         self.tok.update_word_tokens()
 737         assert word_table.count_housenumbers() == 1