]> git.openstreetmap.org Git - nominatim.git/blob - test/python/tokenizer/test_icu.py
Merge remote-tracking branch 'upstream/master'
[nominatim.git] / 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 yaml
11 import itertools
12
13 import pytest
14
15 from nominatim.tokenizer import icu_tokenizer
16 import nominatim.tokenizer.icu_rule_loader
17 from nominatim.db import properties
18 from nominatim.db.sql_preprocessor import SQLPreprocessor
19 from nominatim.data.place_info import PlaceInfo
20
21 from mock_icu_word_table import MockIcuWordTable
22
23 @pytest.fixture
24 def word_table(temp_db_conn):
25     return MockIcuWordTable(temp_db_conn)
26
27
28 @pytest.fixture
29 def test_config(project_env, tmp_path):
30     sqldir = tmp_path / 'sql'
31     sqldir.mkdir()
32     (sqldir / 'tokenizer').mkdir()
33     (sqldir / 'tokenizer' / 'icu_tokenizer.sql').write_text("SELECT 'a'")
34
35     project_env.lib_dir.sql = sqldir
36
37     return project_env
38
39
40 @pytest.fixture
41 def tokenizer_factory(dsn, tmp_path, property_table,
42                       sql_preprocessor, place_table, word_table):
43     (tmp_path / 'tokenizer').mkdir()
44
45     def _maker():
46         return icu_tokenizer.create(dsn, tmp_path / 'tokenizer')
47
48     return _maker
49
50
51 @pytest.fixture
52 def db_prop(temp_db_conn):
53     def _get_db_property(name):
54         return properties.get_property(temp_db_conn, name)
55
56     return _get_db_property
57
58
59 @pytest.fixture
60 def analyzer(tokenizer_factory, test_config, monkeypatch,
61              temp_db_with_extensions, tmp_path):
62     sql = tmp_path / 'sql' / 'tokenizer' / 'icu_tokenizer.sql'
63     sql.write_text("SELECT 'a';")
64
65     monkeypatch.setenv('NOMINATIM_TERM_NORMALIZATION', ':: lower();')
66     tok = tokenizer_factory()
67     tok.init_new_db(test_config)
68     monkeypatch.undo()
69
70     def _mk_analyser(norm=("[[:Punctuation:][:Space:]]+ > ' '",), trans=(':: upper()',),
71                      variants=('~gasse -> gasse', 'street => st', ),
72                      sanitizers=[], with_housenumber=False,
73                      with_postcode=False):
74         cfgstr = {'normalization': list(norm),
75                   'sanitizers': sanitizers,
76                   'transliteration': list(trans),
77                   'token-analysis': [{'analyzer': 'generic',
78                                       'variants': [{'words': list(variants)}]}]}
79         if with_housenumber:
80             cfgstr['token-analysis'].append({'id': '@housenumber',
81                                              'analyzer': 'housenumbers'})
82         if with_postcode:
83             cfgstr['token-analysis'].append({'id': '@postcode',
84                                              'analyzer': 'postcodes'})
85         (test_config.project_dir / 'icu_tokenizer.yaml').write_text(yaml.dump(cfgstr))
86         tok.loader = nominatim.tokenizer.icu_rule_loader.ICURuleLoader(test_config)
87
88         return tok.name_analyzer()
89
90     return _mk_analyser
91
92 @pytest.fixture
93 def sql_functions(temp_db_conn, def_config, src_dir):
94     orig_sql = def_config.lib_dir.sql
95     def_config.lib_dir.sql = src_dir / 'lib-sql'
96     sqlproc = SQLPreprocessor(temp_db_conn, def_config)
97     sqlproc.run_sql_file(temp_db_conn, 'functions/utils.sql')
98     sqlproc.run_sql_file(temp_db_conn, 'tokenizer/icu_tokenizer.sql')
99     def_config.lib_dir.sql = orig_sql
100
101
102 @pytest.fixture
103 def getorcreate_full_word(temp_db_cursor):
104     temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION getorcreate_full_word(
105                                                  norm_term TEXT, lookup_terms TEXT[],
106                                                  OUT full_token INT,
107                                                  OUT partial_tokens INT[])
108   AS $$
109 DECLARE
110   partial_terms TEXT[] = '{}'::TEXT[];
111   term TEXT;
112   term_id INTEGER;
113   term_count INTEGER;
114 BEGIN
115   SELECT min(word_id) INTO full_token
116     FROM word WHERE info->>'word' = norm_term and type = 'W';
117
118   IF full_token IS NULL THEN
119     full_token := nextval('seq_word');
120     INSERT INTO word (word_id, word_token, type, info)
121       SELECT full_token, lookup_term, 'W',
122              json_build_object('word', norm_term, 'count', 0)
123         FROM unnest(lookup_terms) as lookup_term;
124   END IF;
125
126   FOR term IN SELECT unnest(string_to_array(unnest(lookup_terms), ' ')) LOOP
127     term := trim(term);
128     IF NOT (ARRAY[term] <@ partial_terms) THEN
129       partial_terms := partial_terms || term;
130     END IF;
131   END LOOP;
132
133   partial_tokens := '{}'::INT[];
134   FOR term IN SELECT unnest(partial_terms) LOOP
135     SELECT min(word_id), max(info->>'count') INTO term_id, term_count
136       FROM word WHERE word_token = term and type = 'w';
137
138     IF term_id IS NULL THEN
139       term_id := nextval('seq_word');
140       term_count := 0;
141       INSERT INTO word (word_id, word_token, type, info)
142         VALUES (term_id, term, 'w', json_build_object('count', term_count));
143     END IF;
144
145     IF NOT (ARRAY[term_id] <@ partial_tokens) THEN
146       partial_tokens := partial_tokens || term_id;
147     END IF;
148   END LOOP;
149 END;
150 $$
151 LANGUAGE plpgsql;
152                               """)
153
154
155
156 def test_init_new(tokenizer_factory, test_config, db_prop):
157     tok = tokenizer_factory()
158     tok.init_new_db(test_config)
159
160     assert db_prop(nominatim.tokenizer.icu_rule_loader.DBCFG_IMPORT_NORM_RULES) \
161             .startswith(':: lower ();')
162
163
164 def test_init_word_table(tokenizer_factory, test_config, place_row, temp_db_cursor):
165     place_row(names={'name' : 'Test Area', 'ref' : '52'})
166     place_row(names={'name' : 'No Area'})
167     place_row(names={'name' : 'Holzstrasse'})
168
169     tok = tokenizer_factory()
170     tok.init_new_db(test_config)
171
172     assert temp_db_cursor.table_exists('word')
173
174
175 def test_init_from_project(test_config, tokenizer_factory):
176     tok = tokenizer_factory()
177     tok.init_new_db(test_config)
178
179     tok = tokenizer_factory()
180     tok.init_from_project(test_config)
181
182     assert tok.loader is not None
183
184
185 def test_update_sql_functions(db_prop, temp_db_cursor,
186                               tokenizer_factory, test_config, table_factory,
187                               monkeypatch):
188     tok = tokenizer_factory()
189     tok.init_new_db(test_config)
190
191     table_factory('test', 'txt TEXT')
192
193     func_file = test_config.lib_dir.sql / 'tokenizer' / 'icu_tokenizer.sql'
194     func_file.write_text("""INSERT INTO test VALUES (1133)""")
195
196     tok.update_sql_functions(test_config)
197
198     test_content = temp_db_cursor.row_set('SELECT * FROM test')
199     assert test_content == set((('1133', ), ))
200
201
202 def test_finalize_import(tokenizer_factory, temp_db_conn,
203                          temp_db_cursor, test_config, sql_preprocessor_cfg):
204     tok = tokenizer_factory()
205     tok.init_new_db(test_config)
206
207     assert not temp_db_conn.index_exists('idx_word_word_id')
208
209     tok.finalize_import(test_config)
210
211     assert temp_db_conn.index_exists('idx_word_word_id')
212
213
214 def test_check_database(test_config, tokenizer_factory,
215                         temp_db_cursor, sql_preprocessor_cfg):
216     tok = tokenizer_factory()
217     tok.init_new_db(test_config)
218
219     assert tok.check_database(test_config) is None
220
221
222 def test_update_statistics_reverse_only(word_table, tokenizer_factory, test_config):
223     tok = tokenizer_factory()
224     tok.update_statistics(test_config)
225
226
227 def test_update_statistics(word_table, table_factory, temp_db_cursor,
228                            tokenizer_factory, test_config):
229     word_table.add_full_word(1000, 'hello')
230     word_table.add_full_word(1001, 'bye')
231     table_factory('search_name',
232                   'place_id BIGINT, name_vector INT[], nameaddress_vector INT[]',
233                   [(12, [1000], [1001])])
234     tok = tokenizer_factory()
235
236     tok.update_statistics(test_config)
237
238     assert temp_db_cursor.scalar("""SELECT count(*) FROM word
239                                     WHERE type = 'W' and word_id = 1000 and
240                                           (info->>'count')::int > 0""") == 1
241     assert temp_db_cursor.scalar("""SELECT count(*) FROM word
242                                     WHERE type = 'W' and word_id = 1001 and
243                                           (info->>'addr_count')::int > 0""") == 1
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 info['street'] == '{}'
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 info['street'] == '{}'
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', '#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', '#ZWICKAU')
578         state = self.name_token_set('SACHSEN', '#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', '#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