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