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