mirror of
https://github.com/osm-search/Nominatim.git
synced 2026-02-14 18:37:58 +00:00
201 lines
6.4 KiB
Python
201 lines
6.4 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
#
|
|
# This file is part of Nominatim. (https://nominatim.org)
|
|
#
|
|
# Copyright (C) 2025 by the Nominatim developer community.
|
|
# For a full list of authors see the git log.
|
|
"""
|
|
Tests for query analyzer for ICU tokenizer.
|
|
"""
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
from nominatim_api import NominatimAPIAsync
|
|
from nominatim_api.search.query import Phrase
|
|
import nominatim_api.search.query as qmod
|
|
import nominatim_api.search.icu_tokenizer as tok
|
|
from nominatim_api.logging import set_log_output, get_and_disable
|
|
|
|
|
|
async def add_word(conn, word_id, word_token, wtype, word, info=None):
|
|
t = conn.t.meta.tables['word']
|
|
await conn.execute(t.insert(), {'word_id': word_id,
|
|
'word_token': word_token,
|
|
'type': wtype,
|
|
'word': word,
|
|
'info': info})
|
|
|
|
|
|
def make_phrase(query):
|
|
return [Phrase(qmod.PHRASE_ANY, s) for s in query.split(',')]
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def conn(table_factory):
|
|
""" Create an asynchronous SQLAlchemy engine for the test DB.
|
|
"""
|
|
table_factory('nominatim_properties',
|
|
definition='property TEXT, value TEXT',
|
|
content=(('tokenizer_import_normalisation', ':: lower();'),
|
|
('tokenizer_import_transliteration', "'1' > '/1/'; 'ä' > 'ä '")))
|
|
table_factory('word',
|
|
definition='word_id INT, word_token TEXT, type TEXT, word TEXT, info JSONB')
|
|
|
|
async with NominatimAPIAsync() as api:
|
|
async with api.begin() as conn:
|
|
yield conn
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_phrase(conn):
|
|
ana = await tok.create_query_analyzer(conn)
|
|
|
|
query = await ana.analyze_query([])
|
|
|
|
assert len(query.source) == 0
|
|
assert query.num_token_slots() == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_single_phrase_with_unknown_terms(conn):
|
|
ana = await tok.create_query_analyzer(conn)
|
|
|
|
await add_word(conn, 1, 'foo', 'w', 'FOO')
|
|
|
|
query = await ana.analyze_query(make_phrase('foo BAR'))
|
|
|
|
assert len(query.source) == 1
|
|
assert query.source[0].ptype == qmod.PHRASE_ANY
|
|
assert query.source[0].text == 'foo bar'
|
|
|
|
assert query.num_token_slots() == 2
|
|
assert query.nodes[0].partial.token == 1
|
|
assert query.nodes[1].partial is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_phrases(conn):
|
|
ana = await tok.create_query_analyzer(conn)
|
|
|
|
await add_word(conn, 1, 'one', 'w', 'one')
|
|
await add_word(conn, 2, 'two', 'w', 'two')
|
|
await add_word(conn, 100, 'one two', 'W', 'one two')
|
|
await add_word(conn, 3, 'three', 'w', 'three')
|
|
|
|
query = await ana.analyze_query(make_phrase('one two,three'))
|
|
|
|
assert len(query.source) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_splitting_in_transliteration(conn):
|
|
ana = await tok.create_query_analyzer(conn)
|
|
|
|
await add_word(conn, 1, 'mä', 'W', 'ma')
|
|
await add_word(conn, 2, 'fo', 'W', 'fo')
|
|
|
|
query = await ana.analyze_query(make_phrase('mäfo'))
|
|
|
|
assert query.num_token_slots() == 2
|
|
assert query.nodes[0].starting
|
|
assert query.nodes[1].starting
|
|
assert query.nodes[1].btype == qmod.BREAK_TOKEN
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('term,order', [('23456', ['P', 'H', 'W']),
|
|
('3', ['H', 'W'])])
|
|
async def test_penalty_postcodes_and_housenumbers(conn, term, order):
|
|
ana = await tok.create_query_analyzer(conn)
|
|
|
|
await add_word(conn, 2, term, 'H', term)
|
|
await add_word(conn, 3, term, 'w', term)
|
|
await add_word(conn, 4, term, 'W', term)
|
|
|
|
query = await ana.analyze_query(make_phrase(term))
|
|
|
|
assert query.num_token_slots() == 1
|
|
|
|
torder = [(tl.tokens[0].penalty, tl.ttype) for tl in query.nodes[0].starting]
|
|
torder.sort()
|
|
|
|
assert [t[1] for t in torder] == order
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_category_words_only_at_beginning(conn):
|
|
ana = await tok.create_query_analyzer(conn)
|
|
|
|
await add_word(conn, 1, 'foo', 'S', 'FOO', {'op': 'in'})
|
|
await add_word(conn, 2, 'bar', 'w', 'BAR')
|
|
|
|
query = await ana.analyze_query(make_phrase('foo BAR foo'))
|
|
|
|
assert query.num_token_slots() == 3
|
|
assert len(query.nodes[0].starting) == 1
|
|
assert query.nodes[0].starting[0].ttype == qmod.TOKEN_NEAR_ITEM
|
|
assert not query.nodes[2].starting
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_freestanding_qualifier_words_become_category(conn):
|
|
ana = await tok.create_query_analyzer(conn)
|
|
|
|
await add_word(conn, 1, 'foo', 'S', 'FOO', {'op': '-'})
|
|
|
|
query = await ana.analyze_query(make_phrase('foo'))
|
|
|
|
assert query.num_token_slots() == 1
|
|
assert len(query.nodes[0].starting) == 1
|
|
assert query.nodes[0].starting[0].ttype == qmod.TOKEN_NEAR_ITEM
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_qualifier_words(conn):
|
|
ana = await tok.create_query_analyzer(conn)
|
|
|
|
await add_word(conn, 1, 'foo', 'S', None, {'op': '-'})
|
|
await add_word(conn, 2, 'bar', 'w', None)
|
|
|
|
query = await ana.analyze_query(make_phrase('foo BAR foo BAR foo'))
|
|
|
|
assert query.num_token_slots() == 5
|
|
assert set(t.ttype for t in query.nodes[0].starting) == {qmod.TOKEN_QUALIFIER}
|
|
assert set(t.ttype for t in query.nodes[2].starting) == {qmod.TOKEN_QUALIFIER}
|
|
assert set(t.ttype for t in query.nodes[4].starting) == {qmod.TOKEN_QUALIFIER}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_add_unknown_housenumbers(conn):
|
|
ana = await tok.create_query_analyzer(conn)
|
|
|
|
await add_word(conn, 1, '23', 'H', '23')
|
|
|
|
query = await ana.analyze_query(make_phrase('466 23 99834 34a'))
|
|
|
|
assert query.num_token_slots() == 4
|
|
assert query.nodes[0].starting[0].ttype == qmod.TOKEN_HOUSENUMBER
|
|
assert len(query.nodes[0].starting[0].tokens) == 1
|
|
assert query.nodes[0].starting[0].tokens[0].token == 0
|
|
assert query.nodes[1].starting[0].ttype == qmod.TOKEN_HOUSENUMBER
|
|
assert len(query.nodes[1].starting[0].tokens) == 1
|
|
assert query.nodes[1].starting[0].tokens[0].token == 1
|
|
assert query.nodes[2].has_tokens(3, qmod.TOKEN_POSTCODE)
|
|
assert not query.nodes[2].has_tokens(3, qmod.TOKEN_HOUSENUMBER)
|
|
assert not query.nodes[2].has_tokens(4, qmod.TOKEN_HOUSENUMBER)
|
|
assert not query.nodes[3].has_tokens(4, qmod.TOKEN_HOUSENUMBER)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('logtype', ['text', 'html'])
|
|
async def test_log_output(conn, logtype):
|
|
|
|
ana = await tok.create_query_analyzer(conn)
|
|
|
|
await add_word(conn, 1, 'foo', 'w', 'FOO')
|
|
|
|
set_log_output(logtype)
|
|
await ana.analyze_query(make_phrase('foo'))
|
|
|
|
assert get_and_disable()
|