test/python/api/test_server_glue_v1.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 the Python web frameworks adaptor, v1 API.
   9 """
  10 import json
  11 import xml.etree.ElementTree as ET
  12
  13 import pytest
  14
  15 from fake_adaptor import FakeAdaptor, FakeError, FakeResponse
  16
  17 import nominatim_api.v1.server_glue as glue
  18 import nominatim_api as napi
  19 import nominatim_api.logging as loglib
  20
  21
  22 # ASGIAdaptor.get_int/bool()
  23
  24 @pytest.mark.parametrize('func', ['get_int', 'get_bool'])
  25 def test_adaptor_get_int_missing_but_required(func):
  26     with pytest.raises(FakeError, match='^400 -- .*missing'):
  27         getattr(FakeAdaptor(), func)('something')
  28
  29
  30 @pytest.mark.parametrize('func, val', [('get_int', 23), ('get_bool', True)])
  31 def test_adaptor_get_int_missing_with_default(func, val):
  32     assert getattr(FakeAdaptor(), func)('something', val) == val
  33
  34
  35 @pytest.mark.parametrize('inp', ['0', '234', '-4566953498567934876'])
  36 def test_adaptor_get_int_success(inp):
  37     assert FakeAdaptor(params={'foo': inp}).get_int('foo') == int(inp)
  38     assert FakeAdaptor(params={'foo': inp}).get_int('foo', 4) == int(inp)
  39
  40
  41 @pytest.mark.parametrize('inp', ['rs', '4.5', '6f'])
  42 def test_adaptor_get_int_bad_number(inp):
  43     with pytest.raises(FakeError, match='^400 -- .*must be a number'):
  44         FakeAdaptor(params={'foo': inp}).get_int('foo')
  45
  46
  47 @pytest.mark.parametrize('inp', ['1', 'true', 'whatever', 'false'])
  48 def test_adaptor_get_bool_trueish(inp):
  49     assert FakeAdaptor(params={'foo': inp}).get_bool('foo')
  50
  51
  52 def test_adaptor_get_bool_falsish():
  53     assert not FakeAdaptor(params={'foo': '0'}).get_bool('foo')
  54
  55
  56 # ASGIAdaptor.parse_format()
  57
  58 def test_adaptor_parse_format_use_default():
  59     adaptor = FakeAdaptor()
  60
  61     assert glue.parse_format(adaptor, napi.StatusResult, 'text') == 'text'
  62     assert adaptor.content_type == 'text/plain; charset=utf-8'
  63
  64
  65 def test_adaptor_parse_format_use_configured():
  66     adaptor = FakeAdaptor(params={'format': 'json'})
  67
  68     assert glue.parse_format(adaptor, napi.StatusResult, 'text') == 'json'
  69     assert adaptor.content_type == 'application/json; charset=utf-8'
  70
  71
  72 def test_adaptor_parse_format_invalid_value():
  73     adaptor = FakeAdaptor(params={'format': '@!#'})
  74
  75     with pytest.raises(FakeError, match='^400 -- .*must be one of'):
  76         glue.parse_format(adaptor, napi.StatusResult, 'text')
  77
  78
  79 # ASGIAdaptor.get_accepted_languages()
  80
  81 def test_accepted_languages_from_param():
  82     a = FakeAdaptor(params={'accept-language': 'de'})
  83     assert glue.get_accepted_languages(a) == 'de'
  84
  85
  86 def test_accepted_languages_from_header():
  87     a = FakeAdaptor(headers={'accept-language': 'de'})
  88     assert glue.get_accepted_languages(a) == 'de'
  89
  90
  91 def test_accepted_languages_from_default(monkeypatch):
  92     monkeypatch.setenv('NOMINATIM_DEFAULT_LANGUAGE', 'de')
  93     a = FakeAdaptor()
  94     assert glue.get_accepted_languages(a) == 'de'
  95
  96
  97 def test_accepted_languages_param_over_header():
  98     a = FakeAdaptor(params={'accept-language': 'de'},
  99                     headers={'accept-language': 'en'})
 100     assert glue.get_accepted_languages(a) == 'de'
 101
 102
 103 def test_accepted_languages_header_over_default(monkeypatch):
 104     monkeypatch.setenv('NOMINATIM_DEFAULT_LANGUAGE', 'en')
 105     a = FakeAdaptor(headers={'accept-language': 'de'})
 106     assert glue.get_accepted_languages(a) == 'de'
 107
 108
 109 # ASGIAdaptor.raise_error()
 110
 111 class TestAdaptorRaiseError:
 112
 113     @pytest.fixture(autouse=True)
 114     def init_adaptor(self):
 115         self.adaptor = FakeAdaptor()
 116         glue.setup_debugging(self.adaptor)
 117
 118     def run_raise_error(self, msg, status):
 119         with pytest.raises(FakeError) as excinfo:
 120             self.adaptor.raise_error(msg, status=status)
 121
 122         return excinfo.value
 123
 124     def test_without_content_set(self):
 125         err = self.run_raise_error('TEST', 404)
 126
 127         assert self.adaptor.content_type == 'text/plain; charset=utf-8'
 128         assert err.msg == 'ERROR 404: TEST'
 129         assert err.status == 404
 130
 131     def test_json(self):
 132         self.adaptor.content_type = 'application/json; charset=utf-8'
 133
 134         err = self.run_raise_error('TEST', 501)
 135
 136         content = json.loads(err.msg)['error']
 137         assert content['code'] == 501
 138         assert content['message'] == 'TEST'
 139
 140     def test_xml(self):
 141         self.adaptor.content_type = 'text/xml; charset=utf-8'
 142
 143         err = self.run_raise_error('this!', 503)
 144
 145         content = ET.fromstring(err.msg)
 146
 147         assert content.tag == 'error'
 148         assert content.find('code').text == '503'
 149         assert content.find('message').text == 'this!'
 150
 151
 152 def test_raise_error_during_debug():
 153     a = FakeAdaptor(params={'debug': '1'})
 154     glue.setup_debugging(a)
 155     loglib.log().section('Ongoing')
 156
 157     with pytest.raises(FakeError) as excinfo:
 158         a.raise_error('badstate')
 159
 160     content = ET.fromstring(excinfo.value.msg)
 161
 162     assert content.tag == 'html'
 163
 164     assert '>Ongoing<' in excinfo.value.msg
 165     assert 'badstate' in excinfo.value.msg
 166
 167
 168 # ASGIAdaptor.build_response
 169
 170 def test_build_response_without_content_type():
 171     resp = glue.build_response(FakeAdaptor(), 'attention')
 172
 173     assert isinstance(resp, FakeResponse)
 174     assert resp.status == 200
 175     assert resp.output == 'attention'
 176     assert resp.content_type == 'text/plain; charset=utf-8'
 177
 178
 179 def test_build_response_with_status():
 180     a = FakeAdaptor(params={'format': 'json'})
 181     glue.parse_format(a, napi.StatusResult, 'text')
 182
 183     resp = glue.build_response(a, 'stuff\nmore stuff', status=404)
 184
 185     assert isinstance(resp, FakeResponse)
 186     assert resp.status == 404
 187     assert resp.output == 'stuff\nmore stuff'
 188     assert resp.content_type == 'application/json; charset=utf-8'
 189
 190
 191 def test_build_response_jsonp_with_json():
 192     a = FakeAdaptor(params={'format': 'json', 'json_callback': 'test.func'})
 193     glue.parse_format(a, napi.StatusResult, 'text')
 194
 195     resp = glue.build_response(a, '{}')
 196
 197     assert isinstance(resp, FakeResponse)
 198     assert resp.status == 200
 199     assert resp.output == 'test.func({})'
 200     assert resp.content_type == 'application/javascript; charset=utf-8'
 201
 202
 203 def test_build_response_jsonp_without_json():
 204     a = FakeAdaptor(params={'format': 'text', 'json_callback': 'test.func'})
 205     glue.parse_format(a, napi.StatusResult, 'text')
 206
 207     resp = glue.build_response(a, '{}')
 208
 209     assert isinstance(resp, FakeResponse)
 210     assert resp.status == 200
 211     assert resp.output == '{}'
 212     assert resp.content_type == 'text/plain; charset=utf-8'
 213
 214
 215 @pytest.mark.parametrize('param', ['alert(); func', '\\n', '', 'a b'])
 216 def test_build_response_jsonp_bad_format(param):
 217     a = FakeAdaptor(params={'format': 'json', 'json_callback': param})
 218     glue.parse_format(a, napi.StatusResult, 'text')
 219
 220     with pytest.raises(FakeError, match='^400 -- .*Invalid'):
 221         glue.build_response(a, '{}')
 222
 223
 224 # get_layers()
 225
 226 def test_get_layers_no_param():
 227     assert glue.get_layers(FakeAdaptor()) is None
 228
 229
 230 @pytest.mark.parametrize('param,expected', [
 231     ('address', napi.DataLayer.ADDRESS),
 232     ('POI', napi.DataLayer.POI),
 233     ('address,poi', napi.DataLayer.ADDRESS | napi.DataLayer.POI),
 234     ('  address , poi ', napi.DataLayer.ADDRESS | napi.DataLayer.POI),
 235     ('address,address', napi.DataLayer.ADDRESS),
 236     ('address,', napi.DataLayer.ADDRESS)])
 237 def test_get_layers_success(param, expected):
 238     assert glue.get_layers(FakeAdaptor(params={'layer': param})) == expected
 239
 240
 241 @pytest.mark.parametrize('param', ['', ' ', ',', ' , '])
 242 def test_get_layers_empty_disables_filter(param):
 243     assert glue.get_layers(FakeAdaptor(params={'layer': param})) is None
 244
 245
 246 @pytest.mark.parametrize('param', ['bogus', 'address,bogus', 'name', '_name_',
 247                                    'address poi', '<script>'])
 248 def test_get_layers_invalid_value(param):
 249     with pytest.raises(FakeError, match='^400 -- .*must be a comma-separated list'):
 250         glue.get_layers(FakeAdaptor(params={'layer': param}))
 251
 252
 253 # status_endpoint()
 254
 255 class TestStatusEndpoint:
 256
 257     @pytest.fixture(autouse=True)
 258     def patch_status_func(self, monkeypatch):
 259         async def _status(*args, **kwargs):
 260             return self.status
 261
 262         monkeypatch.setattr(napi.NominatimAPIAsync, 'status', _status)
 263
 264     @pytest.mark.asyncio
 265     async def test_status_without_params(self):
 266         a = FakeAdaptor()
 267         self.status = napi.StatusResult(0, 'foo')
 268
 269         resp = await glue.status_endpoint(napi.NominatimAPIAsync(), a)
 270
 271         assert isinstance(resp, FakeResponse)
 272         assert resp.status == 200
 273         assert resp.content_type == 'text/plain; charset=utf-8'
 274
 275     @pytest.mark.asyncio
 276     async def test_status_with_error(self):
 277         a = FakeAdaptor()
 278         self.status = napi.StatusResult(405, 'foo')
 279
 280         resp = await glue.status_endpoint(napi.NominatimAPIAsync(), a)
 281
 282         assert isinstance(resp, FakeResponse)
 283         assert resp.status == 500
 284         assert resp.content_type == 'text/plain; charset=utf-8'
 285
 286     @pytest.mark.asyncio
 287     async def test_status_json_with_error(self):
 288         a = FakeAdaptor(params={'format': 'json'})
 289         self.status = napi.StatusResult(405, 'foo')
 290
 291         resp = await glue.status_endpoint(napi.NominatimAPIAsync(), a)
 292
 293         assert isinstance(resp, FakeResponse)
 294         assert resp.status == 200
 295         assert resp.content_type == 'application/json; charset=utf-8'
 296
 297     @pytest.mark.asyncio
 298     async def test_status_bad_format(self):
 299         a = FakeAdaptor(params={'format': 'foo'})
 300         self.status = napi.StatusResult(0, 'foo')
 301
 302         with pytest.raises(FakeError):
 303             await glue.status_endpoint(napi.NominatimAPIAsync(), a)
 304
 305
 306 # details_endpoint()
 307
 308 class TestDetailsEndpoint:
 309
 310     @pytest.fixture(autouse=True)
 311     def patch_lookup_func(self, monkeypatch):
 312         self.result = napi.DetailedResult(napi.SourceTable.PLACEX,
 313                                           ('place', 'thing'),
 314                                           napi.Point(1.0, 2.0))
 315         self.lookup_args = []
 316
 317         async def _lookup(*args, **kwargs):
 318             self.lookup_args.extend(args[1:])
 319             return self.result
 320
 321         monkeypatch.setattr(napi.NominatimAPIAsync, 'details', _lookup)
 322
 323     @pytest.mark.asyncio
 324     async def test_details_no_params(self):
 325         a = FakeAdaptor()
 326
 327         with pytest.raises(FakeError, match='^400 -- .*Missing'):
 328             await glue.details_endpoint(napi.NominatimAPIAsync(), a)
 329
 330     @pytest.mark.asyncio
 331     async def test_details_by_place_id(self):
 332         a = FakeAdaptor(params={'place_id': '4573'})
 333
 334         await glue.details_endpoint(napi.NominatimAPIAsync(), a)
 335
 336         assert self.lookup_args[0].place_id == 4573
 337
 338     @pytest.mark.asyncio
 339     async def test_details_by_osm_id(self):
 340         a = FakeAdaptor(params={'osmtype': 'N', 'osmid': '45'})
 341
 342         await glue.details_endpoint(napi.NominatimAPIAsync(), a)
 343
 344         assert self.lookup_args[0].osm_type == 'N'
 345         assert self.lookup_args[0].osm_id == 45
 346         assert self.lookup_args[0].osm_class is None
 347
 348     @pytest.mark.asyncio
 349     async def test_details_by_postcode(self):
 350         a = FakeAdaptor(params={'postcode': 'us:94110'})
 351
 352         await glue.details_endpoint(napi.NominatimAPIAsync(), a)
 353
 354         assert self.lookup_args[0].country_code == 'us'
 355         assert self.lookup_args[0].postcode == '94110'
 356
 357     @pytest.mark.asyncio
 358     async def test_details_by_postcode_id(self):
 359         a = FakeAdaptor(params={'postcode': 'Pus:94110'})
 360
 361         await glue.details_endpoint(napi.NominatimAPIAsync(), a)
 362
 363         assert self.lookup_args[0].country_code == 'us'
 364         assert self.lookup_args[0].postcode == '94110'
 365
 366     @pytest.mark.asyncio
 367     async def test_details_with_debugging(self):
 368         a = FakeAdaptor(params={'osmtype': 'N', 'osmid': '45', 'debug': '1'})
 369
 370         resp = await glue.details_endpoint(napi.NominatimAPIAsync(), a)
 371         content = ET.fromstring(resp.output)
 372
 373         assert resp.content_type == 'text/html; charset=utf-8'
 374         assert content.tag == 'html'
 375
 376     @pytest.mark.asyncio
 377     async def test_details_no_result(self):
 378         a = FakeAdaptor(params={'place_id': '4573'})
 379         self.result = None
 380
 381         with pytest.raises(FakeError, match='^404 -- .*found'):
 382             await glue.details_endpoint(napi.NominatimAPIAsync(), a)
 383
 384
 385 # reverse_endpoint()
 386 class TestReverseEndPoint:
 387
 388     @pytest.fixture(autouse=True)
 389     def patch_reverse_func(self, monkeypatch):
 390         self.result = napi.ReverseResult(napi.SourceTable.PLACEX,
 391                                          ('place', 'thing'),
 392                                          napi.Point(1.0, 2.0))
 393
 394         async def _reverse(*args, **kwargs):
 395             return self.result
 396
 397         monkeypatch.setattr(napi.NominatimAPIAsync, 'reverse', _reverse)
 398
 399     @pytest.mark.asyncio
 400     @pytest.mark.parametrize('params', [{}, {'lat': '3.4'}, {'lon': '6.7'}])
 401     async def test_reverse_no_params(self, params):
 402         a = FakeAdaptor()
 403         a.params = params
 404         a.params['format'] = 'xml'
 405
 406         with pytest.raises(FakeError, match='^400 -- (?s:.*)missing'):
 407             await glue.reverse_endpoint(napi.NominatimAPIAsync(), a)
 408
 409     @pytest.mark.asyncio
 410     async def test_reverse_success(self):
 411         a = FakeAdaptor()
 412         a.params['lat'] = '56.3'
 413         a.params['lon'] = '6.8'
 414
 415         assert await glue.reverse_endpoint(napi.NominatimAPIAsync(), a)
 416
 417     @pytest.mark.asyncio
 418     async def test_reverse_from_search(self):
 419         a = FakeAdaptor()
 420         a.params['q'] = '34.6 2.56'
 421         a.params['format'] = 'json'
 422
 423         res = await glue.search_endpoint(napi.NominatimAPIAsync(), a)
 424
 425         assert len(json.loads(res.output)) == 1
 426
 427
 428 # lookup_endpoint()
 429
 430 class TestLookupEndpoint:
 431
 432     @pytest.fixture(autouse=True)
 433     def patch_lookup_func(self, monkeypatch):
 434         self.results = [napi.SearchResult(napi.SourceTable.PLACEX,
 435                                           ('place', 'thing'),
 436                                           napi.Point(1.0, 2.0))]
 437
 438         async def _lookup(*args, **kwargs):
 439             return napi.SearchResults(self.results)
 440
 441         monkeypatch.setattr(napi.NominatimAPIAsync, 'lookup', _lookup)
 442
 443     @pytest.mark.asyncio
 444     async def test_lookup_no_params(self):
 445         a = FakeAdaptor()
 446         a.params['format'] = 'json'
 447
 448         res = await glue.lookup_endpoint(napi.NominatimAPIAsync(), a)
 449
 450         assert res.output == '[]'
 451
 452     @pytest.mark.asyncio
 453     @pytest.mark.parametrize('param', ['w', 'bad', ''])
 454     async def test_lookup_bad_params(self, param):
 455         a = FakeAdaptor()
 456         a.params['format'] = 'json'
 457         a.params['osm_ids'] = f'W34,{param},N33333'
 458
 459         res = await glue.lookup_endpoint(napi.NominatimAPIAsync(), a)
 460
 461         assert len(json.loads(res.output)) == 1
 462
 463     @pytest.mark.asyncio
 464     @pytest.mark.parametrize('param', ['p234234', '4563'])
 465     async def test_lookup_bad_osm_type(self, param):
 466         a = FakeAdaptor()
 467         a.params['format'] = 'json'
 468         a.params['osm_ids'] = f'W34,{param},N33333'
 469
 470         res = await glue.lookup_endpoint(napi.NominatimAPIAsync(), a)
 471
 472         assert len(json.loads(res.output)) == 1
 473
 474     @pytest.mark.asyncio
 475     async def test_lookup_working(self):
 476         a = FakeAdaptor()
 477         a.params['format'] = 'json'
 478         a.params['osm_ids'] = 'N23,W34'
 479
 480         res = await glue.lookup_endpoint(napi.NominatimAPIAsync(), a)
 481
 482         assert len(json.loads(res.output)) == 1
 483
 484
 485 # search_endpoint()
 486
 487 class TestSearchEndPointSearch:
 488
 489     @pytest.fixture(autouse=True)
 490     def patch_lookup_func(self, monkeypatch):
 491         self.results = [napi.SearchResult(napi.SourceTable.PLACEX,
 492                                           ('place', 'thing'),
 493                                           napi.Point(1.0, 2.0))]
 494
 495         async def _search(*args, **kwargs):
 496             return napi.SearchResults(self.results)
 497
 498         monkeypatch.setattr(napi.NominatimAPIAsync, 'search', _search)
 499
 500     @pytest.mark.asyncio
 501     async def test_search_free_text(self):
 502         a = FakeAdaptor()
 503         a.params['q'] = 'something'
 504
 505         res = await glue.search_endpoint(napi.NominatimAPIAsync(), a)
 506
 507         assert len(json.loads(res.output)) == 1
 508
 509     @pytest.mark.asyncio
 510     async def test_search_free_text_xml(self):
 511         a = FakeAdaptor()
 512         a.params['q'] = 'something'
 513         a.params['format'] = 'xml'
 514
 515         res = await glue.search_endpoint(napi.NominatimAPIAsync(), a)
 516
 517         assert res.status == 200
 518         assert res.output.index('something') > 0
 519
 520     @pytest.mark.asyncio
 521     async def test_search_free_text_xml_uses_stable_postcode_exclude_ids(self):
 522         self.results = [napi.SearchResult(napi.SourceTable.POSTCODE,
 523                                           ('place', 'postcode'),
 524                                           napi.Point(1.0, 2.0),
 525                                           place_id=123,
 526                                           names={'ref': 'EH4 7EA'},
 527                                           country_code='gb')]
 528         a = FakeAdaptor(params={'q': 'something', 'format': 'xml'})
 529
 530         res = await glue.search_endpoint(napi.NominatimAPIAsync(), a)
 531
 532         assert 'exclude_place_ids="Pgb:EH4_7EA"' in res.output
 533
 534     @pytest.mark.asyncio
 535     async def test_search_free_text_jsonv2_emits_postcode_id(self):
 536         self.results = [napi.SearchResult(napi.SourceTable.POSTCODE,
 537                                           ('place', 'postcode'),
 538                                           napi.Point(1.0, 2.0),
 539                                           place_id=123,
 540                                           names={'ref': 'EH4 7EA'},
 541                                           country_code='gb')]
 542         a = FakeAdaptor(params={'q': 'something'})
 543
 544         res = await glue.search_endpoint(napi.NominatimAPIAsync(), a)
 545
 546         assert '"postcode_id":"Pgb:EH4_7EA"' in res.output
 547
 548     @pytest.mark.asyncio
 549     async def test_search_free_and_structured(self):
 550         a = FakeAdaptor()
 551         a.params['q'] = 'something'
 552         a.params['city'] = 'ignored'
 553
 554         with pytest.raises(FakeError, match='^400 -- .*cannot be used together'):
 555             await glue.search_endpoint(napi.NominatimAPIAsync(), a)
 556
 557     @pytest.mark.asyncio
 558     @pytest.mark.parametrize('dedupe,numres', [(True, 1), (False, 2)])
 559     async def test_search_dedupe(self, dedupe, numres):
 560         self.results = self.results * 2
 561         a = FakeAdaptor()
 562         a.params['q'] = 'something'
 563         if not dedupe:
 564             a.params['dedupe'] = '0'
 565
 566         res = await glue.search_endpoint(napi.NominatimAPIAsync(), a)
 567
 568         assert len(json.loads(res.output)) == numres
 569
 570
 571 class TestSearchEndPointSearchAddress:
 572
 573     @pytest.fixture(autouse=True)
 574     def patch_lookup_func(self, monkeypatch):
 575         self.results = [napi.SearchResult(napi.SourceTable.PLACEX,
 576                                           ('place', 'thing'),
 577                                           napi.Point(1.0, 2.0))]
 578
 579         async def _search(*args, **kwargs):
 580             return napi.SearchResults(self.results)
 581
 582         monkeypatch.setattr(napi.NominatimAPIAsync, 'search_address', _search)
 583
 584     @pytest.mark.asyncio
 585     async def test_search_structured(self):
 586         a = FakeAdaptor()
 587         a.params['street'] = 'something'
 588
 589         res = await glue.search_endpoint(napi.NominatimAPIAsync(), a)
 590
 591         assert len(json.loads(res.output)) == 1
 592
 593
 594 class TestSearchEndPointSearchCategory:
 595
 596     @pytest.fixture(autouse=True)
 597     def patch_lookup_func(self, monkeypatch):
 598         self.results = [napi.SearchResult(napi.SourceTable.PLACEX,
 599                                           ('place', 'thing'),
 600                                           napi.Point(1.0, 2.0))]
 601
 602         async def _search(*args, **kwargs):
 603             return napi.SearchResults(self.results)
 604
 605         monkeypatch.setattr(napi.NominatimAPIAsync, 'search_category', _search)
 606
 607     @pytest.mark.asyncio
 608     async def test_search_category(self):
 609         a = FakeAdaptor()
 610         a.params['q'] = '[shop=fog]'
 611
 612         res = await glue.search_endpoint(napi.NominatimAPIAsync(), a)
 613
 614         assert len(json.loads(res.output)) == 1