test/python/api/search/test_icu_query_analyzer.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) 2024 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Tests for query analyzer for ICU tokenizer.
   9 """
  10 from pathlib import Path
  11
  12 import pytest
  13 import pytest_asyncio
  14
  15 from nominatim_api import NominatimAPIAsync
  16 from nominatim_api.search.query import Phrase, PhraseType, TokenType, BreakType
  17 import nominatim_api.search.icu_tokenizer as tok
  18 from nominatim_api.logging import set_log_output, get_and_disable
  19
  20 async def add_word(conn, word_id, word_token, wtype, word, info = None):
  21     t = conn.t.meta.tables['word']
  22     await conn.execute(t.insert(), {'word_id': word_id,
  23                                     'word_token': word_token,
  24                                     'type': wtype,
  25                                     'word': word,
  26                                     'info': info})
  27
  28
  29 def make_phrase(query):
  30     return [Phrase(PhraseType.NONE, s) for s in query.split(',')]
  31
  32 @pytest_asyncio.fixture
  33 async def conn(table_factory):
  34     """ Create an asynchronous SQLAlchemy engine for the test DB.
  35     """
  36     table_factory('nominatim_properties',
  37                   definition='property TEXT, value TEXT',
  38                   content=(('tokenizer_import_normalisation', ':: lower();'),
  39                            ('tokenizer_import_transliteration', "'1' > '/1/'; 'ä' > 'ä '")))
  40     table_factory('word',
  41                   definition='word_id INT, word_token TEXT, type TEXT, word TEXT, info JSONB')
  42
  43     api = NominatimAPIAsync(Path('/invalid'), {})
  44     async with api.begin() as conn:
  45         yield conn
  46     await api.close()
  47
  48
  49 @pytest.mark.asyncio
  50 async def test_empty_phrase(conn):
  51     ana = await tok.create_query_analyzer(conn)
  52
  53     query = await ana.analyze_query([])
  54
  55     assert len(query.source) == 0
  56     assert query.num_token_slots() == 0
  57
  58
  59 @pytest.mark.asyncio
  60 async def test_single_phrase_with_unknown_terms(conn):
  61     ana = await tok.create_query_analyzer(conn)
  62
  63     await add_word(conn, 1, 'foo', 'w', 'FOO')
  64
  65     query = await ana.analyze_query(make_phrase('foo BAR'))
  66
  67     assert len(query.source) == 1
  68     assert query.source[0].ptype == PhraseType.NONE
  69     assert query.source[0].text == 'foo bar'
  70
  71     assert query.num_token_slots() == 2
  72     assert len(query.nodes[0].starting) == 1
  73     assert not query.nodes[1].starting
  74
  75
  76 @pytest.mark.asyncio
  77 async def test_multiple_phrases(conn):
  78     ana = await tok.create_query_analyzer(conn)
  79
  80     await add_word(conn, 1, 'one', 'w', 'one')
  81     await add_word(conn, 2, 'two', 'w', 'two')
  82     await add_word(conn, 100, 'one two', 'W', 'one two')
  83     await add_word(conn, 3, 'three', 'w', 'three')
  84
  85     query = await ana.analyze_query(make_phrase('one two,three'))
  86
  87     assert len(query.source) == 2
  88
  89
  90 @pytest.mark.asyncio
  91 async def test_splitting_in_transliteration(conn):
  92     ana = await tok.create_query_analyzer(conn)
  93
  94     await add_word(conn, 1, 'mä', 'W', 'ma')
  95     await add_word(conn, 2, 'fo', 'W', 'fo')
  96
  97     query = await ana.analyze_query(make_phrase('mäfo'))
  98
  99     assert query.num_token_slots() == 2
 100     assert query.nodes[0].starting
 101     assert query.nodes[1].starting
 102     assert query.nodes[1].btype == BreakType.TOKEN
 103
 104
 105 @pytest.mark.asyncio
 106 @pytest.mark.parametrize('term,order', [('23456', ['POSTCODE', 'HOUSENUMBER', 'WORD', 'PARTIAL']),
 107                                         ('3', ['HOUSENUMBER', 'POSTCODE', 'WORD', 'PARTIAL'])
 108                                        ])
 109 async def test_penalty_postcodes_and_housenumbers(conn, term, order):
 110     ana = await tok.create_query_analyzer(conn)
 111
 112     await add_word(conn, 1, term, 'P', None)
 113     await add_word(conn, 2, term, 'H', term)
 114     await add_word(conn, 3, term, 'w', term)
 115     await add_word(conn, 4, term, 'W', term)
 116
 117     query = await ana.analyze_query(make_phrase(term))
 118
 119     assert query.num_token_slots() == 1
 120
 121     torder = [(tl.tokens[0].penalty, tl.ttype.name) for tl in query.nodes[0].starting]
 122     torder.sort()
 123
 124     assert [t[1] for t in torder] == order
 125
 126 @pytest.mark.asyncio
 127 async def test_category_words_only_at_beginning(conn):
 128     ana = await tok.create_query_analyzer(conn)
 129
 130     await add_word(conn, 1, 'foo', 'S', 'FOO', {'op': 'in'})
 131     await add_word(conn, 2, 'bar', 'w', 'BAR')
 132
 133     query = await ana.analyze_query(make_phrase('foo BAR foo'))
 134
 135     assert query.num_token_slots() == 3
 136     assert len(query.nodes[0].starting) == 1
 137     assert query.nodes[0].starting[0].ttype == TokenType.NEAR_ITEM
 138     assert not query.nodes[2].starting
 139
 140
 141 @pytest.mark.asyncio
 142 async def test_freestanding_qualifier_words_become_category(conn):
 143     ana = await tok.create_query_analyzer(conn)
 144
 145     await add_word(conn, 1, 'foo', 'S', 'FOO', {'op': '-'})
 146
 147     query = await ana.analyze_query(make_phrase('foo'))
 148
 149     assert query.num_token_slots() == 1
 150     assert len(query.nodes[0].starting) == 1
 151     assert query.nodes[0].starting[0].ttype == TokenType.NEAR_ITEM
 152
 153
 154 @pytest.mark.asyncio
 155 async def test_qualifier_words(conn):
 156     ana = await tok.create_query_analyzer(conn)
 157
 158     await add_word(conn, 1, 'foo', 'S', None, {'op': '-'})
 159     await add_word(conn, 2, 'bar', 'w', None)
 160
 161     query = await ana.analyze_query(make_phrase('foo BAR foo BAR foo'))
 162
 163     assert query.num_token_slots() == 5
 164     assert set(t.ttype for t in query.nodes[0].starting) == {TokenType.QUALIFIER}
 165     assert set(t.ttype for t in query.nodes[2].starting) == {TokenType.QUALIFIER}
 166     assert set(t.ttype for t in query.nodes[4].starting) == {TokenType.QUALIFIER}
 167
 168
 169 @pytest.mark.asyncio
 170 async def test_add_unknown_housenumbers(conn):
 171     ana = await tok.create_query_analyzer(conn)
 172
 173     await add_word(conn, 1, '23', 'H', '23')
 174
 175     query = await ana.analyze_query(make_phrase('466 23 99834 34a'))
 176
 177     assert query.num_token_slots() == 4
 178     assert query.nodes[0].starting[0].ttype == TokenType.HOUSENUMBER
 179     assert len(query.nodes[0].starting[0].tokens) == 1
 180     assert query.nodes[0].starting[0].tokens[0].token == 0
 181     assert query.nodes[1].starting[0].ttype == TokenType.HOUSENUMBER
 182     assert len(query.nodes[1].starting[0].tokens) == 1
 183     assert query.nodes[1].starting[0].tokens[0].token == 1
 184     assert not query.nodes[2].starting
 185     assert not query.nodes[3].starting
 186
 187
 188 @pytest.mark.asyncio
 189 @pytest.mark.parametrize('logtype', ['text', 'html'])
 190 async def test_log_output(conn, logtype):
 191
 192     ana = await tok.create_query_analyzer(conn)
 193
 194     await add_word(conn, 1, 'foo', 'w', 'FOO')
 195
 196     set_log_output(logtype)
 197     await ana.analyze_query(make_phrase('foo'))
 198
 199     assert get_and_disable()