test/python/api/search/test_search_near.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2025 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Tests for running the near searcher.
   9 """
  10 import pytest
  11
  12 import nominatim_api as napi
  13 from nominatim_api.types import SearchDetails
  14 from nominatim_api.search.db_searches import NearSearch, PlaceSearch
  15 from nominatim_api.search.db_search_fields import WeightedStrings, WeightedCategories, \
  16                                                   FieldLookup
  17 from nominatim_api.search.db_search_lookups import LookupAll
  18
  19
  20 def run_search(apiobj, frontend, global_penalty, cat, cat_penalty=None, ccodes=[],
  21                details=SearchDetails()):
  22
  23     class PlaceSearchData:
  24         penalty = 0.0
  25         postcodes = WeightedStrings([], [])
  26         countries = WeightedStrings(ccodes, [0.0] * len(ccodes))
  27         housenumbers = WeightedStrings([], [])
  28         qualifiers = WeightedStrings([], [])
  29         lookups = [FieldLookup('name_vector', [56], LookupAll)]
  30         rankings = []
  31
  32     if ccodes is not None:
  33         details.countries = ccodes
  34
  35     place_search = PlaceSearch(0.0, PlaceSearchData(), 2)
  36
  37     if cat_penalty is None:
  38         cat_penalty = [0.0] * len(cat)
  39
  40     near_search = NearSearch(0.1, WeightedCategories(cat, cat_penalty), place_search)
  41
  42     api = frontend(apiobj, options=['search'])
  43
  44     async def run():
  45         async with api._async_api.begin() as conn:
  46             return await near_search.lookup(conn, details)
  47
  48     results = api._loop.run_until_complete(run())
  49     results.sort(key=lambda r: r.accuracy)
  50
  51     return results
  52
  53
  54 def test_no_results_inner_query(apiobj, frontend):
  55     assert not run_search(apiobj, frontend, 0.4, [('this', 'that')])
  56
  57
  58 def test_no_appropriate_results_inner_query(apiobj, frontend):
  59     apiobj.add_placex(place_id=100, country_code='us',
  60                       centroid=(5.6, 4.3),
  61                       geometry='POLYGON((0.0 0.0, 10.0 0.0, 10.0 2.0, 0.0 2.0, 0.0 0.0))')
  62     apiobj.add_search_name(100, names=[56], country_code='us',
  63                            centroid=(5.6, 4.3))
  64     apiobj.add_placex(place_id=22, class_='amenity', type='bank',
  65                       centroid=(5.6001, 4.2994))
  66
  67     assert not run_search(apiobj, frontend, 0.4, [('amenity', 'bank')])
  68
  69
  70 class TestNearSearch:
  71
  72     @pytest.fixture(autouse=True)
  73     def fill_database(self, apiobj):
  74         apiobj.add_placex(place_id=100, country_code='us',
  75                           centroid=(5.6, 4.3))
  76         apiobj.add_search_name(100, names=[56], country_code='us',
  77                                centroid=(5.6, 4.3))
  78         apiobj.add_placex(place_id=101, country_code='mx',
  79                           centroid=(-10.3, 56.9))
  80         apiobj.add_search_name(101, names=[56], country_code='mx',
  81                                centroid=(-10.3, 56.9))
  82
  83     def test_near_in_placex(self, apiobj, frontend):
  84         apiobj.add_placex(place_id=22, class_='amenity', type='bank',
  85                           centroid=(5.6001, 4.2994))
  86         apiobj.add_placex(place_id=23, class_='amenity', type='bench',
  87                           centroid=(5.6001, 4.2994))
  88
  89         results = run_search(apiobj, frontend, 0.1, [('amenity', 'bank')])
  90
  91         assert [r.place_id for r in results] == [22]
  92
  93     def test_multiple_types_near_in_placex(self, apiobj, frontend):
  94         apiobj.add_placex(place_id=22, class_='amenity', type='bank',
  95                           importance=0.002,
  96                           centroid=(5.6001, 4.2994))
  97         apiobj.add_placex(place_id=23, class_='amenity', type='bench',
  98                           importance=0.001,
  99                           centroid=(5.6001, 4.2994))
 100
 101         results = run_search(apiobj, frontend, 0.1, [('amenity', 'bank'),
 102                                                      ('amenity', 'bench')])
 103
 104         assert [r.place_id for r in results] == [22, 23]
 105
 106     def test_near_in_classtype(self, apiobj, frontend):
 107         apiobj.add_placex(place_id=22, class_='amenity', type='bank',
 108                           centroid=(5.6, 4.34))
 109         apiobj.add_placex(place_id=23, class_='amenity', type='bench',
 110                           centroid=(5.6, 4.34))
 111         apiobj.add_class_type_table('amenity', 'bank')
 112         apiobj.add_class_type_table('amenity', 'bench')
 113
 114         results = run_search(apiobj, frontend, 0.1, [('amenity', 'bank')])
 115
 116         assert [r.place_id for r in results] == [22]
 117
 118     @pytest.mark.parametrize('cc,rid', [('us', 22), ('mx', 23)])
 119     def test_restrict_by_country(self, apiobj, frontend, cc, rid):
 120         apiobj.add_placex(place_id=22, class_='amenity', type='bank',
 121                           centroid=(5.6001, 4.2994),
 122                           country_code='us')
 123         apiobj.add_placex(place_id=122, class_='amenity', type='bank',
 124                           centroid=(5.6001, 4.2994),
 125                           country_code='mx')
 126         apiobj.add_placex(place_id=23, class_='amenity', type='bank',
 127                           centroid=(-10.3001, 56.9),
 128                           country_code='mx')
 129         apiobj.add_placex(place_id=123, class_='amenity', type='bank',
 130                           centroid=(-10.3001, 56.9),
 131                           country_code='us')
 132
 133         results = run_search(apiobj, frontend, 0.1, [('amenity', 'bank')], ccodes=[cc, 'fr'])
 134
 135         assert [r.place_id for r in results] == [rid]
 136
 137     @pytest.mark.parametrize('excluded,rid', [(22, 122), (122, 22)])
 138     def test_exclude_place_by_id(self, apiobj, frontend, excluded, rid):
 139         apiobj.add_placex(place_id=22, class_='amenity', type='bank',
 140                           centroid=(5.6001, 4.2994),
 141                           country_code='us')
 142         apiobj.add_placex(place_id=122, class_='amenity', type='bank',
 143                           centroid=(5.6001, 4.2994),
 144                           country_code='us')
 145
 146         results = run_search(apiobj, frontend, 0.1, [('amenity', 'bank')],
 147                              details=SearchDetails(excluded=[excluded]))
 148
 149         assert [r.place_id for r in results] == [rid]
 150
 151     @pytest.mark.parametrize('layer,rids', [(napi.DataLayer.POI, [22]),
 152                                             (napi.DataLayer.MANMADE, [])])
 153     def test_with_layer(self, apiobj, frontend, layer, rids):
 154         apiobj.add_placex(place_id=22, class_='amenity', type='bank',
 155                           centroid=(5.6001, 4.2994),
 156                           country_code='us')
 157
 158         results = run_search(apiobj, frontend, 0.1, [('amenity', 'bank')],
 159                              details=SearchDetails(layers=layer))
 160
 161         assert [r.place_id for r in results] == rids