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