test/python/tokenizer/test_icu.py

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