test/python/api/search/test_legacy_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 legacy 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.legacy_tokenizer as tok
  18 from nominatim_api.logging import set_log_output, get_and_disable
  19
  20
  21 async def add_word(conn, word_id, word_token, word, count):
  22     t = conn.t.meta.tables['word']
  23     await conn.execute(t.insert(), {'word_id': word_id,
  24                                     'word_token': word_token,
  25                                     'search_name_count': count,
  26                                     'word': word})
  27
  28
  29 async def add_housenumber(conn, word_id, hnr):
  30     t = conn.t.meta.tables['word']
  31     await conn.execute(t.insert(), {'word_id': word_id,
  32                                     'word_token': ' ' + hnr,
  33                                     'word': hnr,
  34                                     'class': 'place',
  35                                     'type': 'house'})
  36
  37
  38 async def add_postcode(conn, word_id, postcode):
  39     t = conn.t.meta.tables['word']
  40     await conn.execute(t.insert(), {'word_id': word_id,
  41                                     'word_token': ' ' + postcode,
  42                                     'word': postcode,
  43                                     'class': 'place',
  44                                     'type': 'postcode'})
  45
  46
  47 async def add_special_term(conn, word_id, word_token, cls, typ, op):
  48     t = conn.t.meta.tables['word']
  49     await conn.execute(t.insert(), {'word_id': word_id,
  50                                     'word_token': word_token,
  51                                     'word': word_token,
  52                                     'class': cls,
  53                                     'type': typ,
  54                                     'operator': op})
  55
  56
  57 def make_phrase(query):
  58     return [Phrase(PhraseType.NONE, s) for s in query.split(',')]
  59
  60
  61 @pytest_asyncio.fixture
  62 async def conn(table_factory, temp_db_cursor):
  63     """ Create an asynchronous SQLAlchemy engine for the test DB.
  64     """
  65     table_factory('nominatim_properties',
  66                   definition='property TEXT, value TEXT',
  67                   content=(('tokenizer_maxwordfreq', '10000'), ))
  68     table_factory('word',
  69                   definition="""word_id INT, word_token TEXT, word TEXT,
  70                                 class TEXT, type TEXT, country_code TEXT,
  71                                 search_name_count INT, operator TEXT
  72                              """)
  73
  74     temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION make_standard_name(name TEXT)
  75                               RETURNS TEXT AS $$ SELECT lower(name); $$ LANGUAGE SQL;""")
  76
  77     api = NominatimAPIAsync(Path('/invalid'), {})
  78     async with api.begin() as conn:
  79         yield conn
  80     await api.close()
  81
  82
  83 @pytest.mark.asyncio
  84 async def test_empty_phrase(conn):
  85     ana = await tok.create_query_analyzer(conn)
  86
  87     query = await ana.analyze_query([])
  88
  89     assert len(query.source) == 0
  90     assert query.num_token_slots() == 0
  91
  92
  93 @pytest.mark.asyncio
  94 async def test_single_phrase_with_unknown_terms(conn):
  95     ana = await tok.create_query_analyzer(conn)
  96
  97     await add_word(conn, 1, 'foo', 'FOO', 3)
  98
  99     query = await ana.analyze_query(make_phrase('foo BAR'))
 100
 101     assert len(query.source) == 1
 102     assert query.source[0].ptype == PhraseType.NONE
 103     assert query.source[0].text == 'foo bar'
 104
 105     assert query.num_token_slots() == 2
 106     assert len(query.nodes[0].starting) == 1
 107     assert not query.nodes[1].starting
 108
 109
 110 @pytest.mark.asyncio
 111 async def test_multiple_phrases(conn):
 112     ana = await tok.create_query_analyzer(conn)
 113
 114     await add_word(conn, 1, 'one', 'one', 13)
 115     await add_word(conn, 2, 'two', 'two', 45)
 116     await add_word(conn, 100, 'one two', 'one two', 3)
 117     await add_word(conn, 3, 'three', 'three', 4584)
 118
 119     query = await ana.analyze_query(make_phrase('one two,three'))
 120
 121     assert len(query.source) == 2
 122
 123
 124 @pytest.mark.asyncio
 125 async def test_housenumber_token(conn):
 126     ana = await tok.create_query_analyzer(conn)
 127
 128     await add_housenumber(conn, 556, '45 a')
 129
 130     query = await ana.analyze_query(make_phrase('45 A'))
 131
 132     assert query.num_token_slots() == 2
 133     assert len(query.nodes[0].starting) == 2
 134
 135     query.nodes[0].starting.sort(key=lambda tl: tl.end)
 136
 137     hn1 = query.nodes[0].starting[0]
 138     assert hn1.ttype == TokenType.HOUSENUMBER
 139     assert hn1.end == 1
 140     assert hn1.tokens[0].token == 0
 141
 142     hn2 = query.nodes[0].starting[1]
 143     assert hn2.ttype == TokenType.HOUSENUMBER
 144     assert hn2.end == 2
 145     assert hn2.tokens[0].token == 556
 146
 147
 148 @pytest.mark.asyncio
 149 async def test_postcode_token(conn):
 150     ana = await tok.create_query_analyzer(conn)
 151
 152     await add_postcode(conn, 34, '45ax')
 153
 154     query = await ana.analyze_query(make_phrase('45AX'))
 155
 156     assert query.num_token_slots() == 1
 157     assert [tl.ttype for tl in query.nodes[0].starting] == [TokenType.POSTCODE]
 158
 159
 160 @pytest.mark.asyncio
 161 async def test_partial_tokens(conn):
 162     ana = await tok.create_query_analyzer(conn)
 163
 164     await add_word(conn, 1, ' foo', 'foo', 99)
 165     await add_word(conn, 1, 'foo', 'FOO', 99)
 166     await add_word(conn, 1, 'bar', 'FOO', 990000)
 167
 168     query = await ana.analyze_query(make_phrase('foo bar'))
 169
 170     assert query.num_token_slots() == 2
 171
 172     first = query.nodes[0].starting
 173     first.sort(key=lambda tl: tl.tokens[0].penalty)
 174     assert [tl.ttype for tl in first] == [TokenType.WORD, TokenType.PARTIAL]
 175     assert all(tl.tokens[0].lookup_word == 'foo' for tl in first)
 176
 177     second = query.nodes[1].starting
 178     assert [tl.ttype for tl in second] == [TokenType.PARTIAL]
 179     assert not second[0].tokens[0].is_indexed
 180
 181
 182 @pytest.mark.asyncio
 183 @pytest.mark.parametrize('term,order', [('23456', ['POSTCODE', 'HOUSENUMBER', 'WORD', 'PARTIAL']),
 184                                         ('3', ['HOUSENUMBER', 'POSTCODE', 'WORD', 'PARTIAL'])
 185                                        ])
 186 async def test_penalty_postcodes_and_housenumbers(conn, term, order):
 187     ana = await tok.create_query_analyzer(conn)
 188
 189     await add_postcode(conn, 1, term)
 190     await add_housenumber(conn, 2, term)
 191     await add_word(conn, 3, term, term, 5)
 192     await add_word(conn, 4, ' ' + term, term, 1)
 193
 194     query = await ana.analyze_query(make_phrase(term))
 195
 196     assert query.num_token_slots() == 1
 197
 198     torder = [(tl.tokens[0].penalty, tl.ttype.name) for tl in query.nodes[0].starting]
 199     torder.sort()
 200
 201     assert [t[1] for t in torder] == order
 202
 203
 204 @pytest.mark.asyncio
 205 async def test_category_words_only_at_beginning(conn):
 206     ana = await tok.create_query_analyzer(conn)
 207
 208     await add_special_term(conn, 1, 'foo', 'amenity', 'restaurant', 'in')
 209     await add_word(conn, 2, ' bar', 'BAR', 1)
 210
 211     query = await ana.analyze_query(make_phrase('foo BAR foo'))
 212
 213     assert query.num_token_slots() == 3
 214     assert len(query.nodes[0].starting) == 1
 215     assert query.nodes[0].starting[0].ttype == TokenType.NEAR_ITEM
 216     assert not query.nodes[2].starting
 217
 218
 219 @pytest.mark.asyncio
 220 async def test_qualifier_words(conn):
 221     ana = await tok.create_query_analyzer(conn)
 222
 223     await add_special_term(conn, 1, 'foo', 'amenity', 'restaurant', '-')
 224     await add_word(conn, 2, ' bar', 'w', None)
 225
 226     query = await ana.analyze_query(make_phrase('foo BAR foo BAR foo'))
 227
 228     assert query.num_token_slots() == 5
 229     assert set(t.ttype for t in query.nodes[0].starting) == {TokenType.NEAR_ITEM, TokenType.QUALIFIER}
 230     assert set(t.ttype for t in query.nodes[2].starting) == {TokenType.QUALIFIER}
 231     assert set(t.ttype for t in query.nodes[4].starting) == {TokenType.NEAR_ITEM, TokenType.QUALIFIER}
 232
 233
 234 @pytest.mark.asyncio
 235 @pytest.mark.parametrize('logtype', ['text', 'html'])
 236 async def test_log_output(conn, logtype):
 237     ana = await tok.create_query_analyzer(conn)
 238
 239     await add_word(conn, 1, 'foo', 'FOO', 99)
 240
 241     set_log_output(logtype)
 242     await ana.analyze_query(make_phrase('foo'))
 243
 244     assert get_and_disable()