E711
per-file-ignores =
__init__.py: F401
+ test/python/utils/test_json_writer.py: E131
+ test/python/conftest.py: E402
pytest test/python
lint:
- flake8 src
+ flake8 src test/python
bdd:
cd test/bdd; behave -DREMOVE_TEMPLATE=1
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Helper fixtures for API call tests.
"""
import pytest
import pytest_asyncio
-import time
import datetime as dt
import sqlalchemy as sa
from nominatim_db.tools import convert_sqlite
import nominatim_api.logging as loglib
+
class APITester:
def __init__(self):
self.api = napi.NominatimAPI()
self.async_to_sync(self.api._async_api.setup_database())
-
def async_to_sync(self, func):
""" Run an asynchronous function until completion using the
internal loop of the API.
"""
return self.api._loop.run_until_complete(func)
-
def add_data(self, table, data):
""" Insert data into the given table.
"""
sql = getattr(self.api._async_api._tables, table).insert()
self.async_to_sync(self.exec_async(sql, data))
-
def add_placex(self, **kw):
name = kw.get('name')
if isinstance(name, str):
geometry = kw.get('geometry', 'POINT(%f %f)' % centroid)
self.add_data('placex',
- {'place_id': kw.get('place_id', 1000),
- 'osm_type': kw.get('osm_type', 'W'),
- 'osm_id': kw.get('osm_id', 4),
- 'class_': kw.get('class_', 'highway'),
- 'type': kw.get('type', 'residential'),
- 'name': name,
- 'address': kw.get('address'),
- 'extratags': kw.get('extratags'),
- 'parent_place_id': kw.get('parent_place_id'),
- 'linked_place_id': kw.get('linked_place_id'),
- 'admin_level': kw.get('admin_level', 15),
- 'country_code': kw.get('country_code'),
- 'housenumber': kw.get('housenumber'),
- 'postcode': kw.get('postcode'),
- 'wikipedia': kw.get('wikipedia'),
- 'rank_search': kw.get('rank_search', 30),
- 'rank_address': kw.get('rank_address', 30),
- 'importance': kw.get('importance'),
- 'centroid': 'POINT(%f %f)' % centroid,
- 'indexed_status': kw.get('indexed_status', 0),
- 'indexed_date': kw.get('indexed_date',
- dt.datetime(2022, 12, 7, 14, 14, 46, 0)),
- 'geometry': geometry})
-
+ {'place_id': kw.get('place_id', 1000),
+ 'osm_type': kw.get('osm_type', 'W'),
+ 'osm_id': kw.get('osm_id', 4),
+ 'class_': kw.get('class_', 'highway'),
+ 'type': kw.get('type', 'residential'),
+ 'name': name,
+ 'address': kw.get('address'),
+ 'extratags': kw.get('extratags'),
+ 'parent_place_id': kw.get('parent_place_id'),
+ 'linked_place_id': kw.get('linked_place_id'),
+ 'admin_level': kw.get('admin_level', 15),
+ 'country_code': kw.get('country_code'),
+ 'housenumber': kw.get('housenumber'),
+ 'postcode': kw.get('postcode'),
+ 'wikipedia': kw.get('wikipedia'),
+ 'rank_search': kw.get('rank_search', 30),
+ 'rank_address': kw.get('rank_address', 30),
+ 'importance': kw.get('importance'),
+ 'centroid': 'POINT(%f %f)' % centroid,
+ 'indexed_status': kw.get('indexed_status', 0),
+ 'indexed_date': kw.get('indexed_date',
+ dt.datetime(2022, 12, 7, 14, 14, 46, 0)),
+ 'geometry': geometry})
def add_address_placex(self, object_id, **kw):
self.add_placex(**kw)
'fromarea': kw.get('fromarea', False),
'isaddress': kw.get('isaddress', True)})
-
def add_osmline(self, **kw):
self.add_data('osmline',
- {'place_id': kw.get('place_id', 10000),
- 'osm_id': kw.get('osm_id', 4004),
- 'parent_place_id': kw.get('parent_place_id'),
- 'indexed_date': kw.get('indexed_date',
- dt.datetime(2022, 12, 7, 14, 14, 46, 0)),
- 'startnumber': kw.get('startnumber', 2),
- 'endnumber': kw.get('endnumber', 6),
- 'step': kw.get('step', 2),
- 'address': kw.get('address'),
- 'postcode': kw.get('postcode'),
- 'country_code': kw.get('country_code'),
- 'linegeo': kw.get('geometry', 'LINESTRING(1.1 -0.2, 1.09 -0.22)')})
-
+ {'place_id': kw.get('place_id', 10000),
+ 'osm_id': kw.get('osm_id', 4004),
+ 'parent_place_id': kw.get('parent_place_id'),
+ 'indexed_date': kw.get('indexed_date',
+ dt.datetime(2022, 12, 7, 14, 14, 46, 0)),
+ 'startnumber': kw.get('startnumber', 2),
+ 'endnumber': kw.get('endnumber', 6),
+ 'step': kw.get('step', 2),
+ 'address': kw.get('address'),
+ 'postcode': kw.get('postcode'),
+ 'country_code': kw.get('country_code'),
+ 'linegeo': kw.get('geometry', 'LINESTRING(1.1 -0.2, 1.09 -0.22)')})
def add_tiger(self, **kw):
self.add_data('tiger',
- {'place_id': kw.get('place_id', 30000),
- 'parent_place_id': kw.get('parent_place_id'),
- 'startnumber': kw.get('startnumber', 2),
- 'endnumber': kw.get('endnumber', 6),
- 'step': kw.get('step', 2),
- 'postcode': kw.get('postcode'),
- 'linegeo': kw.get('geometry', 'LINESTRING(1.1 -0.2, 1.09 -0.22)')})
-
+ {'place_id': kw.get('place_id', 30000),
+ 'parent_place_id': kw.get('parent_place_id'),
+ 'startnumber': kw.get('startnumber', 2),
+ 'endnumber': kw.get('endnumber', 6),
+ 'step': kw.get('step', 2),
+ 'postcode': kw.get('postcode'),
+ 'linegeo': kw.get('geometry', 'LINESTRING(1.1 -0.2, 1.09 -0.22)')})
def add_postcode(self, **kw):
self.add_data('postcode',
- {'place_id': kw.get('place_id', 1000),
- 'parent_place_id': kw.get('parent_place_id'),
- 'country_code': kw.get('country_code'),
- 'postcode': kw.get('postcode'),
- 'rank_search': kw.get('rank_search', 20),
- 'rank_address': kw.get('rank_address', 22),
- 'indexed_date': kw.get('indexed_date',
- dt.datetime(2022, 12, 7, 14, 14, 46, 0)),
- 'geometry': kw.get('geometry', 'POINT(23 34)')})
-
+ {'place_id': kw.get('place_id', 1000),
+ 'parent_place_id': kw.get('parent_place_id'),
+ 'country_code': kw.get('country_code'),
+ 'postcode': kw.get('postcode'),
+ 'rank_search': kw.get('rank_search', 20),
+ 'rank_address': kw.get('rank_address', 22),
+ 'indexed_date': kw.get('indexed_date',
+ dt.datetime(2022, 12, 7, 14, 14, 46, 0)),
+ 'geometry': kw.get('geometry', 'POINT(23 34)')})
def add_country(self, country_code, geometry):
self.add_data('country_grid',
'area': 0.1,
'geometry': geometry})
-
def add_country_name(self, country_code, names, partition=0):
self.add_data('country_name',
{'country_code': country_code,
'name': names,
'partition': partition})
-
def add_search_name(self, place_id, **kw):
centroid = kw.get('centroid', (23.0, 34.0))
self.add_data('search_name',
'country_code': kw.get('country_code', 'xx'),
'centroid': 'POINT(%f %f)' % centroid})
-
def add_class_type_table(self, cls, typ):
self.async_to_sync(
self.exec_async(sa.text(f"""CREATE TABLE place_classtype_{cls}_{typ}
WHERE class = '{cls}' AND type = '{typ}')
""")))
-
def add_word_table(self, content):
data = [dict(zip(['word_id', 'word_token', 'type', 'word', 'info'], c))
for c in content]
self.async_to_sync(_do_sql())
-
async def exec_async(self, sql, *args, **kwargs):
async with self.api._async_api.begin() as conn:
return await conn.execute(sql, *args, **kwargs)
-
async def create_tables(self):
async with self.api._async_api._engine.begin() as conn:
await conn.run_sync(self.api._async_api._tables.meta.create_all)
db = str(tmp_path / 'test_nominatim_python_unittest.sqlite')
def mkapi(apiobj, options={'reverse'}):
- apiobj.add_data('properties',
- [{'property': 'tokenizer', 'value': 'icu'},
- {'property': 'tokenizer_import_normalisation', 'value': ':: lower();'},
- {'property': 'tokenizer_import_transliteration', 'value': "'1' > '/1/'; 'ä' > 'ä '"},
- ])
+ apiobj.add_data(
+ 'properties',
+ [{'property': 'tokenizer', 'value': 'icu'},
+ {'property': 'tokenizer_import_normalisation', 'value': ':: lower();'},
+ {'property': 'tokenizer_import_transliteration',
+ 'value': "'1' > '/1/'; 'ä' > 'ä '"}])
async def _do_sql():
async with apiobj.api._async_api.begin() as conn:
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Provides dummy implementations of ASGIAdaptor for testing.
from nominatim_api.v1.format import dispatch as formatting
from nominatim_api.config import Configuration
+
class FakeError(BaseException):
def __init__(self, msg, status):
def __str__(self):
return f'{self.status} -- {self.msg}'
+
FakeResponse = namedtuple('FakeResponse', ['status', 'output', 'content_type'])
+
class FakeAdaptor(glue.ASGIAdaptor):
def __init__(self, params=None, headers=None, config=None):
self.headers = headers or {}
self._config = config or Configuration(None)
-
def get(self, name, default=None):
return self.params.get(name, default)
-
def get_header(self, name, default=None):
return self.headers.get(name, default)
-
def error(self, msg, status=400):
return FakeError(msg, status)
-
def create_response(self, status, output, num_results):
return FakeResponse(status, output, self.content_type)
-
def base_uri(self):
return 'http://test'
def formatting(self):
return formatting
-
-
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for normalizing search queries.
"""
-from pathlib import Path
-
-import pytest
-
from icu import Transliterator
import nominatim_api.search.query as qmod
from nominatim_api.query_preprocessing.config import QueryConfig
from nominatim_api.query_preprocessing import normalize
+
def run_preprocessor_on(query, norm):
normalizer = Transliterator.createFromRules("normalization", norm)
proc = normalize.create(QueryConfig().set_normalizer(normalizer))
"""
Tests for japanese phrase splitting.
"""
-from pathlib import Path
-
import pytest
-from icu import Transliterator
-
import nominatim_api.search.query as qmod
from nominatim_api.query_preprocessing.config import QueryConfig
from nominatim_api.query_preprocessing import split_japanese_phrases
+
def run_preprocessor_on(query):
proc = split_japanese_phrases.create(QueryConfig().set_normalizer(None))
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for tokenized query data structures.
from nominatim_api.search import query
+
class MyToken(query.Token):
def get_category(self):
return MyToken(penalty=3.0, token=tid, count=1, addr_count=1,
lookup_word='foo')
+
@pytest.fixture
def qnode():
- return query.QueryNode(query.BREAK_PHRASE, query.PHRASE_ANY, 0.0 ,'', '')
+ return query.QueryNode(query.BREAK_PHRASE, query.PHRASE_ANY, 0.0, '', '')
+
@pytest.mark.parametrize('ptype,ttype', [(query.PHRASE_ANY, 'W'),
(query.PHRASE_AMENITY, 'Q'),
assert len(q.get_tokens(query.TokenRange(1, 2), query.TOKEN_PARTIAL)) == 1
assert len(q.get_tokens(query.TokenRange(1, 2), query.TOKEN_NEAR_ITEM)) == 0
assert len(q.get_tokens(query.TokenRange(1, 2), query.TOKEN_QUALIFIER)) == 1
-
from nominatim_api.types import SearchDetails
import nominatim_api.search.db_searches as dbs
+
class MyToken(Token):
def get_category(self):
return 'this', 'that'
token=tid, count=1, addr_count=1,
lookup_word=word))
-
return q
[(2, qmod.TOKEN_PARTIAL, [(2, 'b')]),
(2, qmod.TOKEN_WORD, [(101, 'b')])],
[(3, qmod.TOKEN_PARTIAL, [(3, 'c')]),
- (3, qmod.TOKEN_WORD, [(102, 'c')])]
- )
+ (3, qmod.TOKEN_WORD, [(102, 'c')])])
builder = SearchBuilder(q, SearchDetails())
searches = list(builder.build(TokenAssignment(name=TokenRange(0, 1),
(3, qmod.TOKEN_WORD, [(101, 'bc')])],
[(3, qmod.TOKEN_PARTIAL, [(3, 'c')])],
[(4, qmod.TOKEN_PARTIAL, [(4, 'd')]),
- (4, qmod.TOKEN_WORD, [(103, 'd')])]
- )
+ (4, qmod.TOKEN_WORD, [(103, 'd')])])
builder = SearchBuilder(q, SearchDetails())
searches = list(builder.build(TokenAssignment(name=TokenRange(0, 1),
assert len(search.lookups) == 2
assert len(search.rankings) == 2
- assert set((l.column, l.lookup_type.__name__) for l in search.lookups) == \
- {('name_vector', 'LookupAll'), ('nameaddress_vector', 'Restrict')}
+ assert set((s.column, s.lookup_type.__name__) for s in search.lookups) == \
+ {('name_vector', 'LookupAll'), ('nameaddress_vector', 'Restrict')}
def test_frequent_partials_in_name_and_address():
assert all(isinstance(s, dbs.PlaceSearch) for s in searches)
searches.sort(key=lambda s: s.penalty)
- assert set((l.column, l.lookup_type.__name__) for l in searches[0].lookups) == \
- {('name_vector', 'LookupAny'), ('nameaddress_vector', 'Restrict')}
- assert set((l.column, l.lookup_type.__name__) for l in searches[1].lookups) == \
- {('nameaddress_vector', 'LookupAll'), ('name_vector', 'LookupAll')}
+ assert set((s.column, s.lookup_type.__name__) for s in searches[0].lookups) == \
+ {('name_vector', 'LookupAny'), ('nameaddress_vector', 'Restrict')}
+ assert set((s.column, s.lookup_type.__name__) for s in searches[1].lookups) == \
+ {('nameaddress_vector', 'LookupAll'), ('name_vector', 'LookupAll')}
def test_too_frequent_partials_in_name_and_address():
assert all(isinstance(s, dbs.PlaceSearch) for s in searches)
searches.sort(key=lambda s: s.penalty)
- assert set((l.column, l.lookup_type.__name__) for l in searches[0].lookups) == \
- {('name_vector', 'LookupAny'), ('nameaddress_vector', 'Restrict')}
+ assert set((s.column, s.lookup_type.__name__) for s in searches[0].lookups) == \
+ {('name_vector', 'LookupAny'), ('nameaddress_vector', 'Restrict')}
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# 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 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):
+
+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,
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.
@pytest.mark.asyncio
@pytest.mark.parametrize('term,order', [('23456', ['P', 'H', 'W', 'w']),
- ('3', ['H', 'W', 'w'])
- ])
+ ('3', ['H', 'W', 'w'])])
async def test_penalty_postcodes_and_housenumbers(conn, term, order):
ana = await tok.create_query_analyzer(conn)
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)
from nominatim_api.search.postcode_parser import PostcodeParser
from nominatim_api.search.query import QueryStruct, PHRASE_ANY, PHRASE_POSTCODE, PHRASE_STREET
+
@pytest.fixture
def pc_config(project_env):
country_file = project_env.project_dir / 'country_settings.yaml'
return project_env
+
def mk_query(inp):
query = QueryStruct([])
phrase_split = re.split(r"([ ,:'-])", inp)
assert result == {(pos, pos + 1, '45325'), (pos, pos + 1, '453 25')}
+
def test_contained_postcode(pc_config):
parser = PostcodeParser(pc_config)
(0, 2, '12345 DX')}
-
@pytest.mark.parametrize('query,frm,to', [('345987', 0, 1), ('345 987', 0, 2),
('Aina 345 987', 1, 3),
('Aina 23 345 987 ff', 2, 4)])
assert result == {(frm, to, '345987')}
+
def test_overlapping_postcode(pc_config):
parser = PostcodeParser(pc_config)
assert not parser.parse(mk_query('ky12233'))
+
def test_postcode_inside_postcode_phrase(pc_config):
parser = PostcodeParser(pc_config)
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Test data types for search queries.
import nominatim_api.search.query as nq
+
def test_token_range_equal():
assert nq.TokenRange(2, 3) == nq.TokenRange(2, 3)
assert not (nq.TokenRange(2, 3) != nq.TokenRange(2, 3))
@pytest.mark.parametrize('lop,rop', [((1, 2), (3, 4)),
- ((3, 4), (3, 5)),
- ((10, 12), (11, 12))])
+ ((3, 4), (3, 5)),
+ ((10, 12), (11, 12))])
def test_token_range_unequal(lop, rop):
assert not (nq.TokenRange(*lop) == nq.TokenRange(*rop))
assert nq.TokenRange(*lop) != nq.TokenRange(*rop)
assert nq.TokenRange(1, 3) < nq.TokenRange(10, 12)
assert nq.TokenRange(5, 6) < nq.TokenRange(7, 8)
assert nq.TokenRange(1, 4) < nq.TokenRange(4, 5)
- assert not(nq.TokenRange(5, 6) < nq.TokenRange(5, 6))
- assert not(nq.TokenRange(10, 11) < nq.TokenRange(4, 5))
+ assert not (nq.TokenRange(5, 6) < nq.TokenRange(5, 6))
+ assert not (nq.TokenRange(10, 11) < nq.TokenRange(4, 5))
def test_token_rankge_gt():
assert nq.TokenRange(3, 4) > nq.TokenRange(1, 2)
assert nq.TokenRange(100, 200) > nq.TokenRange(10, 11)
assert nq.TokenRange(10, 11) > nq.TokenRange(4, 10)
- assert not(nq.TokenRange(5, 6) > nq.TokenRange(5, 6))
- assert not(nq.TokenRange(1, 2) > nq.TokenRange(3, 4))
- assert not(nq.TokenRange(4, 10) > nq.TokenRange(3, 5))
+ assert not (nq.TokenRange(5, 6) > nq.TokenRange(5, 6))
+ assert not (nq.TokenRange(1, 2) > nq.TokenRange(3, 4))
+ assert not (nq.TokenRange(4, 10) > nq.TokenRange(3, 5))
def test_token_range_unimplemented_ops():
words = q.extract_words(base_penalty=1.0)
assert set(words.keys()) \
- == {'12', 'ab', 'hallo', '12 ab', 'ab 12', '12 ab 12'}
+ == {'12', 'ab', 'hallo', '12 ab', 'ab 12', '12 ab 12'}
assert sorted(words['12']) == [nq.TokenRange(0, 1, 1.0), nq.TokenRange(2, 3, 1.0)]
assert words['12 ab'] == [nq.TokenRange(0, 2, 1.1)]
assert words['hallo'] == [nq.TokenRange(3, 4, 1.0)]
-
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for query analyzer creation.
"""
-from pathlib import Path
-
import pytest
from nominatim_api.search.query_analyzer_factory import make_query_analyzer
from nominatim_api.search.icu_tokenizer import ICUQueryAnalyzer
+
@pytest.mark.asyncio
async def test_import_icu_tokenizer(table_factory, api):
table_factory('nominatim_properties',
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for running the country searcher.
assert results[0].place_id == 55
assert results[0].accuracy == 0.8
+
def test_find_from_fallback_countries(apiobj, frontend):
apiobj.add_country('ro', 'POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))')
apiobj.add_country_name('ro', {'name': 'România'})
apiobj.add_country('ro', 'POLYGON((0 0, 0 1, 1 1, 1 0, 0 0))')
apiobj.add_country_name('ro', {'name': 'România'})
-
@pytest.mark.parametrize('geom', [napi.GeometryFormat.GEOJSON,
napi.GeometryFormat.KML,
napi.GeometryFormat.SVG,
assert len(results) == 1
assert geom.name.lower() in results[0].geometry
-
@pytest.mark.parametrize('pid,rids', [(76, [55]), (55, [])])
def test_exclude_place_id(self, apiobj, frontend, pid, rids):
results = run_search(apiobj, frontend, 0.5, ['yw', 'ro'],
assert [r.place_id for r in results] == rids
-
@pytest.mark.parametrize('viewbox,rids', [((9, 9, 11, 11), [55]),
((-10, -10, -3, -3), [])])
def test_bounded_viewbox_in_placex(self, apiobj, frontend, viewbox, rids):
assert [r.place_id for r in results] == rids
-
@pytest.mark.parametrize('viewbox,numres', [((0, 0, 1, 1), 1),
- ((-10, -10, -3, -3), 0)])
+ ((-10, -10, -3, -3), 0)])
def test_bounded_viewbox_in_fallback(self, apiobj, frontend, viewbox, numres):
results = run_search(apiobj, frontend, 0.5, ['ro'],
details=SearchDetails.from_kwargs({'viewbox': viewbox,
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for running the near searcher.
import nominatim_api as napi
from nominatim_api.types import SearchDetails
from nominatim_api.search.db_searches import NearSearch, PlaceSearch
-from nominatim_api.search.db_search_fields import WeightedStrings, WeightedCategories,\
- FieldLookup, FieldRanking, RankedTokens
+from nominatim_api.search.db_search_fields import WeightedStrings, WeightedCategories, \
+ FieldLookup
from nominatim_api.search.db_search_lookups import LookupAll
apiobj.add_search_name(101, names=[56], country_code='mx',
centroid=(-10.3, 56.9))
-
def test_near_in_placex(self, apiobj, frontend):
apiobj.add_placex(place_id=22, class_='amenity', type='bank',
centroid=(5.6001, 4.2994))
assert [r.place_id for r in results] == [22]
-
def test_multiple_types_near_in_placex(self, apiobj, frontend):
apiobj.add_placex(place_id=22, class_='amenity', type='bank',
importance=0.002,
assert [r.place_id for r in results] == [22, 23]
-
def test_near_in_classtype(self, apiobj, frontend):
apiobj.add_placex(place_id=22, class_='amenity', type='bank',
centroid=(5.6, 4.34))
assert [r.place_id for r in results] == [22]
-
@pytest.mark.parametrize('cc,rid', [('us', 22), ('mx', 23)])
def test_restrict_by_country(self, apiobj, frontend, cc, rid):
apiobj.add_placex(place_id=22, class_='amenity', type='bank',
assert [r.place_id for r in results] == [rid]
-
@pytest.mark.parametrize('excluded,rid', [(22, 122), (122, 22)])
def test_exclude_place_by_id(self, apiobj, frontend, excluded, rid):
apiobj.add_placex(place_id=22, class_='amenity', type='bank',
centroid=(5.6001, 4.2994),
country_code='us')
-
results = run_search(apiobj, frontend, 0.1, [('amenity', 'bank')],
details=SearchDetails(excluded=[excluded]))
assert [r.place_id for r in results] == [rid]
-
@pytest.mark.parametrize('layer,rids', [(napi.DataLayer.POI, [22]),
(napi.DataLayer.MANMADE, [])])
def test_with_layer(self, apiobj, frontend, layer, rids):
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for running the generic place searcher.
import nominatim_api as napi
from nominatim_api.types import SearchDetails
from nominatim_api.search.db_searches import PlaceSearch
-from nominatim_api.search.db_search_fields import WeightedStrings, WeightedCategories,\
+from nominatim_api.search.db_search_fields import WeightedStrings, WeightedCategories, \
FieldLookup, FieldRanking, RankedTokens
from nominatim_api.search.db_search_lookups import LookupAll, LookupAny, Restrict
APIOPTIONS = ['search']
+
def run_search(apiobj, frontend, global_penalty, lookup, ranking, count=2,
hnrs=[], pcs=[], ccodes=[], quals=[],
details=SearchDetails()):
def fill_database(self, apiobj):
apiobj.add_placex(place_id=100, country_code='us',
centroid=(5.6, 4.3))
- apiobj.add_search_name(100, names=[1,2,10,11], country_code='us',
+ apiobj.add_search_name(100, names=[1, 2, 10, 11], country_code='us',
centroid=(5.6, 4.3))
apiobj.add_placex(place_id=101, country_code='mx',
centroid=(-10.3, 56.9))
- apiobj.add_search_name(101, names=[1,2,20,21], country_code='mx',
+ apiobj.add_search_name(101, names=[1, 2, 20, 21], country_code='mx',
centroid=(-10.3, 56.9))
-
@pytest.mark.parametrize('lookup_type', [LookupAll, Restrict])
@pytest.mark.parametrize('rank,res', [([10], [100, 101]),
([20], [101, 100])])
def test_lookup_all_match(self, apiobj, frontend, lookup_type, rank, res):
- lookup = FieldLookup('name_vector', [1,2], lookup_type)
+ lookup = FieldLookup('name_vector', [1, 2], lookup_type)
ranking = FieldRanking('name_vector', 0.4, [RankedTokens(0.0, rank)])
results = run_search(apiobj, frontend, 0.1, [lookup], [ranking])
assert [r.place_id for r in results] == res
-
@pytest.mark.parametrize('lookup_type', [LookupAll, Restrict])
def test_lookup_all_partial_match(self, apiobj, frontend, lookup_type):
- lookup = FieldLookup('name_vector', [1,20], lookup_type)
+ lookup = FieldLookup('name_vector', [1, 20], lookup_type)
ranking = FieldRanking('name_vector', 0.4, [RankedTokens(0.0, [21])])
results = run_search(apiobj, frontend, 0.1, [lookup], [ranking])
@pytest.mark.parametrize('rank,res', [([10], [100, 101]),
([20], [101, 100])])
def test_lookup_any_match(self, apiobj, frontend, rank, res):
- lookup = FieldLookup('name_vector', [11,21], LookupAny)
+ lookup = FieldLookup('name_vector', [11, 21], LookupAny)
ranking = FieldRanking('name_vector', 0.4, [RankedTokens(0.0, rank)])
results = run_search(apiobj, frontend, 0.1, [lookup], [ranking])
assert [r.place_id for r in results] == res
-
def test_lookup_any_partial_match(self, apiobj, frontend):
lookup = FieldLookup('name_vector', [20], LookupAll)
ranking = FieldRanking('name_vector', 0.4, [RankedTokens(0.0, [21])])
assert len(results) == 1
assert results[0].place_id == 101
-
@pytest.mark.parametrize('cc,res', [('us', 100), ('mx', 101)])
def test_lookup_restrict_country(self, apiobj, frontend, cc, res):
- lookup = FieldLookup('name_vector', [1,2], LookupAll)
+ lookup = FieldLookup('name_vector', [1, 2], LookupAll)
ranking = FieldRanking('name_vector', 0.4, [RankedTokens(0.0, [10])])
results = run_search(apiobj, frontend, 0.1, [lookup], [ranking], ccodes=[cc])
assert [r.place_id for r in results] == [res]
-
def test_lookup_restrict_placeid(self, apiobj, frontend):
- lookup = FieldLookup('name_vector', [1,2], LookupAll)
+ lookup = FieldLookup('name_vector', [1, 2], LookupAll)
ranking = FieldRanking('name_vector', 0.4, [RankedTokens(0.0, [10])])
results = run_search(apiobj, frontend, 0.1, [lookup], [ranking],
assert [r.place_id for r in results] == [100]
-
@pytest.mark.parametrize('geom', [napi.GeometryFormat.GEOJSON,
napi.GeometryFormat.KML,
napi.GeometryFormat.SVG,
assert geom.name.lower() in results[0].geometry
-
@pytest.mark.parametrize('factor,npoints', [(0.0, 3), (1.0, 2)])
def test_return_simplified_geometry(self, apiobj, frontend, factor, npoints):
apiobj.add_placex(place_id=333, country_code='us',
assert result.place_id == 333
assert len(geom['coordinates']) == npoints
-
@pytest.mark.parametrize('viewbox', ['5.0,4.0,6.0,5.0', '5.7,4.0,6.0,5.0'])
@pytest.mark.parametrize('wcount,rids', [(2, [100, 101]), (20000, [100])])
def test_prefer_viewbox(self, apiobj, frontend, viewbox, wcount, rids):
details=SearchDetails.from_kwargs({'viewbox': viewbox}))
assert [r.place_id for r in results] == rids
-
@pytest.mark.parametrize('viewbox', ['5.0,4.0,6.0,5.0', '5.55,4.27,5.62,4.31'])
def test_force_viewbox(self, apiobj, frontend, viewbox):
lookup = FieldLookup('name_vector', [1, 2], LookupAll)
- details=SearchDetails.from_kwargs({'viewbox': viewbox,
- 'bounded_viewbox': True})
+ details = SearchDetails.from_kwargs({'viewbox': viewbox,
+ 'bounded_viewbox': True})
results = run_search(apiobj, frontend, 0.1, [lookup], [], details=details)
assert [r.place_id for r in results] == [100]
-
def test_prefer_near(self, apiobj, frontend):
lookup = FieldLookup('name_vector', [1, 2], LookupAll)
ranking = FieldRanking('name_vector', 0.4, [RankedTokens(0.0, [21])])
results.sort(key=lambda r: -r.importance)
assert [r.place_id for r in results] == [100, 101]
-
@pytest.mark.parametrize('radius', [0.09, 0.11])
def test_force_near(self, apiobj, frontend, radius):
lookup = FieldLookup('name_vector', [1, 2], LookupAll)
- details=SearchDetails.from_kwargs({'near': '5.6,4.3',
- 'near_radius': radius})
+ details = SearchDetails.from_kwargs({'near': '5.6,4.3',
+ 'near_radius': radius})
results = run_search(apiobj, frontend, 0.1, [lookup], [], details=details)
apiobj.add_placex(place_id=1000, class_='highway', type='residential',
rank_search=26, rank_address=26,
country_code='es')
- apiobj.add_search_name(1000, names=[1,2,10,11],
+ apiobj.add_search_name(1000, names=[1, 2, 10, 11],
search_rank=26, address_rank=26,
country_code='es')
apiobj.add_placex(place_id=91, class_='place', type='house',
apiobj.add_placex(place_id=2000, class_='highway', type='residential',
rank_search=26, rank_address=26,
country_code='pt')
- apiobj.add_search_name(2000, names=[1,2,20,21],
+ apiobj.add_search_name(2000, names=[1, 2, 20, 21],
search_rank=26, address_rank=26,
country_code='pt')
-
@pytest.mark.parametrize('hnr,res', [('20', [91, 1]), ('20 a', [1]),
('21', [2]), ('22', [2, 92]),
('24', [93]), ('25', [])])
def test_lookup_by_single_housenumber(self, apiobj, frontend, hnr, res):
- lookup = FieldLookup('name_vector', [1,2], LookupAll)
+ lookup = FieldLookup('name_vector', [1, 2], LookupAll)
ranking = FieldRanking('name_vector', 0.3, [RankedTokens(0.0, [10])])
results = run_search(apiobj, frontend, 0.1, [lookup], [ranking], hnrs=[hnr])
assert [r.place_id for r in results] == res + [1000, 2000]
-
@pytest.mark.parametrize('cc,res', [('es', [2, 1000]), ('pt', [92, 2000])])
def test_lookup_with_country_restriction(self, apiobj, frontend, cc, res):
- lookup = FieldLookup('name_vector', [1,2], LookupAll)
+ lookup = FieldLookup('name_vector', [1, 2], LookupAll)
ranking = FieldRanking('name_vector', 0.3, [RankedTokens(0.0, [10])])
results = run_search(apiobj, frontend, 0.1, [lookup], [ranking], hnrs=['22'],
assert [r.place_id for r in results] == res
-
def test_lookup_exclude_housenumber_placeid(self, apiobj, frontend):
- lookup = FieldLookup('name_vector', [1,2], LookupAll)
+ lookup = FieldLookup('name_vector', [1, 2], LookupAll)
ranking = FieldRanking('name_vector', 0.3, [RankedTokens(0.0, [10])])
results = run_search(apiobj, frontend, 0.1, [lookup], [ranking], hnrs=['22'],
assert [r.place_id for r in results] == [2, 1000, 2000]
-
def test_lookup_exclude_street_placeid(self, apiobj, frontend):
- lookup = FieldLookup('name_vector', [1,2], LookupAll)
+ lookup = FieldLookup('name_vector', [1, 2], LookupAll)
ranking = FieldRanking('name_vector', 0.3, [RankedTokens(0.0, [10])])
results = run_search(apiobj, frontend, 0.1, [lookup], [ranking], hnrs=['22'],
assert [r.place_id for r in results] == [2, 92, 2000]
-
def test_lookup_only_house_qualifier(self, apiobj, frontend):
- lookup = FieldLookup('name_vector', [1,2], LookupAll)
+ lookup = FieldLookup('name_vector', [1, 2], LookupAll)
ranking = FieldRanking('name_vector', 0.3, [RankedTokens(0.0, [10])])
results = run_search(apiobj, frontend, 0.1, [lookup], [ranking], hnrs=['22'],
assert [r.place_id for r in results] == [2, 92]
-
def test_lookup_only_street_qualifier(self, apiobj, frontend):
- lookup = FieldLookup('name_vector', [1,2], LookupAll)
+ lookup = FieldLookup('name_vector', [1, 2], LookupAll)
ranking = FieldRanking('name_vector', 0.3, [RankedTokens(0.0, [10])])
results = run_search(apiobj, frontend, 0.1, [lookup], [ranking], hnrs=['22'],
assert [r.place_id for r in results] == [1000, 2000]
-
@pytest.mark.parametrize('rank,found', [(26, True), (27, False), (30, False)])
def test_lookup_min_rank(self, apiobj, frontend, rank, found):
- lookup = FieldLookup('name_vector', [1,2], LookupAll)
+ lookup = FieldLookup('name_vector', [1, 2], LookupAll)
ranking = FieldRanking('name_vector', 0.3, [RankedTokens(0.0, [10])])
results = run_search(apiobj, frontend, 0.1, [lookup], [ranking], hnrs=['22'],
assert [r.place_id for r in results] == ([2, 92, 1000, 2000] if found else [2, 92])
-
@pytest.mark.parametrize('geom', [napi.GeometryFormat.GEOJSON,
napi.GeometryFormat.KML,
napi.GeometryFormat.SVG,
apiobj.add_placex(place_id=2000, class_='highway', type='residential',
rank_search=26, rank_address=26,
country_code='pt')
- apiobj.add_search_name(2000, names=[1,2],
+ apiobj.add_search_name(2000, names=[1, 2],
search_rank=26, address_rank=26,
country_code='pt')
centroid=(10.0, 10.00001),
geometry='LINESTRING(9.995 10.00001, 10.005 10.00001)')
-
@pytest.mark.parametrize('hnr,res', [('21', [992]), ('22', []), ('23', [991])])
def test_lookup_housenumber(self, apiobj, frontend, hnr, res):
lookup = FieldLookup('name_vector', [111], LookupAll)
assert [r.place_id for r in results] == res + [990]
-
@pytest.mark.parametrize('geom', [napi.GeometryFormat.GEOJSON,
napi.GeometryFormat.KML,
napi.GeometryFormat.SVG,
assert geom.name.lower() in results[0].geometry
-
class TestTiger:
@pytest.fixture(autouse=True)
centroid=(10.0, 10.00001),
geometry='LINESTRING(9.995 10.00001, 10.005 10.00001)')
-
@pytest.mark.parametrize('hnr,res', [('21', [992]), ('22', []), ('23', [991])])
def test_lookup_housenumber(self, apiobj, frontend, hnr, res):
lookup = FieldLookup('name_vector', [111], LookupAll)
assert [r.place_id for r in results] == res + [990]
-
@pytest.mark.parametrize('geom', [napi.GeometryFormat.GEOJSON,
napi.GeometryFormat.KML,
napi.GeometryFormat.SVG,
importance=0.0005,
address_rank=0, search_rank=30)
-
- @pytest.mark.parametrize('layer,res', [(napi.DataLayer.ADDRESS, [223]),
- (napi.DataLayer.POI, [224]),
- (napi.DataLayer.ADDRESS | napi.DataLayer.POI, [223, 224]),
- (napi.DataLayer.MANMADE, [225]),
- (napi.DataLayer.RAILWAY, [226]),
- (napi.DataLayer.NATURAL, [227]),
- (napi.DataLayer.MANMADE | napi.DataLayer.NATURAL, [225, 227]),
- (napi.DataLayer.MANMADE | napi.DataLayer.RAILWAY, [225, 226])])
+ @pytest.mark.parametrize('layer,res',
+ [(napi.DataLayer.ADDRESS, [223]),
+ (napi.DataLayer.POI, [224]),
+ (napi.DataLayer.ADDRESS | napi.DataLayer.POI, [223, 224]),
+ (napi.DataLayer.MANMADE, [225]),
+ (napi.DataLayer.RAILWAY, [226]),
+ (napi.DataLayer.NATURAL, [227]),
+ (napi.DataLayer.MANMADE | napi.DataLayer.NATURAL, [225, 227]),
+ (napi.DataLayer.MANMADE | napi.DataLayer.RAILWAY, [225, 226])])
def test_layers_rank30(self, apiobj, frontend, layer, res):
lookup = FieldLookup('name_vector', [34], LookupAny)
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for running the POI searcher.
"""
import pytest
-import nominatim_api as napi
from nominatim_api.types import SearchDetails
from nominatim_api.search.db_searches import PoiSearch
from nominatim_api.search.db_search_fields import WeightedStrings, WeightedCategories
else:
self.args = {'near': '34.3, 56.100021', 'near_radius': 0.001}
-
def test_unrestricted(self, apiobj, frontend):
results = run_search(apiobj, frontend, 0.1, [('highway', 'bus_stop')], [0.5],
details=SearchDetails.from_kwargs(self.args))
assert [r.place_id for r in results] == [1, 2]
-
def test_restict_country(self, apiobj, frontend):
results = run_search(apiobj, frontend, 0.1, [('highway', 'bus_stop')], [0.5],
ccodes=['de', 'nz'],
assert [r.place_id for r in results] == [2]
-
def test_restrict_by_viewbox(self, apiobj, frontend):
args = {'bounded_viewbox': True, 'viewbox': '34.299,56.0,34.3001,56.10001'}
args.update(self.args)
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for running the postcode searcher.
from nominatim_api.search.db_search_fields import WeightedStrings, FieldLookup, \
FieldRanking, RankedTokens
+
def run_search(apiobj, frontend, global_penalty, pcs, pc_penalties=None,
ccodes=[], lookup=[], ranking=[], details=SearchDetails()):
if pc_penalties is None:
apiobj.add_placex(place_id=1000, class_='place', type='village',
rank_search=22, rank_address=22,
country_code='ch')
- apiobj.add_search_name(1000, names=[1,2,10,11],
+ apiobj.add_search_name(1000, names=[1, 2, 10, 11],
search_rank=22, address_rank=22,
country_code='ch')
apiobj.add_placex(place_id=2000, class_='place', type='village',
rank_search=22, rank_address=22,
country_code='pl')
- apiobj.add_search_name(2000, names=[1,2,20,21],
+ apiobj.add_search_name(2000, names=[1, 2, 20, 21],
search_rank=22, address_rank=22,
country_code='pl')
-
def test_lookup_both(self, apiobj, frontend):
- lookup = FieldLookup('name_vector', [1,2], 'restrict')
+ lookup = FieldLookup('name_vector', [1, 2], 'restrict')
ranking = FieldRanking('name_vector', 0.3, [RankedTokens(0.0, [10])])
results = run_search(apiobj, frontend, 0.1, ['12345'], lookup=[lookup], ranking=[ranking])
assert [r.place_id for r in results] == [100, 101]
-
def test_restrict_by_name(self, apiobj, frontend):
lookup = FieldLookup('name_vector', [10], 'restrict')
assert [r.place_id for r in results] == [100]
-
@pytest.mark.parametrize('coord,place_id', [((16.5, 5), 100),
((-45.1, 7.004), 101)])
def test_lookup_near(self, apiobj, frontend, coord, place_id):
- lookup = FieldLookup('name_vector', [1,2], 'restrict')
+ lookup = FieldLookup('name_vector', [1, 2], 'restrict')
ranking = FieldRanking('name_vector', 0.3, [RankedTokens(0.0, [10])])
results = run_search(apiobj, frontend, 0.1, ['12345'],
assert [r.place_id for r in results] == [place_id]
-
@pytest.mark.parametrize('geom', [napi.GeometryFormat.GEOJSON,
napi.GeometryFormat.KML,
napi.GeometryFormat.SVG,
assert results
assert all(geom.name.lower() in r.geometry for r in results)
-
- @pytest.mark.parametrize('viewbox, rids', [('-46,6,-44,8', [101,100]),
- ('16,4,18,6', [100,101])])
+ @pytest.mark.parametrize('viewbox, rids', [('-46,6,-44,8', [101, 100]),
+ ('16,4,18,6', [100, 101])])
def test_prefer_viewbox(self, apiobj, frontend, viewbox, rids):
results = run_search(apiobj, frontend, 0.1, ['12345'],
details=SearchDetails.from_kwargs({'viewbox': viewbox}))
assert [r.place_id for r in results] == rids
-
@pytest.mark.parametrize('viewbox, rid', [('-46,6,-44,8', 101),
- ('16,4,18,6', 100)])
+ ('16,4,18,6', 100)])
def test_restrict_to_viewbox(self, apiobj, frontend, viewbox, rid):
results = run_search(apiobj, frontend, 0.1, ['12345'],
details=SearchDetails.from_kwargs({'viewbox': viewbox,
assert [r.place_id for r in results] == [rid]
-
@pytest.mark.parametrize('coord,rids', [((17.05, 5), [100, 101]),
((-45, 7.1), [101, 100])])
def test_prefer_near(self, apiobj, frontend, coord, rids):
assert [r.place_id for r in results] == rids
-
@pytest.mark.parametrize('pid,rid', [(100, 101), (101, 100)])
def test_exclude(self, apiobj, frontend, pid, rid):
results = run_search(apiobj, frontend, 0.1, ['12345'],
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Test for creation of token assignments from tokenized queries.
from nominatim_api.search.query import QueryStruct, Phrase, TokenRange, Token
import nominatim_api.search.query as qmod
-from nominatim_api.search.token_assignment import yield_token_assignments, TokenAssignment, PENALTY_TOKENCHANGE
+from nominatim_api.search.token_assignment import (yield_token_assignments,
+ TokenAssignment,
+ PENALTY_TOKENCHANGE)
+
class MyToken(Token):
def get_category(self):
TokenAssignment(penalty=penalty, name=TokenRange(1, 3),
address=[TokenRange(0, 1)]),
TokenAssignment(penalty=penalty, name=TokenRange(2, 3),
- address=[TokenRange(0, 2)])
- )
+ address=[TokenRange(0, 2)]))
def test_multiple_words_respect_phrase_break():
address=[TokenRange(0, 1), TokenRange(2, 3)],
postcode=TokenRange(3, 4)))
+
def test_postcode_and_housenumber():
q = make_query((qmod.BREAK_START, qmod.PHRASE_ANY, [(1, qmod.TOKEN_PARTIAL)]),
(qmod.BREAK_WORD, qmod.PHRASE_ANY, [(2, qmod.TOKEN_POSTCODE)]),
check_assignments(yield_token_assignments(q),
TokenAssignment(penalty=0.1,
name=TokenRange(4, 5),
- housenumber=TokenRange(3, 4),\
+ housenumber=TokenRange(3, 4),
address=[TokenRange(0, 1), TokenRange(1, 2),
TokenRange(2, 3)]),
TokenAssignment(penalty=0.1,
- housenumber=TokenRange(3, 4),\
+ housenumber=TokenRange(3, 4),
address=[TokenRange(0, 1), TokenRange(1, 2),
TokenRange(2, 3), TokenRange(4, 5)]))
(qmod.BREAK_WORD, qmod.PHRASE_ANY, [(2, qmod.TOKEN_PARTIAL)]),
(qmod.BREAK_WORD, qmod.PHRASE_ANY, [(3, qmod.TOKEN_PARTIAL)]))
-
check_assignments(yield_token_assignments(q),
TokenAssignment(penalty=0.1, name=TokenRange(1, 3),
qualifier=TokenRange(0, 1)),
(qmod.BREAK_WORD, qmod.PHRASE_ANY, [(4, qmod.TOKEN_PARTIAL)]),
(qmod.BREAK_WORD, qmod.PHRASE_ANY, [(5, qmod.TOKEN_PARTIAL)]))
-
check_assignments(yield_token_assignments(q),
TokenAssignment(penalty=0.2, name=TokenRange(0, 2),
qualifier=TokenRange(2, 3),
(qmod.BREAK_PHRASE, qmod.PHRASE_ANY, [(5, qmod.TOKEN_PARTIAL)]))
check_assignments(yield_token_assignments(q))
-
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for enhanced connection class for API functions.
"""
-from pathlib import Path
import pytest
import sqlalchemy as sa
@pytest.mark.asyncio
-async def test_get_db_property_existing(api):
+async def test_get_db_property_bad_name(api):
async with api.begin() as conn:
with pytest.raises(ValueError):
await conn.get_db_property('dfkgjd.rijg')
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the deletable v1 API call.
"""
import json
-from pathlib import Path
import pytest
-from fake_adaptor import FakeAdaptor, FakeError, FakeResponse
+from fake_adaptor import FakeAdaptor
import nominatim_api.v1.server_glue as glue
+
class TestDeletableEndPoint:
@pytest.fixture(autouse=True)
content=[(345, 'N', 'boundary', 'administrative'),
(781, 'R', 'landuse', 'wood'),
(781, 'R', 'landcover', 'grass')])
- table_factory('placex',
- definition="""place_id bigint, osm_id bigint, osm_type char(1),
- class text, type text, name HSTORE, country_code char(2)""",
- content=[(1, 345, 'N', 'boundary', 'administrative', {'old_name': 'Former'}, 'ab'),
- (2, 781, 'R', 'landuse', 'wood', {'name': 'Wood'}, 'cd'),
- (3, 781, 'R', 'landcover', 'grass', None, 'cd')])
-
-
+ table_factory(
+ 'placex',
+ definition="""place_id bigint, osm_id bigint, osm_type char(1),
+ class text, type text, name HSTORE, country_code char(2)""",
+ content=[(1, 345, 'N', 'boundary', 'administrative', {'old_name': 'Former'}, 'ab'),
+ (2, 781, 'R', 'landuse', 'wood', {'name': 'Wood'}, 'cd'),
+ (3, 781, 'R', 'landcover', 'grass', None, 'cd')])
@pytest.mark.asyncio
async def test_deletable(self, api):
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for details API call.
import nominatim_api as napi
+
@pytest.mark.parametrize('idobj', (napi.PlaceID(332), napi.OsmID('W', 4),
napi.OsmID('W', 4, 'highway')))
def test_lookup_in_placex(apiobj, frontend, idobj):
import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0)
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential',
- name={'name': 'Road'}, address={'city': 'Barrow'},
- extratags={'surface': 'paved'},
- parent_place_id=34, linked_place_id=55,
- admin_level=15, country_code='gb',
- housenumber='4',
- postcode='34425', wikipedia='en:Faa',
- rank_search=27, rank_address=26,
- importance=0.01,
- centroid=(23, 34),
- indexed_date=import_date,
- geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
+ class_='highway', type='residential',
+ name={'name': 'Road'}, address={'city': 'Barrow'},
+ extratags={'surface': 'paved'},
+ parent_place_id=34, linked_place_id=55,
+ admin_level=15, country_code='gb',
+ housenumber='4',
+ postcode='34425', wikipedia='en:Faa',
+ rank_search=27, rank_address=26,
+ importance=0.01,
+ centroid=(23, 34),
+ indexed_date=import_date,
+ geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
api = frontend(apiobj, options={'details'})
result = api.details(idobj)
def test_lookup_in_placex_minimal_info(apiobj, frontend):
import_date = dt.datetime(2022, 12, 7, 14, 14, 46, 0)
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential',
- admin_level=15,
- rank_search=27, rank_address=26,
- centroid=(23, 34),
- indexed_date=import_date,
- geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
+ class_='highway', type='residential',
+ admin_level=15,
+ rank_search=27, rank_address=26,
+ centroid=(23, 34),
+ indexed_date=import_date,
+ geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(332))
def test_lookup_placex_with_address_details(apiobj, frontend):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential', name='Street',
- country_code='pl',
- rank_search=27, rank_address=26)
+ class_='highway', type='residential', name='Street',
+ country_code='pl',
+ rank_search=27, rank_address=26)
apiobj.add_address_placex(332, fromarea=False, isaddress=False,
distance=0.0034,
place_id=1000, osm_type='N', osm_id=3333,
def test_lookup_place_with_linked_places_none_existing(apiobj, frontend):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential', name='Street',
- country_code='pl', linked_place_id=45,
- rank_search=27, rank_address=26)
+ class_='highway', type='residential', name='Street',
+ country_code='pl', linked_place_id=45,
+ rank_search=27, rank_address=26)
api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(332), linked_places=True)
def test_lookup_place_with_linked_places_existing(apiobj, frontend):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential', name='Street',
- country_code='pl', linked_place_id=45,
- rank_search=27, rank_address=26)
+ class_='highway', type='residential', name='Street',
+ country_code='pl', linked_place_id=45,
+ rank_search=27, rank_address=26)
apiobj.add_placex(place_id=1001, osm_type='W', osm_id=5,
- class_='highway', type='residential', name='Street',
- country_code='pl', linked_place_id=332,
- rank_search=27, rank_address=26)
+ class_='highway', type='residential', name='Street',
+ country_code='pl', linked_place_id=332,
+ rank_search=27, rank_address=26)
apiobj.add_placex(place_id=1002, osm_type='W', osm_id=6,
- class_='highway', type='residential', name='Street',
- country_code='pl', linked_place_id=332,
- rank_search=27, rank_address=26)
+ class_='highway', type='residential', name='Street',
+ country_code='pl', linked_place_id=332,
+ rank_search=27, rank_address=26)
api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(332), linked_places=True)
def test_lookup_place_with_parented_places_not_existing(apiobj, frontend):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential', name='Street',
- country_code='pl', parent_place_id=45,
- rank_search=27, rank_address=26)
+ class_='highway', type='residential', name='Street',
+ country_code='pl', parent_place_id=45,
+ rank_search=27, rank_address=26)
api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(332), parented_places=True)
def test_lookup_place_with_parented_places_existing(apiobj, frontend):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential', name='Street',
- country_code='pl', parent_place_id=45,
- rank_search=27, rank_address=26)
+ class_='highway', type='residential', name='Street',
+ country_code='pl', parent_place_id=45,
+ rank_search=27, rank_address=26)
apiobj.add_placex(place_id=1001, osm_type='N', osm_id=5,
- class_='place', type='house', housenumber='23',
- country_code='pl', parent_place_id=332,
- rank_search=30, rank_address=30)
+ class_='place', type='house', housenumber='23',
+ country_code='pl', parent_place_id=332,
+ rank_search=30, rank_address=30)
apiobj.add_placex(place_id=1002, osm_type='W', osm_id=6,
- class_='highway', type='residential', name='Street',
- country_code='pl', parent_place_id=332,
- rank_search=27, rank_address=26)
+ class_='highway', type='residential', name='Street',
+ country_code='pl', parent_place_id=332,
+ rank_search=27, rank_address=26)
api = frontend(apiobj, options={'details'})
result = api.details(napi.PlaceID(332), parented_places=True)
startnumber=2, endnumber=4, step=1,
parent_place_id=332)
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential', name='Street',
- country_code='pl',
- rank_search=27, rank_address=26)
+ class_='highway', type='residential', name='Street',
+ country_code='pl',
+ rank_search=27, rank_address=26)
apiobj.add_address_placex(332, fromarea=False, isaddress=False,
distance=0.0034,
place_id=1000, osm_type='N', osm_id=3333,
startnumber=2, endnumber=4, step=1,
parent_place_id=332)
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential', name='Street',
- country_code='us',
- rank_search=27, rank_address=26)
+ class_='highway', type='residential', name='Street',
+ country_code='us',
+ rank_search=27, rank_address=26)
apiobj.add_address_placex(332, fromarea=False, isaddress=False,
distance=0.0034,
place_id=1000, osm_type='N', osm_id=3333,
rank_address=4, distance=0.0)
]
+
@pytest.mark.parametrize('objid', [napi.PlaceID(1736),
napi.OsmID('W', 55),
napi.OsmID('N', 55, 'amenity')])
@pytest.mark.parametrize('gtype', (napi.GeometryFormat.KML,
- napi.GeometryFormat.SVG,
- napi.GeometryFormat.TEXT))
+ napi.GeometryFormat.SVG,
+ napi.GeometryFormat.TEXT))
def test_lookup_unsupported_geometry(apiobj, frontend, gtype):
apiobj.add_placex(place_id=332)
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for lookup API call.
import nominatim_api as napi
+
def test_lookup_empty_list(apiobj, frontend):
api = frontend(apiobj, options={'details'})
assert api.lookup([]) == []
napi.OsmID('W', 4, 'highway')))
def test_lookup_single_placex(apiobj, frontend, idobj):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential',
- name={'name': 'Road'}, address={'city': 'Barrow'},
- extratags={'surface': 'paved'},
- parent_place_id=34, linked_place_id=55,
- admin_level=15, country_code='gb',
- housenumber='4',
- postcode='34425', wikipedia='en:Faa',
- rank_search=27, rank_address=26,
- importance=0.01,
- centroid=(23, 34),
- geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
+ class_='highway', type='residential',
+ name={'name': 'Road'}, address={'city': 'Barrow'},
+ extratags={'surface': 'paved'},
+ parent_place_id=34, linked_place_id=55,
+ admin_level=15, country_code='gb',
+ housenumber='4',
+ postcode='34425', wikipedia='en:Faa',
+ rank_search=27, rank_address=26,
+ importance=0.01,
+ centroid=(23, 34),
+ geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
api = frontend(apiobj, options={'details'})
result = api.lookup([idobj])
def test_lookup_multiple_places(apiobj, frontend):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential',
- name={'name': 'Road'}, address={'city': 'Barrow'},
- extratags={'surface': 'paved'},
- parent_place_id=34, linked_place_id=55,
- admin_level=15, country_code='gb',
- housenumber='4',
- postcode='34425', wikipedia='en:Faa',
- rank_search=27, rank_address=26,
- importance=0.01,
- centroid=(23, 34),
- geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
+ class_='highway', type='residential',
+ name={'name': 'Road'}, address={'city': 'Barrow'},
+ extratags={'surface': 'paved'},
+ parent_place_id=34, linked_place_id=55,
+ admin_level=15, country_code='gb',
+ housenumber='4',
+ postcode='34425', wikipedia='en:Faa',
+ rank_search=27, rank_address=26,
+ importance=0.01,
+ centroid=(23, 34),
+ geometry='LINESTRING(23 34, 23.1 34, 23.1 34.1, 23 34)')
apiobj.add_osmline(place_id=4924, osm_id=9928,
parent_place_id=12,
startnumber=1, endnumber=4, step=1,
address={'city': 'Big'},
geometry='LINESTRING(23 34, 23 35)')
-
api = frontend(apiobj, options={'details'})
result = api.lookup((napi.OsmID('W', 1),
napi.OsmID('W', 4),
@pytest.mark.parametrize('gtype', list(napi.GeometryFormat))
def test_simple_place_with_geometry(apiobj, frontend, gtype):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential',
- name={'name': 'Road'}, address={'city': 'Barrow'},
- extratags={'surface': 'paved'},
- parent_place_id=34, linked_place_id=55,
- admin_level=15, country_code='gb',
- housenumber='4',
- postcode='34425', wikipedia='en:Faa',
- rank_search=27, rank_address=26,
- importance=0.01,
- centroid=(23, 34),
- geometry='POLYGON((23 34, 23.1 34, 23.1 34.1, 23 34))')
+ class_='highway', type='residential',
+ name={'name': 'Road'}, address={'city': 'Barrow'},
+ extratags={'surface': 'paved'},
+ parent_place_id=34, linked_place_id=55,
+ admin_level=15, country_code='gb',
+ housenumber='4',
+ postcode='34425', wikipedia='en:Faa',
+ rank_search=27, rank_address=26,
+ importance=0.01,
+ centroid=(23, 34),
+ geometry='POLYGON((23 34, 23.1 34, 23.1 34.1, 23 34))')
api = frontend(apiobj, options={'details'})
result = api.lookup([napi.OsmID('W', 4)], geometry_output=gtype)
def test_simple_place_with_geometry_simplified(apiobj, frontend):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential',
- name={'name': 'Road'}, address={'city': 'Barrow'},
- extratags={'surface': 'paved'},
- parent_place_id=34, linked_place_id=55,
- admin_level=15, country_code='gb',
- housenumber='4',
- postcode='34425', wikipedia='en:Faa',
- rank_search=27, rank_address=26,
- importance=0.01,
- centroid=(23, 34),
- geometry='POLYGON((23 34, 22.999 34, 23.1 34, 23.1 34.1, 23 34))')
+ class_='highway', type='residential',
+ name={'name': 'Road'}, address={'city': 'Barrow'},
+ extratags={'surface': 'paved'},
+ parent_place_id=34, linked_place_id=55,
+ admin_level=15, country_code='gb',
+ housenumber='4',
+ postcode='34425', wikipedia='en:Faa',
+ rank_search=27, rank_address=26,
+ importance=0.01,
+ centroid=(23, 34),
+ geometry='POLYGON((23 34, 22.999 34, 23.1 34, 23.1 34.1, 23 34))')
api = frontend(apiobj, options={'details'})
result = api.lookup([napi.OsmID('W', 4)],
geom = json.loads(result[0].geometry['geojson'])
- assert geom['type'] == 'Polygon'
+ assert geom['type'] == 'Polygon'
assert geom['coordinates'] == [[[23, 34], [23.1, 34], [23.1, 34.1], [23, 34]]]
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the deletable v1 API call.
"""
import json
import datetime as dt
-from pathlib import Path
import pytest
-from fake_adaptor import FakeAdaptor, FakeError, FakeResponse
+from fake_adaptor import FakeAdaptor
import nominatim_api.v1.server_glue as glue
+
class TestPolygonsEndPoint:
@pytest.fixture(autouse=True)
errormessage text,
prevgeometry geometry(Geometry,4326),
newgeometry geometry(Geometry,4326)""",
- content=[(345, 'N', 'boundary', 'administrative',
- {'name': 'Foo'}, 'xx', self.recent,
- 'some text', None, None),
- (781, 'R', 'landuse', 'wood',
- None, 'ds', self.now,
- 'Area reduced by lots', None, None)])
-
+ content=[(345, 'N', 'boundary', 'administrative',
+ {'name': 'Foo'}, 'xx', self.recent,
+ 'some text', None, None),
+ (781, 'R', 'landuse', 'wood',
+ None, 'ds', self.now,
+ 'Area reduced by lots', None, None)])
@pytest.mark.asyncio
async def test_polygons_simple(self, api):
'errormessage': 'Area reduced by lots',
'updated': self.now.isoformat(sep=' ', timespec='seconds')}]
-
@pytest.mark.asyncio
async def test_polygons_days(self, api):
a = FakeAdaptor()
assert [r['osm_id'] for r in results] == [781]
-
@pytest.mark.asyncio
async def test_polygons_class(self, api):
a = FakeAdaptor()
assert [r['osm_id'] for r in results] == [781]
-
-
@pytest.mark.asyncio
async def test_polygons_reduced(self, api):
a = FakeAdaptor()
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for reverse API call.
API_OPTIONS = {'reverse'}
+
def test_reverse_rank_30(apiobj, frontend):
apiobj.add_placex(place_id=223, class_='place', type='house',
housenumber='1',
def test_reverse_street(apiobj, frontend, country):
apiobj.add_placex(place_id=990, class_='highway', type='service',
rank_search=27, rank_address=27,
- name = {'name': 'My Street'},
+ name={'name': 'My Street'},
centroid=(10.0, 10.0),
country_code=country,
geometry='LINESTRING(9.995 10, 10.005 10)')
assert result is None
-@pytest.mark.parametrize('y,layer,place_id', [(0.7, napi.DataLayer.ADDRESS, 223),
- (0.70001, napi.DataLayer.POI, 224),
- (0.7, napi.DataLayer.ADDRESS | napi.DataLayer.POI, 224),
- (0.70001, napi.DataLayer.ADDRESS | napi.DataLayer.POI, 223),
- (0.7, napi.DataLayer.MANMADE, 225),
- (0.7, napi.DataLayer.RAILWAY, 226),
- (0.7, napi.DataLayer.NATURAL, 227),
- (0.70003, napi.DataLayer.MANMADE | napi.DataLayer.RAILWAY, 225),
- (0.70003, napi.DataLayer.MANMADE | napi.DataLayer.NATURAL, 225),
- (5, napi.DataLayer.ADDRESS, 229)])
+@pytest.mark.parametrize('y,layer,place_id',
+ [(0.7, napi.DataLayer.ADDRESS, 223),
+ (0.70001, napi.DataLayer.POI, 224),
+ (0.7, napi.DataLayer.ADDRESS | napi.DataLayer.POI, 224),
+ (0.70001, napi.DataLayer.ADDRESS | napi.DataLayer.POI, 223),
+ (0.7, napi.DataLayer.MANMADE, 225),
+ (0.7, napi.DataLayer.RAILWAY, 226),
+ (0.7, napi.DataLayer.NATURAL, 227),
+ (0.70003, napi.DataLayer.MANMADE | napi.DataLayer.RAILWAY, 225),
+ (0.70003, napi.DataLayer.MANMADE | napi.DataLayer.NATURAL, 225),
+ (5, napi.DataLayer.ADDRESS, 229)])
def test_reverse_rank_30_layers(apiobj, frontend, y, layer, place_id):
apiobj.add_placex(place_id=223, osm_type='N', class_='place', type='house',
housenumber='1',
api = frontend(apiobj, options=API_OPTIONS)
assert api.reverse((1.3, 0.70001), max_rank=29,
- layers=napi.DataLayer.POI) is None
+ layers=napi.DataLayer.POI) is None
@pytest.mark.parametrize('with_geom', [True, False])
def test_reverse_housenumber_on_street(apiobj, frontend, with_geom):
apiobj.add_placex(place_id=990, class_='highway', type='service',
rank_search=27, rank_address=27,
- name = {'name': 'My Street'},
+ name={'name': 'My Street'},
centroid=(10.0, 10.0),
geometry='LINESTRING(9.995 10, 10.005 10)')
apiobj.add_placex(place_id=991, class_='place', type='house',
centroid=(10.0, 10.00001))
apiobj.add_placex(place_id=1990, class_='highway', type='service',
rank_search=27, rank_address=27,
- name = {'name': 'Other Street'},
+ name={'name': 'Other Street'},
centroid=(10.0, 1.0),
geometry='LINESTRING(9.995 1, 10.005 1)')
apiobj.add_placex(place_id=1991, class_='place', type='house',
def test_reverse_housenumber_interpolation(apiobj, frontend, with_geom):
apiobj.add_placex(place_id=990, class_='highway', type='service',
rank_search=27, rank_address=27,
- name = {'name': 'My Street'},
+ name={'name': 'My Street'},
centroid=(10.0, 10.0),
geometry='LINESTRING(9.995 10, 10.005 10)')
apiobj.add_placex(place_id=991, class_='place', type='house',
geometry='LINESTRING(9.995 10.00001, 10.005 10.00001)')
apiobj.add_placex(place_id=1990, class_='highway', type='service',
rank_search=27, rank_address=27,
- name = {'name': 'Other Street'},
+ name={'name': 'Other Street'},
centroid=(10.0, 20.0),
geometry='LINESTRING(9.995 20, 10.005 20)')
apiobj.add_osmline(place_id=1992,
def test_reverse_housenumber_point_interpolation(apiobj, frontend):
apiobj.add_placex(place_id=990, class_='highway', type='service',
rank_search=27, rank_address=27,
- name = {'name': 'My Street'},
+ name={'name': 'My Street'},
centroid=(10.0, 10.0),
geometry='LINESTRING(9.995 10, 10.005 10)')
apiobj.add_osmline(place_id=992,
def test_reverse_tiger_number(apiobj, frontend):
apiobj.add_placex(place_id=990, class_='highway', type='service',
rank_search=27, rank_address=27,
- name = {'name': 'My Street'},
+ name={'name': 'My Street'},
centroid=(10.0, 10.0),
country_code='us',
geometry='LINESTRING(9.995 10, 10.005 10)')
def test_reverse_point_tiger(apiobj, frontend):
apiobj.add_placex(place_id=990, class_='highway', type='service',
rank_search=27, rank_address=27,
- name = {'name': 'My Street'},
+ name={'name': 'My Street'},
centroid=(10.0, 10.0),
country_code='us',
geometry='LINESTRING(9.995 10, 10.005 10)')
geometry='LINESTRING(9.995 10.00001, 10.005 10.00001)')
api = frontend(apiobj, options=API_OPTIONS)
- assert api.reverse((10.0, 10.0), geometry_output=napi.GeometryFormat.TEXT)\
- .geometry['text'] == 'POINT(10 10.00001)'
+ result = api.reverse((10.0, 10.0), geometry_output=napi.GeometryFormat.TEXT)
+
+ assert result.geometry['text'] == 'POINT(10 10.00001)'
def test_reverse_tiger_geometry(apiobj, frontend):
apiobj.add_placex(place_id=990, class_='highway', type='service',
rank_search=27, rank_address=27,
- name = {'name': 'My Street'},
+ name={'name': 'My Street'},
centroid=(10.0, 10.0),
country_code='us',
geometry='LINESTRING(9.995 10, 10.005 10)')
geometry='LINESTRING(9.995 10.00001, 10.005 10.00001)')
apiobj.add_placex(place_id=1000, class_='highway', type='service',
rank_search=27, rank_address=27,
- name = {'name': 'My Street'},
+ name={'name': 'My Street'},
centroid=(11.0, 11.0),
country_code='us',
geometry='LINESTRING(10.995 11, 11.005 11)')
params = {'geometry_output': napi.GeometryFormat.GEOJSON}
output = api.reverse((10.0, 10.0), **params)
- assert json.loads(output.geometry['geojson']) == {'coordinates': [10, 10.00001], 'type': 'Point'}
+ assert json.loads(output.geometry['geojson']) \
+ == {'coordinates': [10, 10.00001], 'type': 'Point'}
output = api.reverse((11.0, 11.0), **params)
- assert json.loads(output.geometry['geojson']) == {'coordinates': [11, 11.00001], 'type': 'Point'}
-
+ assert json.loads(output.geometry['geojson']) \
+ == {'coordinates': [11, 11.00001], 'type': 'Point'}
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for search API calls.
These tests make sure that all Python code is correct and executable.
Functional tests can be found in the BDD test suite.
"""
-import json
-
import pytest
-import sqlalchemy as sa
-
-import nominatim_api as napi
import nominatim_api.logging as loglib
API_OPTIONS = {'search'}
+
@pytest.fixture(autouse=True)
def setup_icu_tokenizer(apiobj):
""" Setup the properties needed for using the ICU tokenizer.
apiobj.add_data('properties',
[{'property': 'tokenizer', 'value': 'icu'},
{'property': 'tokenizer_import_normalisation', 'value': ':: lower();'},
- {'property': 'tokenizer_import_transliteration', 'value': "'1' > '/1/'; 'ä' > 'ä '"},
- ])
+ {'property': 'tokenizer_import_transliteration',
+ 'value': "'1' > '/1/'; 'ä' > 'ä '"},
+ ])
def test_search_no_content(apiobj, frontend):
api = frontend(apiobj, options=API_OPTIONS)
loglib.set_log_output(logtype)
- results = api.search('TEST')
+ api.search('TEST')
assert loglib.get_and_disable()
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the status API call.
"""
import datetime as dt
-import pytest
-from nominatim_db.version import NominatimVersion
from nominatim_api.version import NOMINATIM_API_VERSION
import nominatim_api as napi
+
def test_status_no_extra_info(apiobj, frontend):
api = frontend(apiobj)
result = api.status()
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for loading of parameter dataclasses.
from nominatim_api.errors import UsageError
import nominatim_api.types as typ
+
def test_no_params_defaults():
params = typ.LookupDetails.from_kwargs({})
('geometry_simplification', 'NaN')])
def test_bad_format_reverse(k, v):
with pytest.raises(UsageError):
- params = typ.ReverseDetails.from_kwargs({k: v})
+ typ.ReverseDetails.from_kwargs({k: v})
@pytest.mark.parametrize('rin,rout', [(-23, 0), (0, 0), (1, 1),
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for export CLI function.
import nominatim_db.cli
+
@pytest.fixture
def run_export(tmp_path, capsys):
def _exec(args):
+ cli_args = ['export', '--project-dir', str(tmp_path)] + args
assert 0 == nominatim_db.cli.nominatim(osm2pgsql_path='OSM2PGSQL NOT AVAILABLE',
- cli_args=['export', '--project-dir', str(tmp_path)]
- + args)
+ cli_args=cli_args)
return capsys.readouterr().out.split('\r\n')
return _exec
@pytest.fixture(autouse=True)
def setup_database_with_context(apiobj):
apiobj.add_placex(place_id=332, osm_type='W', osm_id=4,
- class_='highway', type='residential', name='Street',
- country_code='pl', postcode='55674',
- rank_search=27, rank_address=26)
+ class_='highway', type='residential', name='Street',
+ country_code='pl', postcode='55674',
+ rank_search=27, rank_address=26)
apiobj.add_address_placex(332, fromarea=False, isaddress=False,
distance=0.0034,
place_id=1000, osm_type='N', osm_id=3333,
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the helper functions for v1 API.
import nominatim_api.v1.helpers as helper
+
@pytest.mark.parametrize('inp', ['',
'abc',
'12 23',
def test_extract_coords_with_text_after():
assert ('abc', 12.456, -78.90) == helper.extract_coords_from_query('-78.90, 12.456 abc')
+
@pytest.mark.parametrize('inp', [' [12.456,-78.90] ', ' 12.456,-78.90 '])
def test_extract_coords_with_spaces(inp):
assert ('', -78.90, 12.456) == helper.extract_coords_from_query(inp)
+
@pytest.mark.parametrize('inp', ['40 26.767 N 79 58.933 W',
- '40° 26.767′ N 79° 58.933′ W',
- "40° 26.767' N 79° 58.933' W",
- "40° 26.767'\n"
- " N 79° 58.933' W",
- 'N 40 26.767, W 79 58.933',
- 'N 40°26.767′, W 79°58.933′',
- ' N 40°26.767′, W 79°58.933′',
- "N 40°26.767', W 79°58.933'",
-
- '40 26 46 N 79 58 56 W',
- '40° 26′ 46″ N 79° 58′ 56″ W',
- '40° 26′ 46.00″ N 79° 58′ 56.00″ W',
- '40°26′46″N 79°58′56″W',
- 'N 40 26 46 W 79 58 56',
- 'N 40° 26′ 46″, W 79° 58′ 56″',
- 'N 40° 26\' 46", W 79° 58\' 56"',
- 'N 40° 26\' 46", W 79° 58\' 56"',
-
- '40.446 -79.982',
- '40.446,-79.982',
- '40.446° N 79.982° W',
- 'N 40.446° W 79.982°',
-
- '[40.446 -79.982]',
- '[40.446,\v-79.982]',
- ' 40.446 , -79.982 ',
- ' 40.446 , -79.982 ',
- ' 40.446 , -79.982 ',
- ' 40.446\v, -79.982 '])
+ '40° 26.767′ N 79° 58.933′ W',
+ "40° 26.767' N 79° 58.933' W",
+ "40° 26.767'\n"
+ " N 79° 58.933' W",
+ 'N 40 26.767, W 79 58.933',
+ 'N 40°26.767′, W 79°58.933′',
+ ' N 40°26.767′, W 79°58.933′',
+ "N 40°26.767', W 79°58.933'",
+
+ '40 26 46 N 79 58 56 W',
+ '40° 26′ 46″ N 79° 58′ 56″ W',
+ '40° 26′ 46.00″ N 79° 58′ 56.00″ W',
+ '40°26′46″N 79°58′56″W',
+ 'N 40 26 46 W 79 58 56',
+ 'N 40° 26′ 46″, W 79° 58′ 56″',
+ 'N 40° 26\' 46", W 79° 58\' 56"',
+ 'N 40° 26\' 46", W 79° 58\' 56"',
+
+ '40.446 -79.982',
+ '40.446,-79.982',
+ '40.446° N 79.982° W',
+ 'N 40.446° W 79.982°',
+
+ '[40.446 -79.982]',
+ '[40.446,\v-79.982]',
+ ' 40.446 , -79.982 ',
+ ' 40.446 , -79.982 ',
+ ' 40.446 , -79.982 ',
+ ' 40.446\v, -79.982 '])
def test_extract_coords_formats(inp):
query, x, y = helper.extract_coords_from_query(inp)
assert cls == 'shop'
assert typ == 'fish'
+
def test_extract_category_only():
assert helper.extract_category_from_query('[shop=market]') == ('', 'shop', 'market')
+
@pytest.mark.parametrize('inp', ['house []', 'nothing', '[352]'])
-def test_extract_category_no_match(inp):
+def test_extract_category_no_match(inp):
assert helper.extract_category_from_query(inp) == (inp, None, None)
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Test functions for adapting results to the user's locale.
from nominatim_api import Locales
+
def test_display_name_empty_names():
- l = Locales(['en', 'de'])
+ loc = Locales(['en', 'de'])
+
+ assert loc.display_name(None) == ''
+ assert loc.display_name({}) == ''
- assert l.display_name(None) == ''
- assert l.display_name({}) == ''
def test_display_name_none_localized():
- l = Locales()
+ loc = Locales()
- assert l.display_name({}) == ''
- assert l.display_name({'name:de': 'DE', 'name': 'ALL'}) == 'ALL'
- assert l.display_name({'ref': '34', 'name:de': 'DE'}) == '34'
+ assert loc.display_name({}) == ''
+ assert loc.display_name({'name:de': 'DE', 'name': 'ALL'}) == 'ALL'
+ assert loc.display_name({'ref': '34', 'name:de': 'DE'}) == '34'
def test_display_name_localized():
- l = Locales(['en', 'de'])
+ loc = Locales(['en', 'de'])
- assert l.display_name({}) == ''
- assert l.display_name({'name:de': 'DE', 'name': 'ALL'}) == 'DE'
- assert l.display_name({'ref': '34', 'name:de': 'DE'}) == 'DE'
+ assert loc.display_name({}) == ''
+ assert loc.display_name({'name:de': 'DE', 'name': 'ALL'}) == 'DE'
+ assert loc.display_name({'ref': '34', 'name:de': 'DE'}) == 'DE'
def test_display_name_preference():
- l = Locales(['en', 'de'])
+ loc = Locales(['en', 'de'])
- assert l.display_name({}) == ''
- assert l.display_name({'name:de': 'DE', 'name:en': 'EN'}) == 'EN'
- assert l.display_name({'official_name:en': 'EN', 'name:de': 'DE'}) == 'DE'
+ assert loc.display_name({}) == ''
+ assert loc.display_name({'name:de': 'DE', 'name:en': 'EN'}) == 'EN'
+ assert loc.display_name({'official_name:en': 'EN', 'name:de': 'DE'}) == 'DE'
@pytest.mark.parametrize('langstr,langlist',
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for formatting results for the V1 API.
# StatusResult
+
def test_status_format_list():
assert set(v1_format.list_formats(napi.StatusResult)) == STATUS_FORMATS
def test_status_format_text():
- assert v1_format.format_result(napi.StatusResult(0, 'message here'), 'text', {}) == 'OK'
+ assert v1_format.format_result(napi.StatusResult(0, 'message here'), 'text', {}) \
+ == 'OK'
-def test_status_format_text():
- assert v1_format.format_result(napi.StatusResult(500, 'message here'), 'text', {}) == 'ERROR: message here'
+def test_status_format_error_text():
+ assert v1_format.format_result(napi.StatusResult(500, 'message here'), 'text', {}) \
+ == 'ERROR: message here'
def test_status_format_json_minimal():
result = v1_format.format_result(status, 'json', {})
- assert result == \
- f'{{"status":700,"message":"Bad format.","software_version":"{napi.__version__}"}}'
+ assert json.loads(result) == {'status': 700,
+ 'message': 'Bad format.',
+ 'software_version': napi.__version__}
def test_status_format_json_full():
result = v1_format.format_result(status, 'json', {})
- assert result == \
- f'{{"status":0,"message":"OK","data_updated":"2010-02-07T20:20:03+00:00","software_version":"{napi.__version__}","database_version":"5.6"}}'
+ assert json.loads(result) == {'status': 0,
+ 'message': 'OK',
+ 'data_updated': '2010-02-07T20:20:03+00:00',
+ 'software_version': napi.__version__,
+ 'database_version': '5.6'}
# DetailedResult
'extratags': {},
'centroid': {'type': 'Point', 'coordinates': [1.0, 2.0]},
'geometry': {'type': 'Point', 'coordinates': [1.0, 2.0]},
- }
+ }
def test_search_details_full():
rank_search=28,
importance=0.0443,
country_code='ll',
- indexed_date = import_date
+ indexed_date=import_date
)
search.localize(napi.Locales())
'isarea': False,
'centroid': {'type': 'Point', 'coordinates': [56.947, -87.44]},
'geometry': {'type': 'Point', 'coordinates': [56.947, -87.44]},
- }
+ }
@pytest.mark.parametrize('gtype,isarea', [('ST_Point', False),
('ST_MultiPolygon', True)])
def test_search_details_no_geometry(gtype, isarea):
search = napi.DetailedResult(napi.SourceTable.PLACEX,
- ('place', 'thing'),
- napi.Point(1.0, 2.0),
- geometry={'type': gtype})
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0),
+ geometry={'type': gtype})
result = v1_format.format_result(search, 'json', {})
js = json.loads(result)
def test_search_details_with_geometry():
- search = napi.DetailedResult(napi.SourceTable.PLACEX,
- ('place', 'thing'),
- napi.Point(1.0, 2.0),
- geometry={'geojson': '{"type":"Point","coordinates":[56.947,-87.44]}'})
+ search = napi.DetailedResult(
+ napi.SourceTable.PLACEX,
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0),
+ geometry={'geojson': '{"type":"Point","coordinates":[56.947,-87.44]}'})
result = v1_format.format_result(search, 'json', {})
js = json.loads(result)
assert js['geometry'] == {'type': 'Point', 'coordinates': [56.947, -87.44]}
- assert js['isarea'] == False
+ assert js['isarea'] is False
def test_search_details_with_icon_available():
@pytest.mark.parametrize('field,outfield', [('address_rows', 'address'),
('linked_rows', 'linked_places'),
('parented_rows', 'hierarchy')
- ])
+ ])
def test_search_details_with_further_infos(field, outfield):
search = napi.DetailedResult(napi.SourceTable.PLACEX,
('place', 'thing'),
js = json.loads(result)
assert js[outfield] == [{'localname': 'Trespass',
- 'place_id': 3498,
- 'osm_id': 442,
- 'osm_type': 'R',
- 'place_type': 'spec',
- 'class': 'bnd',
- 'type': 'note',
- 'admin_level': 4,
- 'rank_address': 10,
- 'distance': 0.034,
- 'isaddress': True}]
+ 'place_id': 3498,
+ 'osm_id': 442,
+ 'osm_type': 'R',
+ 'place_type': 'spec',
+ 'class': 'bnd',
+ 'type': 'note',
+ 'admin_level': 4,
+ 'rank_address': 10,
+ 'distance': 0.034,
+ 'isaddress': True}]
def test_search_details_grouped_hierarchy():
search = napi.DetailedResult(napi.SourceTable.PLACEX,
('place', 'thing'),
napi.Point(1.0, 2.0),
- parented_rows =
- [napi.AddressLine(place_id=3498,
- osm_object=('R', 442),
- category=('bnd', 'note'),
- names={'name': 'Trespass'},
- extratags={'access': 'no',
- 'place_type': 'spec'},
- admin_level=4,
- fromarea=True,
- isaddress=True,
- rank_address=10,
- distance=0.034)
- ])
+ parented_rows=[napi.AddressLine(
+ place_id=3498,
+ osm_object=('R', 442),
+ category=('bnd', 'note'),
+ names={'name': 'Trespass'},
+ extratags={'access': 'no',
+ 'place_type': 'spec'},
+ admin_level=4,
+ fromarea=True,
+ isaddress=True,
+ rank_address=10,
+ distance=0.034)])
result = v1_format.format_result(search, 'json', {'group_hierarchy': True})
js = json.loads(result)
assert js['hierarchy'] == {'note': [{'localname': 'Trespass',
- 'place_id': 3498,
- 'osm_id': 442,
- 'osm_type': 'R',
- 'place_type': 'spec',
- 'class': 'bnd',
- 'type': 'note',
- 'admin_level': 4,
- 'rank_address': 10,
- 'distance': 0.034,
- 'isaddress': True}]}
+ 'place_id': 3498,
+ 'osm_id': 442,
+ 'osm_type': 'R',
+ 'place_type': 'spec',
+ 'class': 'bnd',
+ 'type': 'note',
+ 'admin_level': 4,
+ 'rank_address': 10,
+ 'distance': 0.034,
+ 'isaddress': True}]}
def test_search_details_keywords_name():
js = json.loads(result)
assert js['keywords'] == {'name': [{'id': 23, 'token': 'foo'},
- {'id': 24, 'token': 'foo'}],
+ {'id': 24, 'token': 'foo'}],
'address': []}
js = json.loads(result)
assert js['keywords'] == {'address': [{'id': 23, 'token': 'foo'},
- {'id': 24, 'token': 'foo'}],
+ {'id': 24, 'token': 'foo'}],
'name': []}
-
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for formatting reverse results for the V1 API.
FORMATS = ['json', 'jsonv2', 'geojson', 'geocodejson', 'xml']
+
@pytest.mark.parametrize('fmt', FORMATS)
def test_format_reverse_minimal(fmt):
reverse = napi.ReverseResult(napi.SourceTable.PLACEX,
reverse.localize(napi.Locales())
raw = v1_format.format_result(napi.ReverseResults([reverse]), fmt,
- {'addressdetails': True})
-
+ {'addressdetails': True})
if fmt == 'xml':
root = ET.fromstring(raw)
reverse.localize(napi.Locales())
raw = v1_format.format_result(napi.ReverseResults([reverse]), 'geocodejson',
- {'addressdetails': True})
+ {'addressdetails': True})
props = json.loads(raw)['features'][0]['properties']['geocoding']
assert props['housenumber'] == '1'
address_rows=napi.AddressLines())
raw = v1_format.format_result(napi.ReverseResults([reverse]), fmt,
- {'addressdetails': True})
-
+ {'addressdetails': True})
if fmt == 'xml':
root = ET.fromstring(raw)
reverse = napi.ReverseResult(napi.SourceTable.PLACEX,
('place', 'thing'),
napi.Point(1.0, 2.0),
- extratags={'one': 'A', 'two':'B'})
+ extratags={'one': 'A', 'two': 'B'})
raw = v1_format.format_result(napi.ReverseResults([reverse]), fmt,
- {'extratags': True})
+ {'extratags': True})
if fmt == 'xml':
root = ET.fromstring(raw)
else:
extra = result['extratags']
- assert extra == {'one': 'A', 'two':'B'}
+ assert extra == {'one': 'A', 'two': 'B'}
@pytest.mark.parametrize('fmt', ['json', 'jsonv2', 'geojson', 'xml'])
napi.Point(1.0, 2.0))
raw = v1_format.format_result(napi.ReverseResults([reverse]), fmt,
- {'extratags': True})
+ {'extratags': True})
if fmt == 'xml':
root = ET.fromstring(raw)
reverse = napi.ReverseResult(napi.SourceTable.PLACEX,
('place', 'thing'),
napi.Point(1.0, 2.0),
- names={'name': 'A', 'ref':'1'})
+ names={'name': 'A', 'ref': '1'})
raw = v1_format.format_result(napi.ReverseResults([reverse]), fmt,
- {'namedetails': True})
+ {'namedetails': True})
if fmt == 'xml':
root = ET.fromstring(raw)
else:
extra = result['namedetails']
- assert extra == {'name': 'A', 'ref':'1'}
+ assert extra == {'name': 'A', 'ref': '1'}
@pytest.mark.parametrize('fmt', ['json', 'jsonv2', 'geojson', 'xml'])
napi.Point(1.0, 2.0))
raw = v1_format.format_result(napi.ReverseResults([reverse]), fmt,
- {'namedetails': True})
+ {'namedetails': True})
if fmt == 'xml':
root = ET.fromstring(raw)
napi.Point(1.0, 2.0))
result = v1_format.format_result(napi.ReverseResults([reverse]), fmt,
- {'icon_base_url': 'foo'})
+ {'icon_base_url': 'foo'})
js = json.loads(result)
napi.Point(1.0, 2.0))
result = v1_format.format_result(napi.ReverseResults([reverse]), fmt,
- {'icon_base_url': 'foo'})
+ {'icon_base_url': 'foo'})
assert 'icon' not in json.loads(result)
-
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for result datatype helper functions.
from binascii import hexlify
import pytest
-import pytest_asyncio
-import sqlalchemy as sa
-
from nominatim_api import SourceTable, DetailedResult, Point
import nominatim_api.results as nresults
+
def mkpoint(x, y):
return hexlify(struct.pack("=biidd", 1, 0x20000001, 4326, x, y)).decode('utf-8')
+
class FakeRow:
def __init__(self, **kwargs):
if 'parent_place_id' not in kwargs:
assert res.lat == 0.5
assert res.calculated_importance() == pytest.approx(0.00001)
+
def test_detailed_result_custom_importance():
res = DetailedResult(SourceTable.PLACEX,
('amenity', 'post_box'),
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the Python web frameworks adaptor, v1 API.
return excinfo.value
-
def test_without_content_set(self):
err = self.run_raise_error('TEST', 404)
assert err.msg == 'ERROR 404: TEST'
assert err.status == 404
-
def test_json(self):
self.adaptor.content_type = 'application/json; charset=utf-8'
assert content['code'] == 501
assert content['message'] == 'TEST'
-
def test_xml(self):
self.adaptor.content_type = 'text/xml; charset=utf-8'
monkeypatch.setattr(napi.NominatimAPIAsync, 'status', _status)
-
@pytest.mark.asyncio
async def test_status_without_params(self):
a = FakeAdaptor()
assert resp.status == 200
assert resp.content_type == 'text/plain; charset=utf-8'
-
@pytest.mark.asyncio
async def test_status_with_error(self):
a = FakeAdaptor()
assert resp.status == 500
assert resp.content_type == 'text/plain; charset=utf-8'
-
@pytest.mark.asyncio
async def test_status_json_with_error(self):
a = FakeAdaptor(params={'format': 'json'})
assert resp.status == 200
assert resp.content_type == 'application/json; charset=utf-8'
-
@pytest.mark.asyncio
async def test_status_bad_format(self):
a = FakeAdaptor(params={'format': 'foo'})
monkeypatch.setattr(napi.NominatimAPIAsync, 'details', _lookup)
-
@pytest.mark.asyncio
async def test_details_no_params(self):
a = FakeAdaptor()
with pytest.raises(FakeError, match='^400 -- .*Missing'):
await glue.details_endpoint(napi.NominatimAPIAsync(), a)
-
@pytest.mark.asyncio
async def test_details_by_place_id(self):
a = FakeAdaptor(params={'place_id': '4573'})
assert self.lookup_args[0].place_id == 4573
-
@pytest.mark.asyncio
async def test_details_by_osm_id(self):
a = FakeAdaptor(params={'osmtype': 'N', 'osmid': '45'})
assert self.lookup_args[0].osm_id == 45
assert self.lookup_args[0].osm_class is None
-
@pytest.mark.asyncio
async def test_details_with_debugging(self):
a = FakeAdaptor(params={'osmtype': 'N', 'osmid': '45', 'debug': '1'})
assert resp.content_type == 'text/html; charset=utf-8'
assert content.tag == 'html'
-
@pytest.mark.asyncio
async def test_details_no_result(self):
a = FakeAdaptor(params={'place_id': '4573'})
@pytest.fixture(autouse=True)
def patch_reverse_func(self, monkeypatch):
self.result = napi.ReverseResult(napi.SourceTable.PLACEX,
- ('place', 'thing'),
- napi.Point(1.0, 2.0))
+ ('place', 'thing'),
+ napi.Point(1.0, 2.0))
+
async def _reverse(*args, **kwargs):
return self.result
monkeypatch.setattr(napi.NominatimAPIAsync, 'reverse', _reverse)
-
@pytest.mark.asyncio
@pytest.mark.parametrize('params', [{}, {'lat': '3.4'}, {'lon': '6.7'}])
async def test_reverse_no_params(self, params):
with pytest.raises(FakeError, match='^400 -- (?s:.*)missing'):
await glue.reverse_endpoint(napi.NominatimAPIAsync(), a)
-
- @pytest.mark.asyncio
- @pytest.mark.parametrize('params', [{'lat': '45.6', 'lon': '4563'}])
- async def test_reverse_success(self, params):
- a = FakeAdaptor()
- a.params = params
- a.params['format'] = 'json'
-
- res = await glue.reverse_endpoint(napi.NominatimAPIAsync(), a)
-
- assert res == ''
-
-
@pytest.mark.asyncio
async def test_reverse_success(self):
a = FakeAdaptor()
assert await glue.reverse_endpoint(napi.NominatimAPIAsync(), a)
-
@pytest.mark.asyncio
async def test_reverse_from_search(self):
a = FakeAdaptor()
self.results = [napi.SearchResult(napi.SourceTable.PLACEX,
('place', 'thing'),
napi.Point(1.0, 2.0))]
+
async def _lookup(*args, **kwargs):
return napi.SearchResults(self.results)
monkeypatch.setattr(napi.NominatimAPIAsync, 'lookup', _lookup)
-
@pytest.mark.asyncio
async def test_lookup_no_params(self):
a = FakeAdaptor()
assert res.output == '[]'
-
@pytest.mark.asyncio
@pytest.mark.parametrize('param', ['w', 'bad', ''])
async def test_lookup_bad_params(self, param):
assert len(json.loads(res.output)) == 1
-
@pytest.mark.asyncio
@pytest.mark.parametrize('param', ['p234234', '4563'])
async def test_lookup_bad_osm_type(self, param):
assert len(json.loads(res.output)) == 1
-
@pytest.mark.asyncio
async def test_lookup_working(self):
a = FakeAdaptor()
self.results = [napi.SearchResult(napi.SourceTable.PLACEX,
('place', 'thing'),
napi.Point(1.0, 2.0))]
+
async def _search(*args, **kwargs):
return napi.SearchResults(self.results)
monkeypatch.setattr(napi.NominatimAPIAsync, 'search', _search)
-
@pytest.mark.asyncio
async def test_search_free_text(self):
a = FakeAdaptor()
assert len(json.loads(res.output)) == 1
-
@pytest.mark.asyncio
async def test_search_free_text_xml(self):
a = FakeAdaptor()
assert res.status == 200
assert res.output.index('something') > 0
-
@pytest.mark.asyncio
async def test_search_free_and_structured(self):
a = FakeAdaptor()
a.params['city'] = 'ignored'
with pytest.raises(FakeError, match='^400 -- .*cannot be used together'):
- res = await glue.search_endpoint(napi.NominatimAPIAsync(), a)
-
+ await glue.search_endpoint(napi.NominatimAPIAsync(), a)
@pytest.mark.asyncio
@pytest.mark.parametrize('dedupe,numres', [(True, 1), (False, 2)])
self.results = [napi.SearchResult(napi.SourceTable.PLACEX,
('place', 'thing'),
napi.Point(1.0, 2.0))]
+
async def _search(*args, **kwargs):
return napi.SearchResults(self.results)
monkeypatch.setattr(napi.NominatimAPIAsync, 'search_address', _search)
-
@pytest.mark.asyncio
async def test_search_structured(self):
a = FakeAdaptor()
self.results = [napi.SearchResult(napi.SourceTable.PLACEX,
('place', 'thing'),
napi.Point(1.0, 2.0))]
+
async def _search(*args, **kwargs):
return napi.SearchResults(self.results)
monkeypatch.setattr(napi.NominatimAPIAsync, 'search_category', _search)
-
@pytest.mark.asyncio
async def test_search_category(self):
a = FakeAdaptor()
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for warm-up CLI function.
import nominatim_db.cli
+
@pytest.fixture(autouse=True)
def setup_database_with_context(apiobj, table_factory):
table_factory('word',
apiobj.add_data('properties',
[{'property': 'tokenizer', 'value': 'icu'},
{'property': 'tokenizer_import_normalisation', 'value': ':: lower();'},
- {'property': 'tokenizer_import_transliteration', 'value': "'1' > '/1/'; 'ä' > 'ä '"},
- ])
+ {'property': 'tokenizer_import_transliteration',
+ 'value': "'1' > '/1/'; 'ä' > 'ä '"}
+ ])
@pytest.mark.parametrize('args', [['--search-only'], ['--reverse-only']])
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
import pytest
import nominatim_db.cli
+
class MockParamCapture:
""" Mock that records the parameters with which a function was called
as well as the number of calls.
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for command line interface wrapper.
correct functionality. They use a lot of monkeypatching to avoid executing
the actual functions.
"""
-import importlib
import pytest
import nominatim_db.indexer.indexer
captured = capsys.readouterr()
assert captured.out.startswith('usage:')
+
def test_cli_version(cli_call, capsys):
""" Running nominatim tool --version prints a version string.
"""
# Make sure tools.freeze.is_frozen doesn't report database as frozen. Monkeypatching failed
table_factory('place')
-
@pytest.mark.parametrize("name,oid", [('file', 'foo.osm'), ('diff', 'foo.osc')])
def test_cli_add_data_file_command(self, cli_call, mock_func_factory, name, oid):
mock_run_legacy = mock_func_factory(nominatim_db.tools.add_osm_data, 'add_data_from_file')
assert mock_run_legacy.called == 1
-
@pytest.mark.parametrize("name,oid", [('node', 12), ('way', 8), ('relation', 32)])
def test_cli_add_data_object_command(self, cli_call, mock_func_factory, name, oid):
mock_run_legacy = mock_func_factory(nominatim_db.tools.add_osm_data, 'add_osm_object')
assert mock_run_legacy.called == 1
-
-
def test_cli_add_data_tiger_data(self, cli_call, cli_tokenizer_mock, async_mock_func_factory):
mock = async_mock_func_factory(nominatim_db.tools.tiger_data, 'add_tiger_data')
assert mock_drop.called == 1
assert mock_flatnode.called == 1
-
@pytest.mark.parametrize("params,do_bnds,do_ranks", [
([], 2, 2),
(['--boundaries-only'], 2, 0),
def test_index_command(self, monkeypatch, async_mock_func_factory, table_factory,
params, do_bnds, do_ranks):
table_factory('import_status', 'indexed bool')
- bnd_mock = async_mock_func_factory(nominatim_db.indexer.indexer.Indexer, 'index_boundaries')
- rank_mock = async_mock_func_factory(nominatim_db.indexer.indexer.Indexer, 'index_by_rank')
- postcode_mock = async_mock_func_factory(nominatim_db.indexer.indexer.Indexer, 'index_postcodes')
-
- monkeypatch.setattr(nominatim_db.indexer.indexer.Indexer, 'has_pending',
+ bnd_mock = async_mock_func_factory(nominatim_db.indexer.indexer.Indexer,
+ 'index_boundaries')
+ rank_mock = async_mock_func_factory(nominatim_db.indexer.indexer.Indexer,
+ 'index_by_rank')
+ postcode_mock = async_mock_func_factory(nominatim_db.indexer.indexer.Indexer,
+ 'index_postcodes')
+
+ monkeypatch.setattr(nominatim_db.indexer.indexer.Indexer, 'has_pending',
[False, True].pop)
assert self.call_nominatim('index', *params) == 0
assert rank_mock.called == do_ranks
assert postcode_mock.called == do_ranks
-
def test_special_phrases_wiki_command(self, mock_func_factory):
func = mock_func_factory(nominatim_db.clicmd.special_phrases.SPImporter, 'import_phrases')
assert func.called == 1
-
def test_special_phrases_csv_command(self, src_dir, mock_func_factory):
func = mock_func_factory(nominatim_db.clicmd.special_phrases.SPImporter, 'import_phrases')
testdata = src_dir / 'test' / 'testdb'
assert func.called == 1
-
def test_special_phrases_csv_bad_file(self, src_dir):
testdata = src_dir / 'something349053905.csv'
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Test for the command line interface wrapper admin subcommand.
assert cli_call('admin', '--clean-deleted', '1 month') == 0
assert mock.called == 1
+
def test_admin_clean_deleted_relations_no_age(cli_call, mock_func_factory):
- mock = mock_func_factory(nominatim_db.tools.admin, 'clean_deleted_relations')
+ mock_func_factory(nominatim_db.tools.admin, 'clean_deleted_relations')
assert cli_call('admin', '--clean-deleted') == 1
+
class TestCliAdminWithDb:
@pytest.fixture(autouse=True)
self.call_nominatim = cli_call
self.tokenizer_mock = cli_tokenizer_mock
-
@pytest.mark.parametrize("func, params", [('analyse_indexing', ('--analyse-indexing', ))])
def test_analyse_indexing(self, mock_func_factory, func, params):
mock = mock_func_factory(nominatim_db.tools.admin, func)
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for API access commands of command-line interface wrapper.
import json
import pytest
-import nominatim_db.clicmd.api
import nominatim_api as napi
+
@pytest.mark.parametrize('call', ['search', 'reverse', 'lookup', 'details', 'status'])
def test_list_format(cli_call, call):
assert 0 == cli_call(call, '--list-formats')
monkeypatch.setattr(napi.NominatimAPI, 'status',
lambda self: napi.StatusResult(200, 'OK'))
-
def test_status_simple(self, cli_call, tmp_path):
result = cli_call('status', '--project-dir', str(tmp_path))
assert result == 0
-
def test_status_json_format(self, cli_call, tmp_path, capsys):
result = cli_call('status', '--project-dir', str(tmp_path),
'--format', 'json')
('--way', '1'),
('--relation', '1'),
('--place_id', '10001')])
-
def test_details_json_format(self, cli_call, tmp_path, capsys, params):
result = cli_call('details', '--project-dir', str(tmp_path), *params)
def setup_reverse_mock(self, monkeypatch):
result = napi.ReverseResult(napi.SourceTable.PLACEX, ('place', 'thing'),
napi.Point(1.0, -3.0),
- names={'name':'Name', 'name:fr': 'Nom'},
- extratags={'extra':'Extra'},
+ names={'name': 'Name', 'name:fr': 'Nom'},
+ extratags={'extra': 'Extra'},
locale_name='Name',
display_name='Name')
monkeypatch.setattr(napi.NominatimAPI, 'reverse',
lambda *args, **kwargs: result)
-
def test_reverse_simple(self, cli_call, tmp_path, capsys):
result = cli_call('reverse', '--project-dir', str(tmp_path),
'--lat', '34', '--lon', '34')
assert 'extratags' not in out
assert 'namedetails' not in out
-
@pytest.mark.parametrize('param,field', [('--addressdetails', 'address'),
('--extratags', 'extratags'),
('--namedetails', 'namedetails')])
out = json.loads(capsys.readouterr().out)
assert field in out
-
def test_reverse_format(self, cli_call, tmp_path, capsys):
result = cli_call('reverse', '--project-dir', str(tmp_path),
'--lat', '34', '--lon', '34', '--format', 'geojson')
@pytest.fixture(autouse=True)
def setup_lookup_mock(self, monkeypatch):
result = napi.SearchResult(napi.SourceTable.PLACEX, ('place', 'thing'),
- napi.Point(1.0, -3.0),
- names={'name':'Name', 'name:fr': 'Nom'},
- extratags={'extra':'Extra'},
- locale_name='Name',
- display_name='Name')
+ napi.Point(1.0, -3.0),
+ names={'name': 'Name', 'name:fr': 'Nom'},
+ extratags={'extra': 'Extra'},
+ locale_name='Name',
+ display_name='Name')
monkeypatch.setattr(napi.NominatimAPI, 'lookup',
lambda *args, **kwargs: napi.SearchResults([result]))
@pytest.mark.parametrize('endpoint, params', [('search', ('--query', 'Berlin')),
('search_address', ('--city', 'Berlin'))
- ])
+ ])
def test_search(cli_call, tmp_path, capsys, monkeypatch, endpoint, params):
result = napi.SearchResult(napi.SourceTable.PLACEX, ('place', 'thing'),
napi.Point(1.0, -3.0),
- names={'name':'Name', 'name:fr': 'Nom'},
- extratags={'extra':'Extra'},
+ names={'name': 'Name', 'name:fr': 'Nom'},
+ extratags={'extra': 'Extra'},
locale_name='Name',
display_name='Name')
monkeypatch.setattr(napi.NominatimAPI, endpoint,
lambda *args, **kwargs: napi.SearchResults([result]))
-
result = cli_call('search', '--project-dir', str(tmp_path), *params)
assert result == 0
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for import command of the command-line interface wrapper.
self.call_nominatim = cli_call
self.tokenizer_mock = cli_tokenizer_mock
-
def test_import_missing_file(self):
assert self.call_nominatim('import', '--osm-file', 'sfsafegwedgw.reh.erh') == 1
-
def test_import_bad_file(self):
assert self.call_nominatim('import', '--osm-file', '.') == 1
-
@pytest.mark.parametrize('with_updates', [True, False])
def test_import_full(self, mock_func_factory, async_mock_func_factory,
with_updates, place_table, property_table):
cf_mock = mock_func_factory(nominatim_db.tools.refresh, 'create_functions')
-
assert self.call_nominatim(*params) == 0
assert self.tokenizer_mock.finalize_import_called
for mock in mocks:
assert mock.called == 1, "Mock '{}' not called".format(mock.func_name)
-
def test_import_continue_load_data(self, mock_func_factory, async_mock_func_factory):
mocks = [
mock_func_factory(nominatim_db.tools.database_import, 'truncate_data_tables'),
for mock in mocks:
assert mock.called == 1, "Mock '{}' not called".format(mock.func_name)
-
def test_import_continue_indexing(self, mock_func_factory, async_mock_func_factory,
placex_table, temp_db_conn):
mocks = [
# Calling it again still works for the index
assert self.call_nominatim('import', '--continue', 'indexing') == 0
-
def test_import_continue_postprocess(self, mock_func_factory, async_mock_func_factory):
mocks = [
async_mock_func_factory(nominatim_db.tools.database_import, 'create_search_indices'),
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for command line interface wrapper for refresk command.
import nominatim_db.tools.postcodes
import nominatim_db.indexer.indexer
+
class TestRefresh:
@pytest.fixture(autouse=True)
self.call_nominatim = cli_call
self.tokenizer_mock = cli_tokenizer_mock
-
@pytest.mark.parametrize("command,func", [
('address-levels', 'load_address_levels_from_config'),
('wiki-data', 'import_wikipedia_articles'),
assert self.call_nominatim('refresh', '--' + command) == 0
assert func_mock.called == 1
-
def test_refresh_word_count(self):
assert self.call_nominatim('refresh', '--word-count') == 0
assert self.tokenizer_mock.update_statistics_called
-
def test_refresh_word_tokens(self):
assert self.call_nominatim('refresh', '--word-tokens') == 0
assert self.tokenizer_mock.update_word_tokens_called
-
def test_refresh_postcodes(self, async_mock_func_factory, mock_func_factory, place_table):
func_mock = mock_func_factory(nominatim_db.tools.postcodes, 'update_postcodes')
idx_mock = async_mock_func_factory(nominatim_db.indexer.indexer.Indexer, 'index_postcodes')
assert func_mock.called == 1
assert idx_mock.called == 1
-
def test_refresh_postcodes_no_place_table(self):
# Do nothing without the place table
assert self.call_nominatim('refresh', '--postcodes') == 0
-
def test_refresh_create_functions(self, mock_func_factory):
func_mock = mock_func_factory(nominatim_db.tools.refresh, 'create_functions')
assert func_mock.called == 1
assert self.tokenizer_mock.update_sql_functions_called
-
def test_refresh_wikidata_file_not_found(self, monkeypatch):
monkeypatch.setenv('NOMINATIM_WIKIPEDIA_DATA_PATH', 'gjoiergjeroi345Q')
assert self.call_nominatim('refresh', '--wiki-data') == 1
-
def test_refresh_secondary_importance_file_not_found(self):
assert self.call_nominatim('refresh', '--secondary-importance') == 1
-
def test_refresh_secondary_importance_new_table(self, mock_func_factory):
mocks = [mock_func_factory(nominatim_db.tools.refresh, 'import_secondary_importance'),
mock_func_factory(nominatim_db.tools.refresh, 'create_functions')]
assert mocks[0].called == 1
assert mocks[1].called == 1
-
def test_refresh_importance_computed_after_wiki_import(self, monkeypatch, mock_func_factory):
calls = []
monkeypatch.setattr(nominatim_db.tools.refresh, 'import_wikipedia_articles',
('--data-object', 'N23', '--data-object', 'N24'),
('--data-area', 'R7723'),
('--data-area', 'r7723', '--data-area', 'r2'),
- ('--data-area', 'R9284425', '--data-object', 'n1234567894567')])
+ ('--data-area', 'R9284425',
+ '--data-object', 'n1234567894567')])
def test_refresh_objects(self, params, mock_func_factory):
func_mock = mock_func_factory(nominatim_db.tools.refresh, 'invalidate_osm_object')
assert func_mock.called == len(params)/2
-
@pytest.mark.parametrize('func', ('--data-object', '--data-area'))
@pytest.mark.parametrize('param', ('234', 'a55', 'R 453', 'Rel'))
def test_refresh_objects_bad_param(self, func, param, mock_func_factory):
import nominatim_db.tools.refresh
from nominatim_db.db import status
+
@pytest.fixture
def tokenizer_mock(monkeypatch):
class DummyTokenizer:
return tok
-
@pytest.fixture
def init_status(temp_db_conn, status_table):
status.set_status(temp_db_conn, date=dt.datetime.now(dt.timezone.utc), seq=1)
def setup_cli_call(self, cli_call, temp_db):
self.call_nominatim = lambda *args: cli_call('replication', *args)
-
@pytest.fixture(autouse=True)
def setup_update_function(self, monkeypatch):
def _mock_updates(states):
monkeypatch.setattr(nominatim_db.tools.replication, 'update',
- lambda *args, **kwargs: states.pop())
+ lambda *args, **kwargs: states.pop())
self.update_states = _mock_updates
-
@pytest.mark.parametrize("params,func", [
(('--init',), 'init_replication'),
(('--init', '--no-update-functions'), 'init_replication'),
if params == ('--init',):
assert umock.called == 1
-
def test_replication_update_bad_interval(self, monkeypatch):
monkeypatch.setenv('NOMINATIM_REPLICATION_UPDATE_INTERVAL', 'xx')
assert self.call_nominatim() == 1
-
def test_replication_update_bad_interval_for_geofabrik(self, monkeypatch):
monkeypatch.setenv('NOMINATIM_REPLICATION_URL',
'https://download.geofabrik.de/europe/italy-updates')
assert self.call_nominatim() == 1
-
def test_replication_update_continuous_no_index(self):
assert self.call_nominatim('--no-index') == 1
assert str(update_mock.last_args[1]['osm2pgsql']).endswith('OSM2PGSQL NOT AVAILABLE')
-
def test_replication_update_custom_osm2pgsql(self, monkeypatch, update_mock):
monkeypatch.setenv('NOMINATIM_OSM2PGSQL_BINARY', '/secret/osm2pgsql')
assert self.call_nominatim('--once', '--no-index') == 0
assert str(update_mock.last_args[1]['osm2pgsql']) == '/secret/osm2pgsql'
-
@pytest.mark.parametrize("update_interval", [60, 3600])
def test_replication_catchup(self, placex_table, monkeypatch, index_mock, update_interval):
monkeypatch.setenv('NOMINATIM_REPLICATION_UPDATE_INTERVAL', str(update_interval))
assert self.call_nominatim('--catch-up') == 0
-
def test_replication_update_custom_threads(self, update_mock):
assert self.call_nominatim('--once', '--no-index', '--threads', '4') == 0
assert update_mock.last_args[1]['threads'] == 4
-
def test_replication_update_continuous(self, index_mock):
self.update_states([nominatim_db.tools.replication.UpdateState.UP_TO_DATE,
nominatim_db.tools.replication.UpdateState.UP_TO_DATE])
assert index_mock.called == 2
-
def test_replication_update_continuous_no_change(self, mock_func_factory,
index_mock):
self.update_states([nominatim_db.tools.replication.UpdateState.NO_CHANGES,
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Test for loading dotenv configuration.
from nominatim_db.config import Configuration, flatten_config_list
from nominatim_db.errors import UsageError
+
@pytest.fixture
def make_config():
""" Create a configuration object from the given project directory.
return _mk_config
+
@pytest.fixture
def make_config_path(tmp_path):
""" Create a configuration object with project and config directories
@pytest.mark.parametrize("val,expect", [('foo bar', "'foo bar'"),
("xy'z", "xy\\'z"),
- ])
+ ])
def test_get_libpq_dsn_convert_php_special_chars(make_config, monkeypatch, val, expect):
config = make_config()
assert config.get_bool('FOOBAR') == result
+
def test_get_bool_empty(make_config):
config = make_config()
(config.config_dir / 'test.yaml').write_text('cow: muh\ncat: miau\n')
with pytest.raises(UsageError, match='Config file not found.'):
- rules = config.load_sub_configuration('test.yaml', config='MY_CONFIG')
+ config.load_sub_configuration('test.yaml', config='MY_CONFIG')
@pytest.mark.parametrize("location", ['project_dir', 'config_dir'])
(config.config_dir / 'test.yaml').write_text('cow: muh\ncat: miau\n')
with pytest.raises(UsageError, match='Config file not found.'):
- rules = config.load_sub_configuration('test.yaml', config='MY_CONFIG')
+ config.load_sub_configuration('test.yaml', config='MY_CONFIG')
def test_load_subconf_json(make_config_path):
assert rules == dict(cow='muh', cat='miau')
+
def test_load_subconf_not_found(make_config_path):
config = make_config_path()
config = make_config_path()
testfile = config.config_dir / 'test.yaml'
- testfile.write_text(f'base: !include inc.yaml\n')
+ testfile.write_text('base: !include inc.yaml\n')
(getattr(config, location) / 'inc.yaml').write_text('first: 1\nsecond: 2\n')
rules = config.load_sub_configuration('test.yaml')
config = make_config_path()
testfile = config.config_dir / 'test.yaml'
- testfile.write_text(f'base: !include inc.txt\n')
+ testfile.write_text('base: !include inc.txt\n')
(config.config_dir / 'inc.txt').write_text('first: 1\nsecond: 2\n')
with pytest.raises(UsageError, match='Cannot handle config file format.'):
- rules = config.load_sub_configuration('test.yaml')
+ config.load_sub_configuration('test.yaml')
def test_load_subconf_include_not_found(make_config_path):
config = make_config_path()
testfile = config.config_dir / 'test.yaml'
- testfile.write_text(f'base: !include inc.txt\n')
+ testfile.write_text('base: !include inc.txt\n')
with pytest.raises(UsageError, match='Config file not found.'):
- rules = config.load_sub_configuration('test.yaml')
+ config.load_sub_configuration('test.yaml')
def test_load_subconf_include_recursive(make_config_path):
config = make_config_path()
testfile = config.config_dir / 'test.yaml'
- testfile.write_text(f'base: !include inc.yaml\n')
+ testfile.write_text('base: !include inc.yaml\n')
(config.config_dir / 'inc.yaml').write_text('- !include more.yaml\n- upper\n')
(config.config_dir / 'more.yaml').write_text('- the end\n')
[[2, 3], [45, [56, 78], 66]],
'end'
]
+
assert flatten_config_list(content) == \
- [34, {'first': '1st', 'second': '2nd'}, {},
- 2, 3, 45, 56, 78, 66, 'end']
+ [34, {'first': '1st', 'second': '2nd'}, {}, 2, 3, 45, 56, 78, 66, 'end']
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Test for loading extra Python modules.
"""
-from pathlib import Path
import sys
import pytest
from nominatim_db.config import Configuration
+
@pytest.fixture
def test_config(src_dir, tmp_path):
""" Create a configuration object with project and config directories
assert isinstance(module.NOMINATIM_VERSION, tuple)
+
def test_load_default_module_with_hyphen(test_config):
module = test_config.load_plugin_module('place-info', 'nominatim_db.data')
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
import itertools
import sys
return temp_db
+
@pytest.fixture
def temp_db_conn(temp_db):
""" Connection to the test database.
if content:
sql = pysql.SQL("INSERT INTO {} VALUES ({})")\
.format(pysql.Identifier(name),
- pysql.SQL(',').join([pysql.Placeholder() for _ in range(len(content[0]))]))
- cur.executemany(sql , content)
+ pysql.SQL(',').join([pysql.Placeholder()
+ for _ in range(len(content[0]))]))
+ cur.executemany(sql, content)
return mk_table
return _insert
+
@pytest.fixture
def placex_table(temp_db_with_extensions, temp_db_conn):
""" Create an empty version of the place table.
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Specialised psycopg cursor with shortcut functions useful for testing.
"""
import psycopg
+
class CursorForTesting(psycopg.Cursor):
""" Extension to the DictCursor class that provides execution
short-cuts that simplify writing assertions.
assert self.rowcount == 1
return self.fetchone()[0]
-
def row_set(self, sql, params=None):
""" Execute a query and return the result as a set of tuples.
Fails when the SQL command returns duplicate rows.
return result
-
def table_exists(self, table):
""" Check that a table with the given name exists in the database.
"""
WHERE tablename = %s""", (table, ))
return num == 1
-
def index_exists(self, table, index):
""" Check that an indexwith the given name exists on the given table.
"""
(table, index))
return num == 1
-
def table_rows(self, table, where=None):
""" Return the number of rows in the given table.
"""
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for function that handle country properties.
from nominatim_db.data import country_info
+
@pytest.fixture
def loaded_country(def_config):
country_info.setup_country_config(def_config)
info = country_info._CountryInfo()
info.load(config)
assert dict(info.items()) == {'de': {'partition': 3,
- 'languages': [],
- 'names': {'name': 'Deutschland'}}}
+ 'languages': [],
+ 'names': {'name': 'Deutschland'}}}
def test_setup_country_config_name_not_loaded(env_with_country_config):
assert dict(info.items()) == {'de': {'partition': 3,
'languages': ['de'],
- 'names': {}
- }}
+ 'names': {}}}
def test_setup_country_config_names_not_loaded(env_with_country_config):
assert dict(info.items()) == {'de': {'partition': 3,
'languages': ['de'],
- 'names': {}
- }}
+ 'names': {}}}
def test_setup_country_config_special_character(env_with_country_config):
bq:
partition: 250
languages: nl
- names:
- name:
+ names:
+ name:
default: "\\N"
""")
assert dict(info.items()) == {'bq': {'partition': 250,
'languages': ['nl'],
- 'names': {'name': '\x85'}
- }}
+ 'names': {'name': '\x85'}}}
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for specialised connection and cursor classes.
import nominatim_db.db.connection as nc
+
@pytest.fixture
def db(dsn):
with nc.connect(dsn) as conn:
assert nc.table_has_column(db, 'stuff', name) == result
+
def test_connection_index_exists(db, table_factory, temp_db_cursor):
assert not nc.index_exists(db, 'some_index')
with pytest.raises(psycopg.ProgrammingError, match='.*does not exist.*'):
nc.drop_tables(db, 'dfkjgjriogjigjgjrdghehtre', if_exists=False)
+
def test_connection_server_version_tuple(db):
ver = nc.server_version_tuple(db)
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for property table manpulation.
from nominatim_db.db import properties
+
@pytest.fixture
def property_factory(property_table, temp_db_cursor):
""" A function fixture that adds a property into the property table.
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for SQL preprocessing.
"""
import pytest
-import pytest_asyncio
+import pytest_asyncio # noqa
from nominatim_db.db.sql_preprocessor import SQLPreprocessor
+
@pytest.fixture
def sql_factory(tmp_path):
def _mk_sql(sql_body):
return _mk_sql
+
@pytest.mark.parametrize("expr,ret", [
("'a'", 'a'),
("'{{db.partitions|join}}'", '012'),
async def test_load_parallel_file(dsn, sql_preprocessor, tmp_path, temp_db_cursor):
(tmp_path / 'test.sql').write_text("""
CREATE TABLE foo (a TEXT);
- CREATE TABLE foo2(a TEXT);""" +
- "\n---\nCREATE TABLE bar (b INT);")
+ CREATE TABLE foo2(a TEXT);""" + "\n---\nCREATE TABLE bar (b INT);")
await sql_preprocessor.run_parallel_sql_file(dsn, 'test.sql', num_threads=4)
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for status table manipulation.
<node id="45673" visible="true" version="1" changeset="2047" timestamp="2006-01-27T22:09:10Z" user="Foo" uid="111" lat="48.7586670" lon="8.1343060">
</node>
</osm>
-"""
+""" # noqa
+
def iso_date(date):
return dt.datetime.strptime(date, nominatim_db.db.status.ISODATE_FORMAT)\
def test_compute_database_date_from_osm2pgsql_nodata(table_factory, temp_db_conn):
table_factory('osm2pgsql_properties', 'property TEXT, value TEXT')
- with pytest.raises(UsageError, match='Cannot determine database date from data in offline mode'):
+ with pytest.raises(UsageError,
+ match='Cannot determine database date from data in offline mode'):
nominatim_db.db.status.compute_database_date(temp_db_conn, offline=True)
place_row(osm_type='N', osm_id=45673)
requested_url = []
+
def mock_url(url):
requested_url.append(url)
return OSM_NODE_DATA
place_row(osm_type='N', osm_id=45673)
requested_url = []
+
def mock_url(url):
requested_url.append(url)
return '<osm version="0.6" generator="OpenStre'
date = dt.datetime.fromordinal(1000000).replace(tzinfo=dt.timezone.utc)
nominatim_db.db.status.set_status(temp_db_conn, date=date)
- assert temp_db_cursor.row_set("SELECT * FROM import_status") == \
- {(date, None, True)}
+ assert temp_db_cursor.row_set("SELECT * FROM import_status") == {(date, None, True)}
def test_set_status_filled_table(temp_db_conn, temp_db_cursor):
date = dt.datetime.fromordinal(1000100).replace(tzinfo=dt.timezone.utc)
nominatim_db.db.status.set_status(temp_db_conn, date=date, seq=456, indexed=False)
- assert temp_db_cursor.row_set("SELECT * FROM import_status") == \
- {(date, 456, False)}
+ assert temp_db_cursor.row_set("SELECT * FROM import_status") == {(date, 456, False)}
def test_set_status_missing_date(temp_db_conn, temp_db_cursor):
nominatim_db.db.status.set_status(temp_db_conn, date=None, seq=456, indexed=False)
- assert temp_db_cursor.row_set("SELECT * FROM import_status") == \
- {(date, 456, False)}
+ assert temp_db_cursor.row_set("SELECT * FROM import_status") == {(date, 456, False)}
def test_get_status_empty_table(temp_db_conn):
date = dt.datetime.fromordinal(1000000).replace(tzinfo=dt.timezone.utc)
nominatim_db.db.status.set_status(temp_db_conn, date=date, seq=667, indexed=False)
- assert nominatim_db.db.status.get_status(temp_db_conn) == \
- (date, 667, False)
+ assert nominatim_db.db.status.get_status(temp_db_conn) == (date, 667, False)
@pytest.mark.parametrize("old_state", [True, False])
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for DB utility functions in db.utils
"""
-import json
-
import pytest
import nominatim_db.db.utils as db_utils
from nominatim_db.errors import UsageError
+
def test_execute_file_success(dsn, temp_db_cursor, tmp_path):
tmpfile = tmp_path / 'test.sql'
tmpfile.write_text('CREATE TABLE test (id INT);\nINSERT INTO test VALUES(56);')
assert temp_db_cursor.row_set('SELECT * FROM test') == {(56, )}
+
def test_execute_file_bad_file(dsn, tmp_path):
with pytest.raises(FileNotFoundError):
db_utils.execute_file(dsn, tmp_path / 'test2.sql')
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tokenizer for testing.
from nominatim_db.data.place_info import PlaceInfo
from nominatim_db.config import Configuration
+
def create(dsn, data_dir):
""" Create a new instance of the tokenizer provided by this module.
"""
return DummyTokenizer(dsn, data_dir)
+
class DummyTokenizer:
def __init__(self, dsn, data_dir):
self.init_state = None
self.analyser_cache = {}
-
def init_new_db(self, *args, **kwargs):
assert self.init_state is None
self.init_state = "new"
-
def init_from_project(self, config):
assert isinstance(config, Configuration)
assert self.init_state is None
self.init_state = "loaded"
-
@staticmethod
def finalize_import(_):
pass
-
def name_analyzer(self):
return DummyNameAnalyzer(self.analyser_cache)
def __exit__(self, exc_type, exc_value, traceback):
self.close()
-
def __init__(self, cache):
self.analyser_cache = cache
cache['countries'] = []
-
def close(self):
pass
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for running the indexing.
"""
import itertools
import pytest
-import pytest_asyncio
+import pytest_asyncio # noqa
from nominatim_db.indexer import indexer
from nominatim_db.tokenizer import factory
+
class IndexerTestDB:
def __init__(self, conn):
SELECT count(*) FROM placex
WHERE indexed_status = 0 AND rank_address between 1 and 27""") == 0
+
@pytest.mark.parametrize("threads", [1, 15])
@pytest.mark.asyncio
async def test_index_boundaries(test_db, threads, test_tokenizer):
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Legacy word table for testing with functions to prefil and test contents
"""
from nominatim_db.db.connection import execute_scalar
+
class MockIcuWordTable:
""" A word table for testing using legacy word table structure.
"""
(word_id, word or word_token, word))
self.conn.commit()
-
def add_special(self, word_token, word, cls, typ, oper):
with self.conn.cursor() as cur:
cur.execute("""INSERT INTO word (word_token, type, word, info)
""", (word_token, word, cls, typ, oper))
self.conn.commit()
-
def add_country(self, country_code, word_token):
with self.conn.cursor() as cur:
cur.execute("""INSERT INTO word (word_token, type, word)
(word_token, country_code))
self.conn.commit()
-
def add_postcode(self, word_token, postcode):
with self.conn.cursor() as cur:
cur.execute("""INSERT INTO word (word_token, type, word)
""", (word_token, postcode))
self.conn.commit()
-
def add_housenumber(self, word_id, word_tokens, word=None):
with self.conn.cursor() as cur:
if isinstance(word_tokens, str):
word = word_tokens[0]
for token in word_tokens:
cur.execute("""INSERT INTO word (word_id, word_token, type, word, info)
- VALUES (%s, %s, 'H', %s, jsonb_build_object('lookup', %s::text))
+ VALUES (%s, %s, 'H', %s,
+ jsonb_build_object('lookup', %s::text))
""", (word_id, token, word, word_tokens[0]))
self.conn.commit()
-
def count(self):
return execute_scalar(self.conn, "SELECT count(*) FROM word")
-
def count_special(self):
return execute_scalar(self.conn, "SELECT count(*) FROM word WHERE type = 'S'")
-
def count_housenumbers(self):
return execute_scalar(self.conn, "SELECT count(*) FROM word WHERE type = 'H'")
-
def get_special(self):
with self.conn.cursor() as cur:
cur.execute("SELECT word_token, info, word FROM word WHERE type = 'S'")
assert len(result) == cur.rowcount, "Word table has duplicates."
return result
-
def get_country(self):
with self.conn.cursor() as cur:
cur.execute("SELECT word, word_token FROM word WHERE type = 'C'")
assert len(result) == cur.rowcount, "Word table has duplicates."
return result
-
def get_postcodes(self):
with self.conn.cursor() as cur:
cur.execute("SELECT word FROM word WHERE type = 'P'")
return set((row[0] for row in cur))
-
def get_partial_words(self):
with self.conn.cursor() as cur:
cur.execute("SELECT word_token, info FROM word WHERE type ='w'")
return set(((row[0], row[1]['count']) for row in cur))
-
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Custom mocks for testing.
from nominatim_db.db import properties
-# This must always point to the mock word table for the default tokenizer.
-from mock_icu_word_table import MockIcuWordTable as MockWordTable
class MockPlacexTable:
""" A placex table for testing.
type, name, admin_level, address,
housenumber, rank_search,
extratags, geometry, country_code)
- VALUES(nextval('seq_place'), %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)""",
+ VALUES(nextval('seq_place'), %s, %s, %s, %s, %s, %s,
+ %s, %s, %s, %s, %s, %s)""",
(osm_type, osm_id or next(self.idseq), cls, typ, names,
admin_level, address, housenumber, rank_search,
extratags, 'SRID=4326;' + geom,
def __init__(self, conn):
self.conn = conn
-
def set(self, name, value):
""" Set a property in the table to the given value.
"""
properties.set_property(self.conn, name, value)
-
def get(self, name):
""" Set a property in the table to the given value.
"""
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the sanitizer that normalizes housenumbers.
from nominatim_db.tokenizer.place_sanitizer import PlaceSanitizer
from nominatim_db.data.place_info import PlaceInfo
+
@pytest.fixture
def sanitize(request, def_config):
sanitizer_args = {'step': 'clean-housenumbers'}
for mark in request.node.iter_markers(name="sanitizer_params"):
- sanitizer_args.update({k.replace('_', '-') : v for k,v in mark.kwargs.items()})
+ sanitizer_args.update({k.replace('_', '-'): v for k, v in mark.kwargs.items()})
def _run(**kwargs):
place = PlaceInfo({'address': kwargs})
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the sanitizer that normalizes postcodes.
from nominatim_db.data.place_info import PlaceInfo
from nominatim_db.data import country_info
+
@pytest.fixture
def sanitize(def_config, request):
country_info.setup_country_config(def_config)
sanitizer_args = {'step': 'clean-postcodes'}
for mark in request.node.iter_markers(name="sanitizer_params"):
- sanitizer_args.update({k.replace('_', '-') : v for k,v in mark.kwargs.items()})
+ sanitizer_args.update({k.replace('_', '-'): v for k, v in mark.kwargs.items()})
def _run(country=None, **kwargs):
pi = {'address': kwargs}
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for sanitizer that clean up TIGER tags.
from nominatim_db.tokenizer.place_sanitizer import PlaceSanitizer
from nominatim_db.data.place_info import PlaceInfo
+
class TestCleanTigerTags:
@pytest.fixture(autouse=True)
def setup_country(self, def_config):
self.config = def_config
-
def run_sanitizer_on(self, addr):
place = PlaceInfo({'address': addr})
- _, outaddr = PlaceSanitizer([{'step': 'clean-tiger-tags'}], self.config).process_names(place)
+ _, outaddr = PlaceSanitizer([{'step': 'clean-tiger-tags'}],
+ self.config).process_names(place)
return sorted([(p.name, p.kind, p.suffix) for p in outaddr])
assert self.run_sanitizer_on({'tiger:county': inname})\
== [(outname, 'county', 'tiger')]
-
@pytest.mark.parametrize('name', ('Hamilton', 'Big, Road', ''))
def test_badly_formatted(self, name):
assert self.run_sanitizer_on({'tiger:county': name})\
== [(name, 'county', 'tiger')]
-
def test_unmatched(self):
assert self.run_sanitizer_on({'tiger:country': 'US'})\
== [('US', 'tiger', 'country')]
#\r
# This file is part of Nominatim. (https://nominatim.org)\r
#\r
-# Copyright (C) 2024 by the Nominatim developer community.\r
+# Copyright (C) 2025 by the Nominatim developer community.\r
# For a full list of authors see the git log.\r
"""\r
Tests for the sanitizer that normalizes housenumbers.\r
def run_sanitizer_on(self, type, **kwargs):\r
\r
place = PlaceInfo({type: {k.replace('_', ':'): v for k, v in kwargs.items()},\r
- 'country_code': 'de', 'rank_address': 30})\r
+ 'country_code': 'de', 'rank_address': 30})\r
\r
sanitizer_args = {'step': 'delete-tags'}\r
\r
name, address = PlaceSanitizer([sanitizer_args],\r
- self.config).process_names(place)\r
-\r
- return {\r
- 'name': sorted([(p.name, p.kind, p.suffix or '') for p in name]),\r
- 'address': sorted([(p.name, p.kind, p.suffix or '') for p in address])\r
- }\r
+ self.config).process_names(place)\r
\r
+ return {'name': sorted([(p.name, p.kind, p.suffix or '') for p in name]),\r
+ 'address': sorted([(p.name, p.kind, p.suffix or '') for p in address])}\r
\r
def test_on_name(self):\r
res = self.run_sanitizer_on('name', name='foo', ref='bar', ref_abc='baz')\r
res = self.run_sanitizer_on('address', name='foo', ref='bar', ref_abc='baz')\r
\r
assert res.get('address') == [('bar', 'ref', ''), ('baz', 'ref', 'abc'),\r
- ('foo', 'name', '')]\r
+ ('foo', 'name', '')]\r
\r
\r
class TestTypeField:\r
def run_sanitizer_on(self, type, **kwargs):\r
\r
place = PlaceInfo({'name': {k.replace('_', ':'): v for k, v in kwargs.items()},\r
- 'country_code': 'de', 'rank_address': 30})\r
+ 'country_code': 'de', 'rank_address': 30})\r
\r
- sanitizer_args = {\r
- 'step': 'delete-tags',\r
- 'type': type,\r
- }\r
+ sanitizer_args = {'step': 'delete-tags',\r
+ 'type': type}\r
\r
name, _ = PlaceSanitizer([sanitizer_args],\r
- self.config).process_names(place)\r
+ self.config).process_names(place)\r
\r
return sorted([(p.name, p.kind, p.suffix or '') for p in name])\r
\r
res = self.run_sanitizer_on('address', name='foo', ref='bar', ref_abc='baz')\r
\r
assert res == [('bar', 'ref', ''), ('baz', 'ref', 'abc'),\r
- ('foo', 'name', '')]\r
+ ('foo', 'name', '')]\r
+\r
\r
class TestFilterKind:\r
\r
def run_sanitizer_on(self, filt, **kwargs):\r
\r
place = PlaceInfo({'name': {k.replace('_', ':'): v for k, v in kwargs.items()},\r
- 'country_code': 'de', 'rank_address': 30})\r
+ 'country_code': 'de', 'rank_address': 30})\r
\r
- sanitizer_args = {\r
- 'step': 'delete-tags',\r
- 'filter-kind': filt,\r
- }\r
+ sanitizer_args = {'step': 'delete-tags',\r
+ 'filter-kind': filt}\r
\r
name, _ = PlaceSanitizer([sanitizer_args],\r
- self.config).process_names(place)\r
+ self.config).process_names(place)\r
\r
return sorted([(p.name, p.kind, p.suffix or '') for p in name])\r
\r
\r
assert res == [('bar', 'ref', 'abc'), ('foo', 'ref', '')]\r
\r
-\r
def test_single_pattern(self):\r
res = self.run_sanitizer_on(['.*name'],\r
name_fr='foo', ref_fr='foo', namexx_fr='bar',\r
\r
assert res == [('bar', 'namexx', 'fr'), ('foo', 'ref', 'fr')]\r
\r
-\r
def test_multiple_patterns(self):\r
res = self.run_sanitizer_on(['.*name', 'ref'],\r
name_fr='foo', ref_fr='foo', oldref_fr='foo',\r
def run_sanitizer_on(self, rank_addr, **kwargs):\r
\r
place = PlaceInfo({'name': {k.replace('_', ':'): v for k, v in kwargs.items()},\r
- 'country_code': 'de', 'rank_address': 30})\r
+ 'country_code': 'de', 'rank_address': 30})\r
\r
- sanitizer_args = {\r
- 'step': 'delete-tags',\r
- 'rank_address': rank_addr\r
- }\r
+ sanitizer_args = {'step': 'delete-tags',\r
+ 'rank_address': rank_addr}\r
\r
name, _ = PlaceSanitizer([sanitizer_args],\r
- self.config).process_names(place)\r
+ self.config).process_names(place)\r
\r
return sorted([(p.name, p.kind, p.suffix or '') for p in name])\r
\r
-\r
def test_single_rank(self):\r
res = self.run_sanitizer_on('30', name='foo', ref='bar')\r
\r
def run_sanitizer_on(self, suffix, **kwargs):\r
\r
place = PlaceInfo({'name': {k.replace('_', ':'): v for k, v in kwargs.items()},\r
- 'country_code': 'de', 'rank_address': 30})\r
+ 'country_code': 'de', 'rank_address': 30})\r
\r
- sanitizer_args = {\r
- 'step': 'delete-tags',\r
- 'suffix': suffix,\r
- }\r
+ sanitizer_args = {'step': 'delete-tags',\r
+ 'suffix': suffix}\r
\r
name, _ = PlaceSanitizer([sanitizer_args],\r
- self.config).process_names(place)\r
+ self.config).process_names(place)\r
\r
return sorted([(p.name, p.kind, p.suffix or '') for p in name])\r
\r
-\r
def test_single_suffix(self):\r
res = self.run_sanitizer_on('abc', name='foo', name_abc='foo',\r
- name_pqr='bar', ref='bar', ref_abc='baz')\r
+ name_pqr='bar', ref='bar', ref_abc='baz')\r
\r
assert res == [('bar', 'name', 'pqr'), ('bar', 'ref', ''), ('foo', 'name', '')]\r
\r
def test_multiple_suffix(self):\r
res = self.run_sanitizer_on(['abc.*', 'pqr'], name='foo', name_abcxx='foo',\r
- ref_pqr='bar', name_pqrxx='baz')\r
+ ref_pqr='bar', name_pqrxx='baz')\r
\r
assert res == [('baz', 'name', 'pqrxx'), ('foo', 'name', '')]\r
\r
\r
-\r
class TestCountryCodes:\r
\r
@pytest.fixture(autouse=True)\r
def run_sanitizer_on(self, country_code, **kwargs):\r
\r
place = PlaceInfo({'name': {k.replace('_', ':'): v for k, v in kwargs.items()},\r
- 'country_code': 'de', 'rank_address': 30})\r
+ 'country_code': 'de', 'rank_address': 30})\r
\r
- sanitizer_args = {\r
- 'step': 'delete-tags',\r
- 'country_code': country_code,\r
- }\r
+ sanitizer_args = {'step': 'delete-tags',\r
+ 'country_code': country_code}\r
\r
name, _ = PlaceSanitizer([sanitizer_args],\r
- self.config).process_names(place)\r
+ self.config).process_names(place)\r
\r
return sorted([(p.name, p.kind) for p in name])\r
\r
-\r
def test_single_country_code_pass(self):\r
res = self.run_sanitizer_on('de', name='foo', ref='bar')\r
\r
\r
assert res == [('bar', 'ref'), ('foo', 'name')]\r
\r
+\r
class TestAllParameters:\r
\r
@pytest.fixture(autouse=True)\r
def run_sanitizer_on(self, country_code, rank_addr, suffix, **kwargs):\r
\r
place = PlaceInfo({'name': {k.replace('_', ':'): v for k, v in kwargs.items()},\r
- 'country_code': 'de', 'rank_address': 30})\r
+ 'country_code': 'de', 'rank_address': 30})\r
\r
sanitizer_args = {\r
'step': 'delete-tags',\r
}\r
\r
name, _ = PlaceSanitizer([sanitizer_args],\r
- self.config).process_names(place)\r
+ self.config).process_names(place)\r
\r
return sorted([(p.name, p.kind, p.suffix or '') for p in name])\r
\r
-\r
def test_string_arguments_pass(self):\r
res = self.run_sanitizer_on('de', '25-30', r'[\s\S]*',\r
name='foo', ref='foo', name_abc='bar', ref_abc='baz')\r
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for sanitizer configuration helper functions.
from nominatim_db.errors import UsageError
from nominatim_db.tokenizer.sanitizers.config import SanitizerConfig
+
def test_string_list_default_empty():
assert SanitizerConfig().get_string_list('op') == []
('ying;;yang', ['ying', 'yang']),
(';a; ;c;d,', ['', 'a', '', 'c', 'd', '']),
('1, 3 ,5', ['1', '3', '5'])
- ])
+ ])
def test_create_split_regex_no_params_split(inp, outp):
regex = SanitizerConfig().get_delimiter()
def test_create_split_regex_empty_delimiter():
with pytest.raises(UsageError):
- regex = SanitizerConfig({'delimiters': ''}).get_delimiter()
+ SanitizerConfig({'delimiters': ''}).get_delimiter()
@pytest.mark.parametrize('inp', ('name', 'name:de', 'na\\me', '.*', ''))
def test_create_name_filter_no_param_default_invalid_string():
with pytest.raises(ValueError):
- filt = SanitizerConfig().get_filter('name', 'abc')
+ SanitizerConfig().get_filter('name', 'abc')
def test_create_name_filter_no_param_default_empty_list():
with pytest.raises(ValueError):
- filt = SanitizerConfig().get_filter('name', [])
+ SanitizerConfig().get_filter('name', [])
@pytest.mark.parametrize('kind', ('de', 'name:de', 'ende'))
@pytest.mark.parametrize('kind', ('lang', 'lang:de', 'langxx'))
def test_create_kind_filter_custom_regex_positive(kind):
filt = SanitizerConfig({'filter-kind': 'lang.*'}
- ).get_filter('filter-kind', ['.*fr'])
+ ).get_filter('filter-kind', ['.*fr'])
assert filt(kind)
@pytest.mark.parametrize('kind', ('name', 'fr', 'name:fr', 'frfr', '34'))
def test_create_kind_filter_many_positive(kind):
filt = SanitizerConfig({'filter-kind': ['.*fr', 'name', r'\d+']}
- ).get_filter('filter-kind')
+ ).get_filter('filter-kind')
assert filt(kind)
@pytest.mark.parametrize('kind', ('name:de', 'fridge', 'a34', '.*', '\\'))
def test_create_kind_filter_many_negative(kind):
filt = SanitizerConfig({'filter-kind': ['.*fr', 'name', r'\d+']}
- ).get_filter('filter-kind')
+ ).get_filter('filter-kind')
assert not filt(kind)
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the sanitizer that splits multivalue lists.
from nominatim_db.errors import UsageError
+
class TestSplitName:
@pytest.fixture(autouse=True)
def setup_country(self, def_config):
self.config = def_config
-
def run_sanitizer_on(self, **kwargs):
place = PlaceInfo({'name': kwargs})
name, _ = PlaceSanitizer([{'step': 'split-name-list'}], self.config).process_names(place)
return sorted([(p.name, p.kind, p.suffix) for p in name])
-
def sanitize_with_delimiter(self, delimiter, name):
place = PlaceInfo({'name': {'name': name}})
san = PlaceSanitizer([{'step': 'split-name-list', 'delimiters': delimiter}],
return sorted([p.name for p in name])
-
def test_simple(self):
assert self.run_sanitizer_on(name='ABC') == [('ABC', 'name', None)]
assert self.run_sanitizer_on(name='') == [('', 'name', None)]
-
def test_splits(self):
assert self.run_sanitizer_on(name='A;B;C') == [('A', 'name', None),
('B', 'name', None),
assert self.run_sanitizer_on(short_name=' House, boat ') == [('House', 'short_name', None),
('boat', 'short_name', None)]
-
def test_empty_fields(self):
assert self.run_sanitizer_on(name='A;;B') == [('A', 'name', None),
('B', 'name', None)]
assert self.run_sanitizer_on(name=' ;B') == [('B', 'name', None)]
assert self.run_sanitizer_on(name='B,') == [('B', 'name', None)]
-
def test_custom_delimiters(self):
assert self.sanitize_with_delimiter(':', '12:45,3') == ['12', '45,3']
assert self.sanitize_with_delimiter('\\', 'a;\\b!#@ \\') == ['a;', 'b!#@']
assert self.sanitize_with_delimiter('[]', 'foo[to]be') == ['be', 'foo', 'to']
assert self.sanitize_with_delimiter(' ', 'morning sun') == ['morning', 'sun']
-
def test_empty_delimiter_set(self):
with pytest.raises(UsageError):
self.sanitize_with_delimiter('', 'abc')
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the sanitizer that handles braced suffixes.
from nominatim_db.tokenizer.place_sanitizer import PlaceSanitizer
from nominatim_db.data.place_info import PlaceInfo
+
class TestStripBrace:
@pytest.fixture(autouse=True)
return sorted([(p.name, p.kind, p.suffix) for p in name])
-
def test_no_braces(self):
assert self.run_sanitizer_on(name='foo', ref='23') == [('23', 'ref', None),
('foo', 'name', None)]
-
def test_simple_braces(self):
- assert self.run_sanitizer_on(name='Halle (Saale)', ref='3')\
- == [('3', 'ref', None), ('Halle', 'name', None), ('Halle (Saale)', 'name', None)]
- assert self.run_sanitizer_on(name='ack ( bar')\
- == [('ack', 'name', None), ('ack ( bar', 'name', None)]
-
+ assert self.run_sanitizer_on(name='Halle (Saale)', ref='3') \
+ == [('3', 'ref', None), ('Halle', 'name', None), ('Halle (Saale)', 'name', None)]
+ assert self.run_sanitizer_on(name='ack ( bar') \
+ == [('ack', 'name', None), ('ack ( bar', 'name', None)]
def test_only_braces(self):
assert self.run_sanitizer_on(name='(maybe)') == [('(maybe)', 'name', None)]
-
def test_double_braces(self):
assert self.run_sanitizer_on(name='a((b))') == [('a', 'name', None),
('a((b))', 'name', None)]
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the sanitizer that enables language-dependent analyzers.
from nominatim_db.tokenizer.place_sanitizer import PlaceSanitizer
from nominatim_db.data.country_info import setup_country_config
+
class TestWithDefaults:
@pytest.fixture(autouse=True)
def setup_country(self, def_config):
self.config = def_config
-
def run_sanitizer_on(self, country, **kwargs):
place = PlaceInfo({'name': {k.replace('_', ':'): v for k, v in kwargs.items()},
'country_code': country})
return sorted([(p.name, p.kind, p.suffix, p.attr) for p in name])
-
def test_no_names(self):
assert self.run_sanitizer_on('de') == []
-
def test_simple(self):
- res = self.run_sanitizer_on('fr', name='Foo',name_de='Zoo', ref_abc='M')
+ res = self.run_sanitizer_on('fr', name='Foo', name_de='Zoo', ref_abc='M')
assert res == [('Foo', 'name', None, {}),
('M', 'ref', 'abc', {'analyzer': 'abc'}),
('Zoo', 'name', 'de', {'analyzer': 'de'})]
-
@pytest.mark.parametrize('suffix', ['DE', 'asbc'])
def test_illegal_suffix(self, suffix):
assert self.run_sanitizer_on('fr', **{'name_' + suffix: 'Foo'}) \
def setup_country(self, def_config):
self.config = def_config
-
def run_sanitizer_on(self, filt, **kwargs):
place = PlaceInfo({'name': {k.replace('_', ':'): v for k, v in kwargs.items()},
'country_code': 'de'})
return sorted([(p.name, p.kind, p.suffix, p.attr) for p in name])
-
def test_single_exact_name(self):
res = self.run_sanitizer_on(['name'], name_fr='A', ref_fr='12',
- shortname_fr='C', name='D')
+ shortname_fr='C', name='D')
assert res == [('12', 'ref', 'fr', {}),
('A', 'name', 'fr', {'analyzer': 'fr'}),
('C', 'shortname', 'fr', {}),
('D', 'name', None, {})]
-
def test_single_pattern(self):
res = self.run_sanitizer_on(['.*name'],
name_fr='A', ref_fr='12', namexx_fr='B',
('C', 'shortname', 'fr', {'analyzer': 'fr'}),
('D', 'name', None, {})]
-
def test_multiple_patterns(self):
res = self.run_sanitizer_on(['.*name', 'ref'],
name_fr='A', ref_fr='12', oldref_fr='X',
setup_country_config(def_config)
self.config = def_config
-
def run_sanitizer_append(self, mode, country, **kwargs):
place = PlaceInfo({'name': {k.replace('_', ':'): v for k, v in kwargs.items()},
'country_code': country})
return sorted([(p.name, p.attr.get('analyzer', '')) for p in name])
-
def run_sanitizer_replace(self, mode, country, **kwargs):
place = PlaceInfo({'name': {k.replace('_', ':'): v for k, v in kwargs.items()},
'country_code': country})
return sorted([(p.name, p.attr.get('analyzer', '')) for p in name])
-
def test_missing_country(self):
place = PlaceInfo({'name': {'name': 'something'}})
name, _ = PlaceSanitizer([{'step': 'tag-analyzer-by-language',
assert name[0].suffix is None
assert 'analyzer' not in name[0].attr
-
def test_mono_unknown_country(self):
expect = [('XX', '')]
assert self.run_sanitizer_replace('mono', 'xx', name='XX') == expect
assert self.run_sanitizer_append('mono', 'xx', name='XX') == expect
-
def test_mono_monoling_replace(self):
res = self.run_sanitizer_replace('mono', 'de', name='Foo')
assert res == [('Foo', 'de')]
-
def test_mono_monoling_append(self):
res = self.run_sanitizer_append('mono', 'de', name='Foo')
assert res == [('Foo', ''), ('Foo', 'de')]
-
def test_mono_multiling(self):
expect = [('XX', '')]
assert self.run_sanitizer_replace('mono', 'ch', name='XX') == expect
assert self.run_sanitizer_append('mono', 'ch', name='XX') == expect
-
def test_all_unknown_country(self):
expect = [('XX', '')]
assert self.run_sanitizer_replace('all', 'xx', name='XX') == expect
assert self.run_sanitizer_append('all', 'xx', name='XX') == expect
-
def test_all_monoling_replace(self):
res = self.run_sanitizer_replace('all', 'de', name='Foo')
assert res == [('Foo', 'de')]
-
def test_all_monoling_append(self):
res = self.run_sanitizer_append('all', 'de', name='Foo')
assert res == [('Foo', ''), ('Foo', 'de')]
-
def test_all_multiling_append(self):
res = self.run_sanitizer_append('all', 'ch', name='XX')
assert res == [('XX', ''),
('XX', 'de'), ('XX', 'fr'), ('XX', 'it'), ('XX', 'rm')]
-
def test_all_multiling_replace(self):
res = self.run_sanitizer_replace('all', 'ch', name='XX')
def setup_country(self, def_config):
self.config = def_config
-
def run_sanitizer_on(self, mode, country, **kwargs):
place = PlaceInfo({'name': {k.replace('_', ':'): v for k, v in kwargs.items()},
'country_code': country})
return sorted([(p.name, p.attr.get('analyzer', '')) for p in name])
-
def test_mono_monoling(self):
assert self.run_sanitizer_on('mono', 'de', name='Foo') == [('Foo', 'de')]
assert self.run_sanitizer_on('mono', 'pt', name='Foo') == [('Foo', '')]
-
def test_mono_multiling(self):
assert self.run_sanitizer_on('mono', 'ca', name='Foo') == [('Foo', '')]
-
def test_all_monoling(self):
assert self.run_sanitizer_on('all', 'de', name='Foo') == [('Foo', 'de')]
assert self.run_sanitizer_on('all', 'pt', name='Foo') == [('Foo', '')]
-
def test_all_multiling(self):
assert self.run_sanitizer_on('all', 'ca', name='Foo') == [('Foo', 'fr')]
assert self.run_sanitizer_on('all', 'ch', name='Foo') \
def setup_country(self, def_config):
self.config = def_config
-
def run_sanitizer_on(self, whitelist, **kwargs):
place = PlaceInfo({'name': {k.replace('_', ':'): v for k, v in kwargs.items()}})
name, _ = PlaceSanitizer([{'step': 'tag-analyzer-by-language',
return sorted([(p.name, p.attr.get('analyzer', '')) for p in name])
-
def test_in_whitelist(self):
assert self.run_sanitizer_on(['de', 'xx'], ref_xx='123') == [('123', 'xx')]
-
def test_not_in_whitelist(self):
assert self.run_sanitizer_on(['de', 'xx'], ref_yy='123') == [('123', '')]
-
def test_empty_whitelist(self):
assert self.run_sanitizer_on([], ref_yy='123') == [('123', '')]
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
-from typing import Mapping, Optional, List
import pytest
from nominatim_db.data.place_info import PlaceInfo
-from nominatim_db.data.place_name import PlaceName
from nominatim_db.tokenizer.place_sanitizer import PlaceSanitizer
+
class TestTagJapanese:
@pytest.fixture(autouse=True)
def setup_country(self, def_config):
self.config = def_config
- def run_sanitizer_on(self,type, **kwargs):
+ def run_sanitizer_on(self, type, **kwargs):
place = PlaceInfo({
'address': kwargs,
'country_code': 'jp'
})
sanitizer_args = {'step': 'tag-japanese'}
_, address = PlaceSanitizer([sanitizer_args], self.config).process_names(place)
- tmp_list = [(p.name,p.kind) for p in address]
+ tmp_list = [(p.name, p.kind) for p in address]
return sorted(tmp_list)
def test_on_address(self):
res = self.run_sanitizer_on('address', name='foo', ref='bar', ref_abc='baz')
- assert res == [('bar','ref'),('baz','ref_abc'),('foo','name')]
+ assert res == [('bar', 'ref'), ('baz', 'ref_abc'), ('foo', 'name')]
def test_housenumber(self):
res = self.run_sanitizer_on('address', housenumber='2')
- assert res == [('2','housenumber')]
+ assert res == [('2', 'housenumber')]
def test_blocknumber(self):
res = self.run_sanitizer_on('address', block_number='6')
- assert res == [('6','housenumber')]
+ assert res == [('6', 'housenumber')]
def test_neighbourhood(self):
res = self.run_sanitizer_on('address', neighbourhood='8')
- assert res == [('8','place')]
+ assert res == [('8', 'place')]
def test_quarter(self):
res = self.run_sanitizer_on('address', quarter='kase')
- assert res==[('kase','place')]
+ assert res == [('kase', 'place')]
def test_housenumber_blocknumber(self):
res = self.run_sanitizer_on('address', housenumber='2', block_number='6')
- assert res == [('6-2','housenumber')]
+ assert res == [('6-2', 'housenumber')]
def test_quarter_neighbourhood(self):
res = self.run_sanitizer_on('address', quarter='kase', neighbourhood='8')
- assert res == [('kase8','place')]
+ assert res == [('kase8', 'place')]
def test_blocknumber_housenumber_quarter(self):
res = self.run_sanitizer_on('address', block_number='6', housenumber='2', quarter='kase')
- assert res == [('6-2','housenumber'),('kase','place')]
+ assert res == [('6-2', 'housenumber'), ('kase', 'place')]
def test_blocknumber_housenumber_quarter_neighbourhood(self):
res = self.run_sanitizer_on('address', block_number='6', housenumber='2', neighbourhood='8')
- assert res == [('6-2','housenumber'),('8','place')]
+ assert res == [('6-2', 'housenumber'), ('8', 'place')]
def test_blocknumber_quarter_neighbourhood(self):
- res = self.run_sanitizer_on('address',block_number='6', quarter='kase', neighbourhood='8')
- assert res == [('6','housenumber'),('kase8','place')]
+ res = self.run_sanitizer_on('address', block_number='6', quarter='kase', neighbourhood='8')
+ assert res == [('6', 'housenumber'), ('kase8', 'place')]
def test_blocknumber_quarter(self):
- res = self.run_sanitizer_on('address',block_number='6', quarter='kase')
- assert res == [('6','housenumber'),('kase','place')]
+ res = self.run_sanitizer_on('address', block_number='6', quarter='kase')
+ assert res == [('6', 'housenumber'), ('kase', 'place')]
def test_blocknumber_neighbourhood(self):
- res = self.run_sanitizer_on('address',block_number='6', neighbourhood='8')
- assert res == [('6','housenumber'),('8','place')]
+ res = self.run_sanitizer_on('address', block_number='6', neighbourhood='8')
+ assert res == [('6', 'housenumber'), ('8', 'place')]
def test_housenumber_quarter_neighbourhood(self):
- res = self.run_sanitizer_on('address',housenumber='2', quarter='kase', neighbourhood='8')
- assert res == [('2','housenumber'),('kase8','place')]
+ res = self.run_sanitizer_on('address', housenumber='2', quarter='kase', neighbourhood='8')
+ assert res == [('2', 'housenumber'), ('kase8', 'place')]
def test_housenumber_quarter(self):
- res = self.run_sanitizer_on('address',housenumber='2', quarter='kase')
- assert res == [('2','housenumber'),('kase','place')]
+ res = self.run_sanitizer_on('address', housenumber='2', quarter='kase')
+ assert res == [('2', 'housenumber'), ('kase', 'place')]
def test_housenumber_blocknumber_neighbourhood_quarter(self):
- res = self.run_sanitizer_on('address', block_number='6', housenumber='2', quarter='kase', neighbourhood='8')
- assert res == [('6-2','housenumber'),('kase8','place')]
+ res = self.run_sanitizer_on('address', block_number='6', housenumber='2',
+ quarter='kase', neighbourhood='8')
+ assert res == [('6-2', 'housenumber'), ('kase8', 'place')]
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for creating new tokenizers.
def init_env(self, project_env, property_table, tokenizer_mock):
self.config = project_env
-
def test_setup_dummy_tokenizer(self, temp_db_conn):
tokenizer = factory.create_tokenizer(self.config)
assert properties.get_property(temp_db_conn, 'tokenizer') == 'dummy'
-
def test_setup_tokenizer_dir_exists(self):
(self.config.project_dir / 'tokenizer').mkdir()
assert isinstance(tokenizer, DummyTokenizer)
assert tokenizer.init_state == "new"
-
def test_setup_tokenizer_dir_failure(self):
(self.config.project_dir / 'tokenizer').write_text("foo")
with pytest.raises(UsageError):
factory.create_tokenizer(self.config)
-
def test_load_tokenizer(self):
factory.create_tokenizer(self.config)
assert isinstance(tokenizer, DummyTokenizer)
assert tokenizer.init_state == "loaded"
-
def test_load_repopulate_tokenizer_dir(self):
factory.create_tokenizer(self.config)
factory.get_tokenizer_for_db(self.config)
assert (self.config.project_dir / 'tokenizer').exists()
-
def test_load_missing_property(self, temp_db_cursor):
factory.create_tokenizer(self.config)
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for ICU tokenizer.
from mock_icu_word_table import MockIcuWordTable
+
@pytest.fixture
def word_table(temp_db_conn):
return MockIcuWordTable(temp_db_conn)
return _mk_analyser
+
@pytest.fixture
def sql_functions(temp_db_conn, def_config, src_dir):
orig_sql = def_config.lib_dir.sql
""")
-
def test_init_new(tokenizer_factory, test_config, db_prop):
tok = tokenizer_factory()
tok.init_new_db(test_config)
- assert db_prop(nominatim_db.tokenizer.icu_rule_loader.DBCFG_IMPORT_NORM_RULES) \
- .startswith(':: lower ();')
+ prop = db_prop(nominatim_db.tokenizer.icu_rule_loader.DBCFG_IMPORT_NORM_RULES)
+
+ assert prop.startswith(':: lower ();')
def test_init_word_table(tokenizer_factory, test_config, place_row, temp_db_cursor):
- place_row(names={'name' : 'Test Area', 'ref' : '52'})
- place_row(names={'name' : 'No Area'})
- place_row(names={'name' : 'Holzstrasse'})
+ place_row(names={'name': 'Test Area', 'ref': '52'})
+ place_row(names={'name': 'No Area'})
+ place_row(names={'name': 'Holzstrasse'})
tok = tokenizer_factory()
tok.init_new_db(test_config)
self.analyzer = anl
yield anl
-
def process_postcode(self, cc, postcode):
return self.analyzer.process_place(PlaceInfo({'country_code': cc,
'address': {'postcode': postcode}}))
-
def test_update_postcodes_deleted(self, word_table):
word_table.add_postcode(' 1234', '1234')
word_table.add_postcode(' 5678', '5678')
assert word_table.count() == 0
-
def test_process_place_postcode_simple(self, word_table):
info = self.process_postcode('de', '12345')
assert info['postcode'] == '12345'
-
def test_process_place_postcode_with_space(self, word_table):
info = self.process_postcode('in', '123 567')
assert info['postcode'] == '123567'
-
def test_update_special_phrase_empty_table(analyzer, word_table):
with analyzer() as anl:
anl.update_special_phrases([
], True)
assert word_table.get_special() \
- == {('KÖNIG BEI', 'König bei', 'amenity', 'royal', 'near'),
- ('KÖNIGE', 'Könige', 'amenity', 'royal', None),
- ('STREET', 'street', 'highway', 'primary', 'in')}
+ == {('KÖNIG BEI', 'König bei', 'amenity', 'royal', 'near'),
+ ('KÖNIGE', 'Könige', 'amenity', 'royal', None),
+ ('STREET', 'street', 'highway', 'primary', 'in')}
def test_update_special_phrase_delete_all(analyzer, word_table):
], True)
assert word_table.get_special() \
- == {('PRISON', 'prison', 'amenity', 'prison', 'in'),
- ('BAR', 'bar', 'highway', 'road', None),
- ('GARDEN', 'garden', 'leisure', 'garden', 'near')}
+ == {('PRISON', 'prison', 'amenity', 'prison', 'in'),
+ ('BAR', 'bar', 'highway', 'road', None),
+ ('GARDEN', 'garden', 'leisure', 'garden', 'near')}
def test_add_country_names_new(analyzer, word_table):
self.analyzer = anl
yield anl
-
def expect_name_terms(self, info, *expected_terms):
tokens = self.analyzer.get_word_token_info(expected_terms)
for token in tokens:
assert eval(info['names']) == set((t[2] for t in tokens))
-
def process_named_place(self, names):
return self.analyzer.process_place(PlaceInfo({'name': names}))
-
def test_simple_names(self):
info = self.process_named_place({'name': 'Soft bAr', 'ref': '34'})
self.expect_name_terms(info, '#Soft bAr', '#34', 'Soft', 'bAr', '34')
-
- @pytest.mark.parametrize('sep', [',' , ';'])
+ @pytest.mark.parametrize('sep', [',', ';'])
def test_names_with_separator(self, sep):
info = self.process_named_place({'name': sep.join(('New York', 'Big Apple'))})
self.expect_name_terms(info, '#New York', '#Big Apple',
'new', 'york', 'big', 'apple')
-
def test_full_names_with_bracket(self):
info = self.process_named_place({'name': 'Houseboat (left)'})
self.expect_name_terms(info, '#Houseboat (left)', '#Houseboat',
'houseboat', 'left')
-
def test_country_name(self, word_table):
- place = PlaceInfo({'name' : {'name': 'Norge'},
+ place = PlaceInfo({'name': {'name': 'Norge'},
'country_code': 'no',
'rank_address': 4,
'class': 'boundary',
self.analyzer = anl
yield anl
-
@pytest.fixture
def getorcreate_hnr_id(self, temp_db_cursor):
temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION getorcreate_hnr_id(lookup_term TEXT)
RETURNS INTEGER AS $$
SELECT -nextval('seq_word')::INTEGER; $$ LANGUAGE SQL""")
-
def process_address(self, **kwargs):
return self.analyzer.process_place(PlaceInfo({'address': kwargs}))
-
def name_token_set(self, *expected_terms):
tokens = self.analyzer.get_word_token_info(expected_terms)
for token in tokens:
return set((t[2] for t in tokens))
-
@pytest.mark.parametrize('pcode', ['12345', 'AB 123', '34-345'])
def test_process_place_postcode(self, word_table, pcode):
info = self.process_address(postcode=pcode)
assert info['postcode'] == pcode
-
@pytest.mark.parametrize('hnr', ['123a', '1', '101'])
def test_process_place_housenumbers_simple(self, hnr, getorcreate_hnr_id):
info = self.process_address(housenumber=hnr)
assert info['hnr'] == hnr.upper()
assert info['hnr_tokens'] == "{-1}"
-
def test_process_place_housenumbers_duplicates(self, getorcreate_hnr_id):
info = self.process_address(housenumber='134',
conscriptionnumber='134',
assert set(info['hnr'].split(';')) == set(('134', '99A'))
assert info['hnr_tokens'] == "{-1,-2}"
-
def test_process_place_housenumbers_cached(self, getorcreate_hnr_id):
info = self.process_address(housenumber="45")
assert info['hnr_tokens'] == "{-1}"
info = self.process_address(housenumber="41")
assert eval(info['hnr_tokens']) == {-3}
-
def test_process_place_street(self):
- self.analyzer.process_place(PlaceInfo({'name': {'name' : 'Grand Road'}}))
+ self.analyzer.process_place(PlaceInfo({'name': {'name': 'Grand Road'}}))
info = self.process_address(street='Grand Road')
assert eval(info['street']) == self.name_token_set('#Grand Road')
-
def test_process_place_nonexisting_street(self):
info = self.process_address(street='Grand Road')
assert info['street'] == '{}'
-
def test_process_place_multiple_street_tags(self):
- self.analyzer.process_place(PlaceInfo({'name': {'name' : 'Grand Road',
+ self.analyzer.process_place(PlaceInfo({'name': {'name': 'Grand Road',
'ref': '05989'}}))
info = self.process_address(**{'street': 'Grand Road',
- 'street:sym_ul': '05989'})
+ 'street:sym_ul': '05989'})
assert eval(info['street']) == self.name_token_set('#Grand Road', '#05989')
-
def test_process_place_street_empty(self):
info = self.process_address(street='🜵')
assert info['street'] == '{}'
-
def test_process_place_street_from_cache(self):
- self.analyzer.process_place(PlaceInfo({'name': {'name' : 'Grand Road'}}))
+ self.analyzer.process_place(PlaceInfo({'name': {'name': 'Grand Road'}}))
self.process_address(street='Grand Road')
# request address again
assert eval(info['street']) == self.name_token_set('#Grand Road')
-
def test_process_place_place(self):
info = self.process_address(place='Honu Lulu')
assert eval(info['place']) == self.name_token_set('HONU', 'LULU', '#HONU LULU')
-
def test_process_place_place_extra(self):
info = self.process_address(**{'place:en': 'Honu Lulu'})
assert 'place' not in info
-
def test_process_place_place_empty(self):
info = self.process_address(place='🜵')
assert 'place' not in info
-
def test_process_place_address_terms(self):
info = self.process_address(country='de', city='Zwickau', state='Sachsen',
suburb='Zwickau', street='Hauptstr',
city = self.name_token_set('ZWICKAU', '#ZWICKAU')
state = self.name_token_set('SACHSEN', '#SACHSEN')
- result = {k: eval(v) for k,v in info['addr'].items()}
+ result = {k: eval(v) for k, v in info['addr'].items()}
assert result == {'city': city, 'suburb': city, 'state': state}
-
def test_process_place_multiple_address_terms(self):
info = self.process_address(**{'city': 'Bruxelles', 'city:de': 'Brüssel'})
- result = {k: eval(v) for k,v in info['addr'].items()}
+ result = {k: eval(v) for k, v in info['addr'].items()}
assert result == {'city': self.name_token_set('Bruxelles', '#Bruxelles')}
-
def test_process_place_address_terms_empty(self):
info = self.process_address(country='de', city=' ', street='Hauptstr',
full='right behind the church')
def setup(self, analyzer, sql_functions):
hnr = {'step': 'clean-housenumbers',
'filter-kind': ['housenumber', 'conscriptionnumber', 'streetnumber']}
- with analyzer(trans=(":: upper()", "'🜵' > ' '"), sanitizers=[hnr], with_housenumber=True) as anl:
+ with analyzer(trans=(":: upper()", "'🜵' > ' '"), sanitizers=[hnr],
+ with_housenumber=True) as anl:
self.analyzer = anl
yield anl
-
@pytest.fixture
def getorcreate_hnr_id(self, temp_db_cursor):
- temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION create_analyzed_hnr_id(norm_term TEXT, lookup_terms TEXT[])
- RETURNS INTEGER AS $$
- SELECT -nextval('seq_word')::INTEGER; $$ LANGUAGE SQL""")
-
+ temp_db_cursor.execute("""
+ CREATE OR REPLACE FUNCTION create_analyzed_hnr_id(norm_term TEXT, lookup_terms TEXT[])
+ RETURNS INTEGER AS $$
+ SELECT -nextval('seq_word')::INTEGER; $$ LANGUAGE SQL""")
def process_address(self, **kwargs):
return self.analyzer.process_place(PlaceInfo({'address': kwargs}))
-
def name_token_set(self, *expected_terms):
tokens = self.analyzer.get_word_token_info(expected_terms)
for token in tokens:
return set((t[2] for t in tokens))
-
@pytest.mark.parametrize('hnr', ['123 a', '1', '101'])
def test_process_place_housenumbers_simple(self, hnr, getorcreate_hnr_id):
info = self.process_address(housenumber=hnr)
assert info['hnr'] == hnr.upper()
assert info['hnr_tokens'] == "{-1}"
-
def test_process_place_housenumbers_duplicates(self, getorcreate_hnr_id):
info = self.process_address(housenumber='134',
conscriptionnumber='134',
assert set(info['hnr'].split(';')) == set(('134', '99 A'))
assert info['hnr_tokens'] == "{-1,-2}"
-
def test_process_place_housenumbers_cached(self, getorcreate_hnr_id):
info = self.process_address(housenumber="45")
assert info['hnr_tokens'] == "{-1}"
table_factory('search_name', 'place_id BIGINT, name_vector INT[]')
self.tok = tokenizer_factory()
-
@pytest.fixture
def search_entry(self, temp_db_cursor):
place_id = itertools.count(1000)
return _insert
-
@pytest.fixture(params=['simple', 'analyzed'])
def add_housenumber(self, request, word_table):
if request.param == 'simple':
return _make
-
@pytest.mark.parametrize('hnr', ('1a', '1234567', '34 5'))
def test_remove_unused_housenumbers(self, add_housenumber, word_table, hnr):
word_table.add_housenumber(1000, hnr)
self.tok.update_word_tokens()
assert word_table.count_housenumbers() == 0
-
def test_keep_unused_numeral_housenumbers(self, add_housenumber, word_table):
add_housenumber(1000, '5432')
self.tok.update_word_tokens()
assert word_table.count_housenumbers() == 1
-
- def test_keep_housenumbers_from_search_name_table(self, add_housenumber, word_table, search_entry):
+ def test_keep_housenumbers_from_search_name_table(self, add_housenumber,
+ word_table, search_entry):
add_housenumber(9999, '5432a')
add_housenumber(9991, '9 a')
search_entry(123, 9999, 34)
self.tok.update_word_tokens()
assert word_table.count_housenumbers() == 1
-
- def test_keep_housenumbers_from_placex_table(self, add_housenumber, word_table, placex_table):
+ def test_keep_housenumbers_from_placex_table(self, add_housenumber, word_table,
+ placex_table):
add_housenumber(9999, '5432a')
add_housenumber(9990, '34z')
placex_table.add(housenumber='34z')
self.tok.update_word_tokens()
assert word_table.count_housenumbers() == 1
-
- def test_keep_housenumbers_from_placex_table_hnr_list(self, add_housenumber, word_table, placex_table):
+ def test_keep_housenumbers_from_placex_table_hnr_list(self, add_housenumber,
+ word_table, placex_table):
add_housenumber(9991, '9 b')
add_housenumber(9990, '34z')
placex_table.add(housenumber='9 a;9 b;9 c')
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for converting a config file to ICU rules.
CONFIG_SECTIONS = ('normalization', 'transliteration', 'token-analysis')
+
class TestIcuRuleLoader:
@pytest.fixture(autouse=True)
def init_env(self, project_env):
self.project_env = project_env
-
def write_config(self, content):
(self.project_env.project_dir / 'icu_tokenizer.yaml').write_text(dedent(content))
-
def config_rules(self, *variants):
content = dedent("""\
normalization:
content += '\n'.join((" - " + s for s in variants)) + '\n'
self.write_config(content)
-
def get_replacements(self, *variants):
self.config_rules(*variants)
loader = ICURuleLoader(self.project_env)
rules = loader.analysis[None].config['replacements']
- return sorted((k, sorted(v)) for k,v in rules)
-
+ return sorted((k, sorted(v)) for k, v in rules)
def test_empty_rule_set(self):
self.write_config("""\
assert rules.get_normalization_rules() == ''
assert rules.get_transliteration_rules() == ''
-
@pytest.mark.parametrize("section", CONFIG_SECTIONS)
def test_missing_section(self, section):
- rule_cfg = { s: [] for s in CONFIG_SECTIONS if s != section}
+ rule_cfg = {s: [] for s in CONFIG_SECTIONS if s != section}
self.write_config(yaml.dump(rule_cfg))
with pytest.raises(UsageError):
ICURuleLoader(self.project_env)
-
def test_get_search_rules(self):
self.config_rules()
loader = ICURuleLoader(self.project_env)
assert trans.transliterate(" Αθήνα ") == " athēna "
assert trans.transliterate(" проспект ") == " prospekt "
-
def test_get_normalization_rules(self):
self.config_rules()
loader = ICURuleLoader(self.project_env)
assert trans.transliterate(" проспект-Prospekt ") == " проспект prospekt "
-
def test_get_transliteration_rules(self):
self.config_rules()
loader = ICURuleLoader(self.project_env)
assert trans.transliterate(" проспект-Prospekt ") == " prospekt Prospekt "
-
def test_transliteration_rules_from_file(self):
self.write_config("""\
normalization:
assert trans.transliterate(" axxt ") == " byt "
-
def test_search_rules(self):
self.config_rules('~street => s,st', 'master => mstr')
proc = ICURuleLoader(self.project_env).make_token_analysis()
assert proc.search.transliterate('Earnes St').strip() == 'earnes st'
assert proc.search.transliterate('Nostreet').strip() == 'nostreet'
-
@pytest.mark.parametrize("variant", ['foo > bar', 'foo -> bar -> bar',
'~foo~ -> bar', 'fo~ o -> bar'])
def test_invalid_variant_description(self, variant):
assert repl == [(' foo ', [' bar', ' foo'])]
-
def test_replace_full(self):
repl = self.get_replacements("foo => bar")
assert repl == [(' foo ', [' bar'])]
-
def test_add_suffix_no_decompose(self):
repl = self.get_replacements("~berg |-> bg")
assert repl == [(' berg ', [' berg', ' bg']),
('berg ', ['berg', 'bg'])]
-
def test_replace_suffix_no_decompose(self):
repl = self.get_replacements("~berg |=> bg")
- assert repl == [(' berg ', [' bg']),('berg ', ['bg'])]
-
+ assert repl == [(' berg ', [' bg']), ('berg ', ['bg'])]
def test_add_suffix_decompose(self):
repl = self.get_replacements("~berg -> bg")
assert repl == [(' berg ', [' berg', ' bg', 'berg', 'bg']),
('berg ', [' berg', ' bg', 'berg', 'bg'])]
-
def test_replace_suffix_decompose(self):
repl = self.get_replacements("~berg => bg")
assert repl == [(' berg ', [' bg', 'bg']),
('berg ', [' bg', 'bg'])]
-
def test_add_prefix_no_compose(self):
repl = self.get_replacements("hinter~ |-> hnt")
assert repl == [(' hinter', [' hinter', ' hnt']),
(' hinter ', [' hinter', ' hnt'])]
-
def test_replace_prefix_no_compose(self):
repl = self.get_replacements("hinter~ |=> hnt")
- assert repl == [(' hinter', [' hnt']), (' hinter ', [' hnt'])]
-
+ assert repl == [(' hinter', [' hnt']), (' hinter ', [' hnt'])]
def test_add_prefix_compose(self):
repl = self.get_replacements("hinter~-> h")
assert repl == [(' hinter', [' h', ' h ', ' hinter', ' hinter ']),
(' hinter ', [' h', ' h', ' hinter', ' hinter'])]
-
def test_replace_prefix_compose(self):
repl = self.get_replacements("hinter~=> h")
assert repl == [(' hinter', [' h', ' h ']),
(' hinter ', [' h', ' h'])]
-
def test_add_beginning_only(self):
repl = self.get_replacements("^Premier -> Pr")
assert repl == [('^ premier ', ['^ pr', '^ premier'])]
-
def test_replace_beginning_only(self):
repl = self.get_replacements("^Premier => Pr")
assert repl == [('^ premier ', ['^ pr'])]
-
def test_add_final_only(self):
repl = self.get_replacements("road$ -> rd")
assert repl == [(' road ^', [' rd ^', ' road ^'])]
-
def test_replace_final_only(self):
repl = self.get_replacements("road$ => rd")
assert repl == [(' road ^', [' rd ^'])]
-
def test_decompose_only(self):
repl = self.get_replacements("~foo -> foo")
assert repl == [(' foo ', [' foo', 'foo']),
('foo ', [' foo', 'foo'])]
-
def test_add_suffix_decompose_end_only(self):
repl = self.get_replacements("~berg |-> bg", "~berg$ -> bg")
('berg ', ['berg', 'bg']),
('berg ^', [' berg ^', ' bg ^', 'berg ^', 'bg ^'])]
-
def test_replace_suffix_decompose_end_only(self):
repl = self.get_replacements("~berg |=> bg", "~berg$ => bg")
('berg ', ['bg']),
('berg ^', [' bg ^', 'bg ^'])]
-
def test_add_multiple_suffix(self):
repl = self.get_replacements("~berg,~burg -> bg")
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for execution of the sanitztion step.
def test_sanitizer_default(def_config):
san = sanitizer.PlaceSanitizer([{'step': 'split-name-list'}], def_config)
- name, address = san.process_names(PlaceInfo({'name': {'name:de:de': '1;2;3'},
- 'address': {'street': 'Bald'}}))
+ name, address = san.process_names(PlaceInfo({'name': {'name:de:de': '1;2;3'},
+ 'address': {'street': 'Bald'}}))
assert len(name) == 3
assert all(isinstance(n, sanitizer.PlaceName) for n in name)
- assert all(n.kind == 'name' for n in name)
- assert all(n.suffix == 'de:de' for n in name)
+ assert all(n.kind == 'name' for n in name)
+ assert all(n.suffix == 'de:de' for n in name)
assert len(address) == 1
assert all(isinstance(n, sanitizer.PlaceName) for n in address)
def test_sanitizer_empty_list(def_config, rules):
san = sanitizer.PlaceSanitizer(rules, def_config)
- name, address = san.process_names(PlaceInfo({'name': {'name:de:de': '1;2;3'}}))
+ name, address = san.process_names(PlaceInfo({'name': {'name:de:de': '1;2;3'}}))
assert len(name) == 1
assert all(isinstance(n, sanitizer.PlaceName) for n in name)
def test_sanitizer_missing_step_definition(def_config):
with pytest.raises(UsageError):
- san = sanitizer.PlaceSanitizer([{'id': 'split-name-list'}], def_config)
+ sanitizer.PlaceSanitizer([{'id': 'split-name-list'}], def_config)
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for special postcode analysis and variant generation.
import nominatim_db.tokenizer.token_analysis.postcodes as module
from nominatim_db.data.place_name import PlaceName
-from nominatim_db.errors import UsageError
DEFAULT_NORMALIZATION = """ :: NFD ();
'🜳' > ' ';
'🜵' > ' ';
"""
+
@pytest.fixture
def analyser():
- rules = { 'analyzer': 'postcodes'}
+ rules = {'analyzer': 'postcodes'}
config = module.configure(rules, DEFAULT_NORMALIZATION)
trans = Transliterator.createFromRules("test_trans", DEFAULT_TRANSLITERATION)
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for import name normalisation and variant generation.
'🜵' > ' ';
"""
+
def make_analyser(*variants, variant_only=False):
- rules = { 'analyzer': 'generic', 'variants': [{'words': variants}]}
+ rules = {'analyzer': 'generic', 'variants': [{'words': variants}]}
if variant_only:
rules['mode'] = 'variant-only'
trans = Transliterator.createFromRules("test_trans", DEFAULT_TRANSLITERATION)
def test_no_variants():
- rules = { 'analyzer': 'generic' }
+ rules = {'analyzer': 'generic'}
trans = Transliterator.createFromRules("test_trans", DEFAULT_TRANSLITERATION)
norm = Transliterator.createFromRules("test_norm", DEFAULT_NORMALIZATION)
config = module.configure(rules, norm, trans)
VARIANT_TESTS = [
-(('~strasse,~straße -> str', '~weg => weg'), "hallo", {'hallo'}),
-(('weg => wg',), "holzweg", {'holzweg'}),
-(('weg -> wg',), "holzweg", {'holzweg'}),
-(('~weg => weg',), "holzweg", {'holz weg', 'holzweg'}),
-(('~weg -> weg',), "holzweg", {'holz weg', 'holzweg'}),
-(('~weg => w',), "holzweg", {'holz w', 'holzw'}),
-(('~weg -> w',), "holzweg", {'holz weg', 'holzweg', 'holz w', 'holzw'}),
-(('~weg => weg',), "Meier Weg", {'meier weg', 'meierweg'}),
-(('~weg -> weg',), "Meier Weg", {'meier weg', 'meierweg'}),
-(('~weg => w',), "Meier Weg", {'meier w', 'meierw'}),
-(('~weg -> w',), "Meier Weg", {'meier weg', 'meierweg', 'meier w', 'meierw'}),
-(('weg => wg',), "Meier Weg", {'meier wg'}),
-(('weg -> wg',), "Meier Weg", {'meier weg', 'meier wg'}),
-(('~strasse,~straße -> str', '~weg => weg'), "Bauwegstraße",
+ (('~strasse,~straße -> str', '~weg => weg'), "hallo", {'hallo'}),
+ (('weg => wg',), "holzweg", {'holzweg'}),
+ (('weg -> wg',), "holzweg", {'holzweg'}),
+ (('~weg => weg',), "holzweg", {'holz weg', 'holzweg'}),
+ (('~weg -> weg',), "holzweg", {'holz weg', 'holzweg'}),
+ (('~weg => w',), "holzweg", {'holz w', 'holzw'}),
+ (('~weg -> w',), "holzweg", {'holz weg', 'holzweg', 'holz w', 'holzw'}),
+ (('~weg => weg',), "Meier Weg", {'meier weg', 'meierweg'}),
+ (('~weg -> weg',), "Meier Weg", {'meier weg', 'meierweg'}),
+ (('~weg => w',), "Meier Weg", {'meier w', 'meierw'}),
+ (('~weg -> w',), "Meier Weg", {'meier weg', 'meierweg', 'meier w', 'meierw'}),
+ (('weg => wg',), "Meier Weg", {'meier wg'}),
+ (('weg -> wg',), "Meier Weg", {'meier weg', 'meier wg'}),
+ (('~strasse,~straße -> str', '~weg => weg'), "Bauwegstraße",
{'bauweg straße', 'bauweg str', 'bauwegstraße', 'bauwegstr'}),
-(('am => a', 'bach => b'), "am bach", {'a b'}),
-(('am => a', '~bach => b'), "am bach", {'a b'}),
-(('am -> a', '~bach -> b'), "am bach", {'am bach', 'a bach', 'am b', 'a b'}),
-(('am -> a', '~bach -> b'), "ambach", {'ambach', 'am bach', 'amb', 'am b'}),
-(('saint -> s,st', 'street -> st'), "Saint Johns Street",
+ (('am => a', 'bach => b'), "am bach", {'a b'}),
+ (('am => a', '~bach => b'), "am bach", {'a b'}),
+ (('am -> a', '~bach -> b'), "am bach", {'am bach', 'a bach', 'am b', 'a b'}),
+ (('am -> a', '~bach -> b'), "ambach", {'ambach', 'am bach', 'amb', 'am b'}),
+ (('saint -> s,st', 'street -> st'), "Saint Johns Street",
{'saint johns street', 's johns street', 'st johns street',
'saint johns st', 's johns st', 'st johns st'}),
-(('river$ -> r',), "River Bend Road", {'river bend road'}),
-(('river$ -> r',), "Bent River", {'bent river', 'bent r'}),
-(('^north => n',), "North 2nd Street", {'n 2nd street'}),
-(('^north => n',), "Airport North", {'airport north'}),
-(('am -> a',), "am am am am am am am am", {'am am am am am am am am'}),
-(('am => a',), "am am am am am am am am", {'a a a a a a a a'})
-]
+ (('river$ -> r',), "River Bend Road", {'river bend road'}),
+ (('river$ -> r',), "Bent River", {'bent river', 'bent r'}),
+ (('^north => n',), "North 2nd Street", {'n 2nd street'}),
+ (('^north => n',), "Airport North", {'airport north'}),
+ (('am -> a',), "am am am am am am am am", {'am am am am am am am am'}),
+ (('am => a',), "am am am am am am am am", {'a a a a a a a a'})
+ ]
+
@pytest.mark.parametrize("rules,name,variants", VARIANT_TESTS)
def test_variants(rules, name, variants):
VARIANT_ONLY_TESTS = [
-(('weg => wg',), "hallo", set()),
-(('weg => wg',), "Meier Weg", {'meier wg'}),
-(('weg -> wg',), "Meier Weg", {'meier wg'}),
-]
+ (('weg => wg',), "hallo", set()),
+ (('weg => wg',), "Meier Weg", {'meier wg'}),
+ (('weg -> wg',), "Meier Weg", {'meier wg'}),
+ ]
+
@pytest.mark.parametrize("rules,name,variants", VARIANT_ONLY_TESTS)
def test_variants_only(rules, name, variants):
@staticmethod
def configure_rules(*variants):
- rules = { 'analyzer': 'generic', 'variants': [{'words': variants}]}
+ rules = {'analyzer': 'generic', 'variants': [{'words': variants}]}
trans = Transliterator.createFromRules("test_trans", DEFAULT_TRANSLITERATION)
norm = Transliterator.createFromRules("test_norm", DEFAULT_NORMALIZATION)
return module.configure(rules, norm, trans)
-
def get_replacements(self, *variants):
config = self.configure_rules(*variants)
- return sorted((k, sorted(v)) for k,v in config['replacements'])
-
+ return sorted((k, sorted(v)) for k, v in config['replacements'])
@pytest.mark.parametrize("variant", ['foo > bar', 'foo -> bar -> bar',
'~foo~ -> bar', 'fo~ o -> bar'])
with pytest.raises(UsageError):
self.configure_rules(variant)
-
@pytest.mark.parametrize("rule", ["!!! -> bar", "bar => !!!"])
def test_ignore_unnormalizable_terms(self, rule):
repl = self.get_replacements(rule)
assert repl == []
-
def test_add_full(self):
repl = self.get_replacements("foo -> bar")
assert repl == [(' foo ', [' bar', ' foo'])]
-
def test_replace_full(self):
repl = self.get_replacements("foo => bar")
assert repl == [(' foo ', [' bar'])]
-
def test_add_suffix_no_decompose(self):
repl = self.get_replacements("~berg |-> bg")
assert repl == [(' berg ', [' berg', ' bg']),
('berg ', ['berg', 'bg'])]
-
def test_replace_suffix_no_decompose(self):
repl = self.get_replacements("~berg |=> bg")
- assert repl == [(' berg ', [' bg']),('berg ', ['bg'])]
-
+ assert repl == [(' berg ', [' bg']), ('berg ', ['bg'])]
def test_add_suffix_decompose(self):
repl = self.get_replacements("~berg -> bg")
assert repl == [(' berg ', [' berg', ' bg', 'berg', 'bg']),
('berg ', [' berg', ' bg', 'berg', 'bg'])]
-
def test_replace_suffix_decompose(self):
repl = self.get_replacements("~berg => bg")
assert repl == [(' berg ', [' bg', 'bg']),
('berg ', [' bg', 'bg'])]
-
def test_add_prefix_no_compose(self):
repl = self.get_replacements("hinter~ |-> hnt")
assert repl == [(' hinter', [' hinter', ' hnt']),
(' hinter ', [' hinter', ' hnt'])]
-
def test_replace_prefix_no_compose(self):
repl = self.get_replacements("hinter~ |=> hnt")
- assert repl == [(' hinter', [' hnt']), (' hinter ', [' hnt'])]
-
+ assert repl == [(' hinter', [' hnt']), (' hinter ', [' hnt'])]
def test_add_prefix_compose(self):
repl = self.get_replacements("hinter~-> h")
assert repl == [(' hinter', [' h', ' h ', ' hinter', ' hinter ']),
(' hinter ', [' h', ' h', ' hinter', ' hinter'])]
-
def test_replace_prefix_compose(self):
repl = self.get_replacements("hinter~=> h")
assert repl == [(' hinter', [' h', ' h ']),
(' hinter ', [' h', ' h'])]
-
def test_add_beginning_only(self):
repl = self.get_replacements("^Premier -> Pr")
assert repl == [('^ premier ', ['^ pr', '^ premier'])]
-
def test_replace_beginning_only(self):
repl = self.get_replacements("^Premier => Pr")
assert repl == [('^ premier ', ['^ pr'])]
-
def test_add_final_only(self):
repl = self.get_replacements("road$ -> rd")
assert repl == [(' road ^', [' rd ^', ' road ^'])]
-
def test_replace_final_only(self):
repl = self.get_replacements("road$ => rd")
assert repl == [(' road ^', [' rd ^'])]
-
def test_decompose_only(self):
repl = self.get_replacements("~foo -> foo")
assert repl == [(' foo ', [' foo', 'foo']),
('foo ', [' foo', 'foo'])]
-
def test_add_suffix_decompose_end_only(self):
repl = self.get_replacements("~berg |-> bg", "~berg$ -> bg")
('berg ', ['berg', 'bg']),
('berg ^', [' berg ^', ' bg ^', 'berg ^', 'bg ^'])]
-
def test_replace_suffix_decompose_end_only(self):
repl = self.get_replacements("~berg |=> bg", "~berg$ => bg")
('berg ', ['bg']),
('berg ^', [' bg ^', 'bg ^'])]
-
@pytest.mark.parametrize('rule', ["~berg,~burg -> bg",
"~berg, ~burg -> bg",
"~berg,,~burg -> bg"])
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for generic token analysis, mutation part.
'🜵' > ' ';
"""
+
class TestMutationNoVariants:
def make_analyser(self, *mutations):
- rules = { 'analyzer': 'generic',
- 'mutations': [ {'pattern': m[0], 'replacements': m[1]}
- for m in mutations]
- }
+ rules = {'analyzer': 'generic',
+ 'mutations': [{'pattern': m[0], 'replacements': m[1]}
+ for m in mutations]
+ }
trans = Transliterator.createFromRules("test_trans", DEFAULT_TRANSLITERATION)
norm = Transliterator.createFromRules("test_norm", DEFAULT_NORMALIZATION)
config = module.configure(rules, norm, trans)
self.analysis = module.create(norm, trans, config)
-
def variants(self, name):
norm = Transliterator.createFromRules("test_norm", DEFAULT_NORMALIZATION)
return set(self.analysis.compute_variants(norm.transliterate(name).strip()))
-
@pytest.mark.parametrize('pattern', ('(capture)', ['a list']))
def test_bad_pattern(self, pattern):
with pytest.raises(UsageError):
self.make_analyser((pattern, ['b']))
-
@pytest.mark.parametrize('replacements', (None, 'a string'))
def test_bad_replacement(self, replacements):
with pytest.raises(UsageError):
self.make_analyser(('a', replacements))
-
def test_simple_replacement(self):
self.make_analyser(('a', ['b']))
assert self.variants('abba') == {'bbbb'}
assert self.variants('2 aar') == {'2 bbr'}
-
def test_multichar_replacement(self):
self.make_analyser(('1 1', ['1 1 1']))
assert self.variants('1 1456') == {'1 1 1456'}
assert self.variants('1 1 1') == {'1 1 1 1'}
-
def test_removement_replacement(self):
self.make_analyser((' ', [' ', '']))
assert self.variants('A 345') == {'a 345', 'a345'}
assert self.variants('a g b') == {'a g b', 'ag b', 'a gb', 'agb'}
-
def test_regex_pattern(self):
self.make_analyser(('[^a-z]+', ['XXX', ' ']))
assert self.variants('a-34n12') == {'aXXXnXXX', 'aXXXn', 'a nXXX', 'a n'}
-
def test_multiple_mutations(self):
self.make_analyser(('ä', ['ä', 'ae']), ('ö', ['ö', 'oe']))
from nominatim_db.tokenizer.token_analysis.simple_trie import SimpleTrie
+
def test_single_item_trie():
t = SimpleTrie([('foob', 42)])
assert t.longest_prefix('foob') == (42, 4)
assert t.longest_prefix('123foofoo', 3) == (None, 3)
+
def test_complex_item_tree():
t = SimpleTrie([('a', 1),
('b', 2),
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
import pytest
+
@pytest.fixture
def osm2pgsql_options(temp_db, tmp_path):
""" A standard set of options for osm2pgsql
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for functions to add additional data to the database.
from nominatim_db.tools import add_osm_data
+
class CaptureGetUrl:
def __init__(self, monkeypatch):
temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION flush_deleted_places()
RETURNS INTEGER AS $$ SELECT 1 $$ LANGUAGE SQL""")
+
def test_import_osm_file_simple(dsn, table_factory, osm2pgsql_options, capfd):
assert add_osm_data.add_data_from_file(dsn, Path('change.osm'), osm2pgsql_options) == 0
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for maintenance and analysis functions.
from nominatim_db.tokenizer import factory
from nominatim_db.db.sql_preprocessor import SQLPreprocessor
+
@pytest.fixture(autouse=True)
def create_placex_table(project_env, tokenizer_mock, temp_db_cursor, placex_table):
""" All tests in this module require the placex table to be set up.
class TestAdminCleanDeleted:
@pytest.fixture(autouse=True)
- def setup_polygon_delete(self, project_env, table_factory, place_table, osmline_table, temp_db_cursor, temp_db_conn, def_config, src_dir):
+ def setup_polygon_delete(self, project_env, table_factory, place_table,
+ osmline_table, temp_db_cursor, temp_db_conn, def_config, src_dir):
""" Set up place_force_delete function and related tables
"""
self.project_env = project_env
class TEXT NOT NULL,
type TEXT NOT NULL""",
((100, 'N', 'boundary', 'administrative'),
- (145, 'N', 'boundary', 'administrative'),
- (175, 'R', 'landcover', 'grass')))
- temp_db_cursor.execute("""INSERT INTO placex (place_id, osm_id, osm_type, class, type, indexed_date, indexed_status)
- VALUES(1, 100, 'N', 'boundary', 'administrative', current_date - INTERVAL '1 month', 1),
- (2, 145, 'N', 'boundary', 'administrative', current_date - INTERVAL '3 month', 1),
- (3, 175, 'R', 'landcover', 'grass', current_date - INTERVAL '3 months', 1)""")
+ (145, 'N', 'boundary', 'administrative'),
+ (175, 'R', 'landcover', 'grass')))
+ temp_db_cursor.execute("""
+ INSERT INTO placex (place_id, osm_id, osm_type, class, type,
+ indexed_date, indexed_status)
+ VALUES(1, 100, 'N', 'boundary', 'administrative', current_date - INTERVAL '1 month', 1),
+ (2, 145, 'N', 'boundary', 'administrative', current_date - INTERVAL '3 month', 1),
+ (3, 175, 'R', 'landcover', 'grass', current_date - INTERVAL '3 months', 1)""")
# set up tables and triggers for utils function
table_factory('place_to_be_deleted',
"""osm_id BIGINT,
sqlproc = SQLPreprocessor(temp_db_conn, def_config)
sqlproc.run_sql_file(temp_db_conn, 'functions/utils.sql')
def_config.lib_dir.sql = orig_sql
-
def test_admin_clean_deleted_no_records(self):
admin.clean_deleted_relations(self.project_env, age='1 year')
- assert self.temp_db_cursor.row_set('SELECT osm_id, osm_type, class, type, indexed_status FROM placex') == {(100, 'N', 'boundary', 'administrative', 1),
- (145, 'N', 'boundary', 'administrative', 1),
- (175, 'R', 'landcover', 'grass', 1)}
- assert self.temp_db_cursor.table_rows('import_polygon_delete') == 3
+ rowset = self.temp_db_cursor.row_set(
+ 'SELECT osm_id, osm_type, class, type, indexed_status FROM placex')
+
+ assert rowset == {(100, 'N', 'boundary', 'administrative', 1),
+ (145, 'N', 'boundary', 'administrative', 1),
+ (175, 'R', 'landcover', 'grass', 1)}
+ assert self.temp_db_cursor.table_rows('import_polygon_delete') == 3
@pytest.mark.parametrize('test_age', ['T week', '1 welk', 'P1E'])
def test_admin_clean_deleted_bad_age(self, test_age):
with pytest.raises(UsageError):
- admin.clean_deleted_relations(self.project_env, age = test_age)
-
+ admin.clean_deleted_relations(self.project_env, age=test_age)
def test_admin_clean_deleted_partial(self):
- admin.clean_deleted_relations(self.project_env, age = '2 months')
- assert self.temp_db_cursor.row_set('SELECT osm_id, osm_type, class, type, indexed_status FROM placex') == {(100, 'N', 'boundary', 'administrative', 1),
- (145, 'N', 'boundary', 'administrative', 100),
- (175, 'R', 'landcover', 'grass', 100)}
+ admin.clean_deleted_relations(self.project_env, age='2 months')
+
+ rowset = self.temp_db_cursor.row_set(
+ 'SELECT osm_id, osm_type, class, type, indexed_status FROM placex')
+
+ assert rowset == {(100, 'N', 'boundary', 'administrative', 1),
+ (145, 'N', 'boundary', 'administrative', 100),
+ (175, 'R', 'landcover', 'grass', 100)}
assert self.temp_db_cursor.table_rows('import_polygon_delete') == 1
@pytest.mark.parametrize('test_age', ['1 week', 'P3D', '5 hours'])
def test_admin_clean_deleted(self, test_age):
- admin.clean_deleted_relations(self.project_env, age = test_age)
- assert self.temp_db_cursor.row_set('SELECT osm_id, osm_type, class, type, indexed_status FROM placex') == {(100, 'N', 'boundary', 'administrative', 100),
- (145, 'N', 'boundary', 'administrative', 100),
- (175, 'R', 'landcover', 'grass', 100)}
+ admin.clean_deleted_relations(self.project_env, age=test_age)
+
+ rowset = self.temp_db_cursor.row_set(
+ 'SELECT osm_id, osm_type, class, type, indexed_status FROM placex')
+
+ assert rowset == {(100, 'N', 'boundary', 'administrative', 100),
+ (145, 'N', 'boundary', 'administrative', 100),
+ (175, 'R', 'landcover', 'grass', 100)}
assert self.temp_db_cursor.table_rows('import_polygon_delete') == 0
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for database integrity checks.
from nominatim_db.tools import check_database as chkdb
import nominatim_db.version
+
def test_check_database_unknown_db(def_config, monkeypatch):
monkeypatch.setenv('NOMINATIM_DATABASE_DSN', 'pgsql:dbname=fjgkhughwgh2423gsags')
assert chkdb.check_database(def_config) == 1
str(nominatim_db.version.NOMINATIM_VERSION))
assert chkdb.check_database_version(temp_db_conn, def_config) == chkdb.CheckState.OK
+
def test_check_database_version_bad(property_table, temp_db_conn, def_config):
property_table.set('database_version', '3.9.9-9')
assert chkdb.check_database_version(temp_db_conn, def_config) == chkdb.CheckState.FATAL
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for functions to import a new database.
from pathlib import Path
import pytest
-import pytest_asyncio
+import pytest_asyncio # noqa
import psycopg
from psycopg import sql as pysql
from nominatim_db.tools import database_import
from nominatim_db.errors import UsageError
+
class TestDatabaseSetup:
DBNAME = 'test_nominatim_python_unittest'
with conn.cursor() as cur:
cur.execute(f'DROP DATABASE IF EXISTS {self.DBNAME}')
-
@pytest.fixture
def cursor(self):
with psycopg.connect(dbname=self.DBNAME) as conn:
with conn.cursor() as cur:
yield cur
-
def conn(self):
return psycopg.connect(dbname=self.DBNAME)
-
def test_setup_skeleton(self):
database_import.setup_database_skeleton(f'dbname={self.DBNAME}')
with conn.cursor() as cur:
cur.execute('CREATE TABLE t (h HSTORE, geom GEOMETRY(Geometry, 4326))')
-
def test_unsupported_pg_version(self, monkeypatch):
monkeypatch.setattr(database_import, 'POSTGRESQL_REQUIRED_VERSION', (100, 4))
with pytest.raises(UsageError, match='PostgreSQL server is too old.'):
database_import.setup_database_skeleton(f'dbname={self.DBNAME}')
-
def test_create_db_explicit_ro_user(self):
database_import.setup_database_skeleton(f'dbname={self.DBNAME}',
rouser='postgres')
-
def test_create_db_missing_ro_user(self):
with pytest.raises(UsageError, match='Missing read-only user.'):
database_import.setup_database_skeleton(f'dbname={self.DBNAME}',
rouser='sdfwkjkjgdugu2;jgsafkljas;')
-
def test_setup_extensions_old_postgis(self, monkeypatch):
monkeypatch.setattr(database_import, 'POSTGIS_REQUIRED_VERSION', (50, 50))
@pytest.mark.parametrize("threads", (1, 5))
@pytest.mark.asyncio
async def test_load_data(dsn, place_row, placex_table, osmline_table,
- temp_db_cursor, threads):
+ temp_db_cursor, threads):
for func in ('precompute_words', 'getorcreate_housenumber_id', 'make_standard_name'):
temp_db_cursor.execute(pysql.SQL("""CREATE FUNCTION {} (src TEXT)
RETURNS TEXT AS $$ SELECT 'a'::TEXT $$ LANGUAGE SQL
self.config = def_config
-
def write_sql(self, fname, content):
(self.config.lib_dir.sql / fname).write_text(content)
-
@pytest.mark.parametrize("reverse", [True, False])
def test_create_tables(self, temp_db_conn, temp_db_cursor, reverse):
self.write_sql('tables.sql',
temp_db_cursor.scalar('SELECT test()') == reverse
-
def test_create_table_triggers(self, temp_db_conn, temp_db_cursor):
self.write_sql('table-triggers.sql',
"""CREATE FUNCTION test() RETURNS TEXT
temp_db_cursor.scalar('SELECT test()') == 'a'
-
def test_create_partition_tables(self, temp_db_conn, temp_db_cursor):
self.write_sql('partition-tables.src.sql',
"""CREATE FUNCTION test() RETURNS TEXT
temp_db_cursor.scalar('SELECT test()') == 'b'
-
@pytest.mark.parametrize("drop", [True, False])
@pytest.mark.asyncio
async def test_create_search_indices(self, temp_db_conn, temp_db_cursor, drop):
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for tools.exec_utils module.
"""
-from pathlib import Path
-import subprocess
-
-import pytest
-
-from nominatim_db.config import Configuration
import nominatim_db.tools.exec_utils as exec_utils
+
def test_run_osm2pgsql(osm2pgsql_options):
osm2pgsql_options['append'] = False
osm2pgsql_options['import_file'] = 'foo.bar'
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for freeze functions (removing unused database parts).
'wikipedia_article', 'wikipedia_redirect'
]
+
def test_drop_tables(temp_db_conn, temp_db_cursor, table_factory):
for table in NOMINATIM_RUNTIME_TABLES + NOMINATIM_DROP_TABLES:
table_factory(table)
assert freeze.is_frozen(temp_db_conn)
+
def test_drop_flatnode_file_no_file():
freeze.drop_flatnode_file(None)
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for import special phrases methods
of the class SPImporter.
"""
-from shutil import copyfile
import pytest
from nominatim_db.tools.special_phrases.sp_importer import SPImporter
from nominatim_db.tools.special_phrases.sp_wiki_loader import SPWikiLoader
from nominatim_db.tools.special_phrases.special_phrase import SpecialPhrase
-from nominatim_db.errors import UsageError
-from cursor import CursorForTesting
@pytest.fixture
def sp_importer(temp_db_conn, def_config, monkeypatch):
contained_table = sp_importer.table_phrases_to_delete.pop()
assert contained_table == 'place_classtype_testclasstypetable'
+
def test_check_sanity_class(sp_importer):
"""
Check for _check_sanity() method.
assert sp_importer._check_sanity(SpecialPhrase('en', 'class', 'type', ''))
+
def test_load_white_and_black_lists(sp_importer):
"""
Test that _load_white_and_black_lists() well return
assert check_placeid_and_centroid_indexes(temp_db_cursor, phrase_class, phrase_type)
+
def test_create_place_classtype_table(temp_db_conn, temp_db_cursor, placex_table, sp_importer):
"""
Test that _create_place_classtype_table() create
assert check_table_exist(temp_db_cursor, phrase_class, phrase_type)
+
def test_grant_access_to_web_user(temp_db_conn, temp_db_cursor, table_factory,
def_config, sp_importer):
"""
sp_importer._grant_access_to_webuser(phrase_class, phrase_type)
temp_db_conn.commit()
- assert check_grant_access(temp_db_cursor, def_config.DATABASE_WEBUSER, phrase_class, phrase_type)
+ assert check_grant_access(temp_db_cursor, def_config.DATABASE_WEBUSER,
+ phrase_class, phrase_type)
+
def test_create_place_classtype_table_and_indexes(
temp_db_cursor, def_config, placex_table,
assert check_placeid_and_centroid_indexes(temp_db_cursor, pair[0], pair[1])
assert check_grant_access(temp_db_cursor, def_config.DATABASE_WEBUSER, pair[0], pair[1])
+
def test_remove_non_existent_tables_from_db(sp_importer, default_phrases,
temp_db_conn, temp_db_cursor):
"""
temp_db_conn.commit()
assert temp_db_cursor.row_set(query_tables) \
- == {('place_classtype_testclasstypetable_to_keep', )}
+ == {('place_classtype_testclasstypetable_to_keep', )}
@pytest.mark.parametrize("should_replace", [(True), (False)])
It should also update the database well by deleting or preserving existing entries
of the database.
"""
- #Add some data to the database before execution in order to test
- #what is deleted and what is preserved.
+ # Add some data to the database before execution in order to test
+ # what is deleted and what is preserved.
table_factory('place_classtype_amenity_animal_shelter')
table_factory('place_classtype_wrongclass_wrongtype')
if should_replace:
assert not temp_db_cursor.table_exists('place_classtype_wrongclass_wrongtype')
+
def check_table_exist(temp_db_cursor, phrase_class, phrase_type):
"""
Verify that the place_classtype table exists for the given
AND privilege_type='SELECT'""".format(table_name, user))
return temp_db_cursor.fetchone()
+
def check_placeid_and_centroid_indexes(temp_db_cursor, phrase_class, phrase_type):
"""
Check that the place_id index and centroid index exist for the
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for migration functions
from nominatim_db.tools import migration
from nominatim_db.errors import UsageError
-from nominatim_db.db.connection import server_version_tuple
import nominatim_db.version
+
class DummyTokenizer:
def update_sql_functions(self, config):
str(nominatim_db.version.NominatimVersion(*oldversion)))
done = {'old': False, 'new': False}
+
def _migration(**_):
""" Dummy migration"""
done['new'] = True
assert property_table.get('database_version') == str(nominatim_db.version.NOMINATIM_VERSION)
-###### Tests for specific migrations
+# Tests for specific migrations
#
# Each migration should come with two tests:
# 1. Test that migration from old to new state works as expected.
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for functions to maintain the artificial postcode table.
from nominatim_db.data import country_info
import dummy_tokenizer
+
class MockPostcodeTable:
""" A location_postcode table for testing.
"""
RETURNS TEXT AS $$ BEGIN RETURN postcode; END; $$ LANGUAGE plpgsql;
CREATE OR REPLACE FUNCTION get_country_code(place geometry)
- RETURNS TEXT AS $$ BEGIN
+ RETURNS TEXT AS $$ BEGIN
RETURN null;
END; $$ LANGUAGE plpgsql;
""")
(country, postcode, x, y))
self.conn.commit()
-
@property
def row_set(self):
with self.conn.cursor() as cur:
('xx', 'CD 4511', -10, -5)}
-def test_postcodes_extern_bad_column(dsn, postcode_table, tmp_path,
+def test_postcodes_extern_bad_column(dsn, postcode_table, tmp_path,
insert_implicit_postcode, tokenizer):
insert_implicit_postcode(1, 'xx', 'POINT(10 12)', dict(postcode='AB 4511'))
assert postcode_table.row_set == {('xx', 'AB 4511', 10, 12),
('xx', 'CD 4511', -10, -5)}
+
def test_can_compute(dsn, table_factory):
assert not postcodes.can_compute(dsn)
table_factory('place')
def test_no_placex_entry(dsn, tmp_path, temp_db_cursor, place_row, postcode_table, tokenizer):
- #Rewrite the get_country_code function to verify its execution.
+ # Rewrite the get_country_code function to verify its execution.
temp_db_cursor.execute("""
CREATE OR REPLACE FUNCTION get_country_code(place geometry)
- RETURNS TEXT AS $$ BEGIN
+ RETURNS TEXT AS $$ BEGIN
RETURN 'yy';
END; $$ LANGUAGE plpgsql;
""")
assert postcode_table.row_set == {('yy', 'AB 4511', 10, 12)}
-def test_discard_badly_formatted_postcodes(dsn, tmp_path, temp_db_cursor, place_row, postcode_table, tokenizer):
- #Rewrite the get_country_code function to verify its execution.
+def test_discard_badly_formatted_postcodes(dsn, tmp_path, temp_db_cursor, place_row,
+ postcode_table, tokenizer):
+ # Rewrite the get_country_code function to verify its execution.
temp_db_cursor.execute("""
CREATE OR REPLACE FUNCTION get_country_code(place geometry)
- RETURNS TEXT AS $$ BEGIN
+ RETURNS TEXT AS $$ BEGIN
RETURN 'fr';
END; $$ LANGUAGE plpgsql;
""")
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Test for various refresh functions.
import pytest
from nominatim_db.tools import refresh
-from nominatim_db.db.connection import postgis_version_tuple
+
def test_refresh_import_wikipedia_not_existing(dsn):
assert refresh.import_wikipedia_articles(dsn, Path('.')) == 1
def test_refresh_import_secondary_importance_non_existing(dsn):
assert refresh.import_secondary_importance(dsn, Path('.')) == 1
+
def test_refresh_import_secondary_importance_testdb(dsn, src_dir, temp_db_conn, temp_db_cursor):
temp_db_cursor.execute('CREATE EXTENSION postgis')
temp_db_cursor.execute('CREATE EXTENSION postgis_raster')
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for function for importing address ranks.
"""
import json
-from pathlib import Path
import pytest
from nominatim_db.tools.refresh import load_address_levels, load_address_levels_from_config
+
def test_load_ranks_def_config(temp_db_conn, temp_db_cursor, def_config):
load_address_levels_from_config(temp_db_conn, def_config)
assert temp_db_cursor.table_rows('address_levels') > 0
+
def test_load_ranks_from_project_dir(project_env, temp_db_conn, temp_db_cursor):
test_file = project_env.project_dir / 'address-levels.json'
test_file.write_text('[{"tags":{"place":{"sea":2}}}]')
"tags": {"place": {"village": 15}}},
{"countries": ['uk', 'us'],
"tags": {"place": {"village": 16}}}
- ])
+ ])
assert temp_db_cursor.row_set('SELECT * FROM levels') == \
set([(None, 'place', 'village', 14, 14),
('de', 'place', 'village', 15, 15),
('uk', 'place', 'village', 16, 16),
('us', 'place', 'village', 16, 16),
- ])
+ ])
def test_load_ranks_default_value(temp_db_conn, temp_db_cursor):
[{"tags": {"boundary": {"": 28}}},
{"countries": ['hu'],
"tags": {"boundary": {"": 29}}}
- ])
+ ])
assert temp_db_cursor.row_set('SELECT * FROM levels') == \
set([(None, 'boundary', None, 28, 28),
('hu', 'boundary', None, 29, 29),
- ])
+ ])
def test_load_ranks_multiple_keys(temp_db_conn, temp_db_cursor):
load_address_levels(temp_db_conn, 'levels',
[{"tags": {"place": {"city": 14},
- "boundary": {"administrative2" : 4}}
- }])
+ "boundary": {"administrative2": 4}}
+ }])
assert temp_db_cursor.row_set('SELECT * FROM levels') == \
set([(None, 'place', 'city', 14, 14),
(None, 'boundary', 'administrative2', 4, 4),
- ])
+ ])
def test_load_ranks_address(temp_db_conn, temp_db_cursor):
load_address_levels(temp_db_conn, 'levels',
[{"tags": {"place": {"city": 14,
- "town" : [14, 13]}}
- }])
+ "town": [14, 13]}}
+ }])
assert temp_db_cursor.row_set('SELECT * FROM levels') == \
set([(None, 'place', 'city', 14, 14),
(None, 'place', 'town', 14, 13),
- ])
+ ])
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for creating PL/pgSQL functions for Nominatim.
from nominatim_db.tools.refresh import create_functions
+
class TestCreateFunctions:
@pytest.fixture(autouse=True)
def init_env(self, sql_preprocessor, temp_db_conn, def_config, tmp_path):
self.config = def_config
def_config.lib_dir.sql = tmp_path
-
def write_functions(self, content):
sqlfile = self.config.lib_dir.sql / 'functions.sql'
sqlfile.write_text(content)
-
def test_create_functions(self, temp_db_cursor):
self.write_functions("""CREATE OR REPLACE FUNCTION test() RETURNS INTEGER
AS $$
assert temp_db_cursor.scalar('SELECT test()') == 43
-
@pytest.mark.parametrize("dbg,ret", ((True, 43), (False, 22)))
def test_create_functions_with_template(self, temp_db_cursor, dbg, ret):
self.write_functions("""CREATE OR REPLACE FUNCTION test() RETURNS INTEGER
import pytest
-from nominatim_db.tools.refresh import import_wikipedia_articles, recompute_importance, create_functions
+from nominatim_db.tools.refresh import (import_wikipedia_articles,
+ recompute_importance,
+ create_functions)
+
@pytest.fixture
def wiki_csv(tmp_path, sql_preprocessor):
for lang, title, importance, wd in data:
writer.writerow({'language': lang, 'type': 'a',
'title': title, 'importance': str(importance),
- 'wikidata_id' : wd})
+ 'wikidata_id': wd})
return tmp_path
return _import
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for replication functionality.
<node id="100" visible="true" version="1" changeset="2047" timestamp="2006-01-27T22:09:10Z" user="Foo" uid="111" lat="48.7586670" lon="8.1343060">
</node>
</osm>
-"""
+""" # noqa
+
@pytest.fixture(autouse=True)
def setup_status_table(status_table):
pass
-### init replication
+
+# init replication
def test_init_replication_bad_base_url(monkeypatch, place_row, temp_db_conn):
place_row(osm_type='N', osm_id=100)
nominatim_db.tools.replication.init_replication(temp_db_conn, 'https://test.io')
expected_date = dt.datetime.strptime('2006-01-27T19:09:10', status.ISODATE_FORMAT)\
- .replace(tzinfo=dt.timezone.utc)
+ .replace(tzinfo=dt.timezone.utc)
assert temp_db_cursor.row_set("SELECT * FROM import_status") \
- == {(expected_date, 234, True)}
+ == {(expected_date, 234, True)}
-### checking for updates
+# checking for updates
def test_check_for_updates_empty_status_table(temp_db_conn):
assert nominatim_db.tools.replication.check_for_updates(temp_db_conn, 'https://test.io') == 254
"get_state_info",
lambda self: OsmosisState(server_sequence, date))
- assert nominatim_db.tools.replication.check_for_updates(temp_db_conn, 'https://test.io') == result
+ assert result == \
+ nominatim_db.tools.replication.check_for_updates(temp_db_conn, 'https://test.io')
-### updating
+# updating
@pytest.fixture
def update_options(tmpdir):
import_file=tmpdir / 'foo.osm',
max_diff_size=1)
+
def test_update_empty_status_table(dsn):
with pytest.raises(UsageError):
nominatim_db.tools.replication.update(dsn, {})
status.set_status(temp_db_conn, dt.datetime.now(dt.timezone.utc), seq=34, indexed=False)
assert nominatim_db.tools.replication.update(dsn, dict(indexed_only=True)) \
- == nominatim_db.tools.replication.UpdateState.MORE_PENDING
+ == nominatim_db.tools.replication.UpdateState.MORE_PENDING
def test_update_no_data_no_sleep(monkeypatch, temp_db_conn, dsn, update_options):
monkeypatch.setattr(time, 'sleep', sleeptime.append)
assert nominatim_db.tools.replication.update(dsn, update_options) \
- == nominatim_db.tools.replication.UpdateState.NO_CHANGES
+ == nominatim_db.tools.replication.UpdateState.NO_CHANGES
assert not sleeptime
monkeypatch.setattr(time, 'sleep', sleeptime.append)
assert nominatim_db.tools.replication.update(dsn, update_options) \
- == nominatim_db.tools.replication.UpdateState.NO_CHANGES
+ == nominatim_db.tools.replication.UpdateState.NO_CHANGES
assert len(sleeptime) == 1
assert sleeptime[0] < 3600
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for methods of the SPCsvLoader class.
from nominatim_db.tools.special_phrases.sp_csv_loader import SPCsvLoader
from nominatim_db.tools.special_phrases.special_phrase import SpecialPhrase
+
@pytest.fixture
def sp_csv_loader(src_dir):
"""
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for methods of the SPWikiLoader class.
"""
phrases = list(sp_wiki_loader.generate_phrases())
- assert set((p.p_label, p.p_class, p.p_type, p.p_operator) for p in phrases) ==\
- {('Zip Line', 'aerialway', 'zip_line', '-'),
- ('Zip Lines', 'aerialway', 'zip_line', '-'),
- ('Zip Line in', 'aerialway', 'zip_line', 'in'),
- ('Zip Lines in', 'aerialway', 'zip_line', 'in'),
- ('Zip Line near', 'aerialway', 'zip_line', 'near'),
- ('Animal shelter', 'amenity', 'animal_shelter', '-'),
- ('Animal shelters', 'amenity', 'animal_shelter', '-'),
- ('Animal shelter in', 'amenity', 'animal_shelter', 'in'),
- ('Animal shelters in', 'amenity', 'animal_shelter', 'in'),
- ('Animal shelter near', 'amenity', 'animal_shelter', 'near'),
- ('Animal shelters near', 'amenity', 'animal_shelter', 'near'),
- ('Drinking Water near', 'amenity', 'drinking_water', 'near'),
- ('Water', 'amenity', 'drinking_water', '-'),
- ('Water in', 'amenity', 'drinking_water', 'in'),
- ('Water near', 'amenity', 'drinking_water', 'near'),
- ('Embassy', 'amenity', 'embassy', '-'),
- ('Embassys', 'amenity', 'embassy', '-'),
- ('Embassies', 'amenity', 'embassy', '-')}
+ assert set((p.p_label, p.p_class, p.p_type, p.p_operator) for p in phrases) == \
+ {('Zip Line', 'aerialway', 'zip_line', '-'),
+ ('Zip Lines', 'aerialway', 'zip_line', '-'),
+ ('Zip Line in', 'aerialway', 'zip_line', 'in'),
+ ('Zip Lines in', 'aerialway', 'zip_line', 'in'),
+ ('Zip Line near', 'aerialway', 'zip_line', 'near'),
+ ('Animal shelter', 'amenity', 'animal_shelter', '-'),
+ ('Animal shelters', 'amenity', 'animal_shelter', '-'),
+ ('Animal shelter in', 'amenity', 'animal_shelter', 'in'),
+ ('Animal shelters in', 'amenity', 'animal_shelter', 'in'),
+ ('Animal shelter near', 'amenity', 'animal_shelter', 'near'),
+ ('Animal shelters near', 'amenity', 'animal_shelter', 'near'),
+ ('Drinking Water near', 'amenity', 'drinking_water', 'near'),
+ ('Water', 'amenity', 'drinking_water', '-'),
+ ('Water in', 'amenity', 'drinking_water', 'in'),
+ ('Water near', 'amenity', 'drinking_water', 'near'),
+ ('Embassy', 'amenity', 'embassy', '-'),
+ ('Embassys', 'amenity', 'embassy', '-'),
+ ('Embassies', 'amenity', 'embassy', '-')}
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Test for tiger data function
from textwrap import dedent
import pytest
-import pytest_asyncio
+import pytest_asyncio # noqa: F401
from nominatim_db.db.connection import execute_scalar
from nominatim_db.tools import tiger_data, freeze
from nominatim_db.errors import UsageError
+
class MockTigerTable:
def __init__(self, conn):
cur.execute("SELECT * FROM tiger LIMIT 1")
return cur.fetchone()
+
@pytest.fixture
def tiger_table(def_config, temp_db_conn, sql_preprocessor,
temp_db_with_extensions, tmp_path):
@pytest.mark.asyncio
async def test_add_tiger_data_database_frozen(def_config, temp_db_conn, tiger_table, tokenizer_mock,
- tmp_path):
+ tmp_path):
freeze.drop_update_tables(temp_db_conn)
with pytest.raises(UsageError) as excinfo:
@pytest.mark.asyncio
async def test_add_tiger_data_no_files(def_config, tiger_table, tokenizer_mock,
- tmp_path):
+ tmp_path):
await tiger_data.add_tiger_data(str(tmp_path), def_config, 1, tokenizer_mock())
assert tiger_table.count() == 0
@pytest.mark.asyncio
async def test_add_tiger_data_bad_file(def_config, tiger_table, tokenizer_mock,
- tmp_path):
+ tmp_path):
sqlfile = tmp_path / '1010.csv'
sqlfile.write_text("""Random text""")
@pytest.mark.asyncio
async def test_add_tiger_data_hnr_nan(def_config, tiger_table, tokenizer_mock,
- csv_factory, tmp_path):
+ csv_factory, tmp_path):
csv_factory('file1', hnr_from=99)
csv_factory('file2', hnr_from='L12')
csv_factory('file3', hnr_to='12.4')
@pytest.mark.parametrize("threads", (1, 5))
@pytest.mark.asyncio
async def test_add_tiger_data_tarfile(def_config, tiger_table, tokenizer_mock,
- tmp_path, src_dir, threads):
+ tmp_path, src_dir, threads):
tar = tarfile.open(str(tmp_path / 'sample.tar.gz'), "w:gz")
tar.add(str(src_dir / 'test' / 'testdb' / 'tiger' / '01001.csv'))
tar.close()
@pytest.mark.asyncio
async def test_add_tiger_data_bad_tarfile(def_config, tiger_table, tokenizer_mock,
- tmp_path):
+ tmp_path):
tarfile = tmp_path / 'sample.tar.gz'
tarfile.write_text("""Random text""")
@pytest.mark.asyncio
async def test_add_tiger_data_empty_tarfile(def_config, tiger_table, tokenizer_mock,
- tmp_path):
+ tmp_path):
tar = tarfile.open(str(tmp_path / 'sample.tar.gz'), "w:gz")
tar.add(__file__)
tar.close()
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for centroid computation.
from nominatim_db.utils.centroid import PointsCentroid
+
def test_empty_set():
c = PointsCentroid()
c.centroid()
-@pytest.mark.parametrize("centroid", [(0,0), (-1, 3), [0.0000032, 88.4938]])
+@pytest.mark.parametrize("centroid", [(0, 0), (-1, 3), [0.0000032, 88.4938]])
def test_one_point_centroid(centroid):
c = PointsCentroid()
#
# This file is part of Nominatim. (https://nominatim.org)
#
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2025 by the Nominatim developer community.
# For a full list of authors see the git log.
"""
Tests for the streaming JSON writer.
from nominatim_api.utils.json_writer import JsonWriter
+
@pytest.mark.parametrize("inval,outstr", [(None, 'null'),
(True, 'true'), (False, 'false'),
(23, '23'), (0, '0'), (-1.3, '-1.3'),
assert writer() == '{"something":5}'
json.loads(writer())
+
def test_object_many_values():
writer = JsonWriter()\
.start_object()\
assert writer() == '{"foo":null,"bar":{},"baz":"b\\taz"}'
json.loads(writer())
+
def test_object_many_values_without_none():
writer = JsonWriter()\
.start_object()\
.keyval_not_none('bar', None)\
.keyval_not_none('baz', '')\
.keyval_not_none('eve', False,
- transform = lambda v: 'yes' if v else 'no')\
+ transform=lambda v: 'yes' if v else 'no')\
.end_object()
assert writer() == '{"foo":0,"baz":"","eve":"no"}'
projects / nominatim.git / commitdiff