mirror of
https://github.com/osm-search/Nominatim.git
synced 2026-02-15 19:07:58 +00:00
882fb16881
Only use the OS environment, when the environ parameter is set to None. Currently it would use the OS env on an empty dict.
197 lines
6.4 KiB
Python
197 lines
6.4 KiB
Python
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
#
|
|
# This file is part of Nominatim. (https://nominatim.org)
|
|
#
|
|
# Copyright (C) 2024 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, PhraseType, TokenType, BreakType
|
|
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(PhraseType.NONE, 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 == PhraseType.NONE
|
|
assert query.source[0].text == 'foo bar'
|
|
|
|
assert query.num_token_slots() == 2
|
|
assert len(query.nodes[0].starting) == 1
|
|
assert not query.nodes[1].starting
|
|
|
|
|
|
@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 == BreakType.TOKEN
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize('term,order', [('23456', ['POSTCODE', 'HOUSENUMBER', 'WORD', 'PARTIAL']),
|
|
('3', ['HOUSENUMBER', 'POSTCODE', 'WORD', 'PARTIAL'])
|
|
])
|
|
async def test_penalty_postcodes_and_housenumbers(conn, term, order):
|
|
ana = await tok.create_query_analyzer(conn)
|
|
|
|
await add_word(conn, 1, term, 'P', None)
|
|
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.name) 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 == TokenType.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 == TokenType.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) == {TokenType.QUALIFIER}
|
|
assert set(t.ttype for t in query.nodes[2].starting) == {TokenType.QUALIFIER}
|
|
assert set(t.ttype for t in query.nodes[4].starting) == {TokenType.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 == TokenType.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 == TokenType.HOUSENUMBER
|
|
assert len(query.nodes[1].starting[0].tokens) == 1
|
|
assert query.nodes[1].starting[0].tokens[0].token == 1
|
|
assert not query.nodes[2].starting
|
|
assert not query.nodes[3].starting
|
|
|
|
|
|
@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()
|