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={}, headers={}, config=None):
  36         self.params = params
  37         self.headers = headers
  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={'http_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={'http_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={'http_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('bad state')
 201
 202     content = ET.fromstring(excinfo.value.msg)
 203
 204     assert content.tag == 'html'
 205
 206     assert '>Ongoing<' in excinfo.value.msg
 207     assert 'bad state' 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)