1140 lines
38 KiB
Python
1140 lines
38 KiB
Python
# Passerelle - uniform access to data and services
|
|
# Copyright (C) 2015 Entr'ouvert
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify it
|
|
# under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
import random
|
|
import warnings
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
import responses
|
|
from django.urls import reverse
|
|
|
|
import tests.utils
|
|
from passerelle.apps.api_particulier.models import APIParticulier
|
|
from passerelle.apps.arcgis.models import ArcGIS
|
|
from passerelle.apps.mdel.models import MDEL
|
|
from passerelle.base.models import BaseResource, LoggingParameters, ProxyLogger, ResourceLog, ResourceStatus
|
|
from passerelle.contrib.stub_invoices.models import StubInvoicesConnector
|
|
from passerelle.utils.api import endpoint
|
|
from passerelle.utils.jsonresponse import APIError
|
|
from tests.test_manager import login
|
|
|
|
|
|
@pytest.fixture
|
|
def mdel(db):
|
|
return tests.utils.setup_access_rights(MDEL.objects.create(slug='test'))
|
|
|
|
|
|
@pytest.fixture
|
|
def arcgis(db):
|
|
instance = tests.utils.setup_access_rights(ArcGIS.objects.create(slug='test'))
|
|
instance.set_log_level('DEBUG')
|
|
return instance
|
|
|
|
|
|
DEMAND_STATUS = {'closed': True, 'status': 'accepted', 'comment': 'dossier trait\xe9.'}
|
|
|
|
|
|
@mock.patch('passerelle.apps.mdel.models.Demand.get_status', lambda x: DEMAND_STATUS)
|
|
@mock.patch('passerelle.apps.mdel.models.Demand.create_zip', lambda x, y: '1-14-ILE-LA')
|
|
def test_generic_payload_logging(caplog, app, mdel):
|
|
filename = os.path.join(os.path.dirname(__file__), 'data', 'mdel', 'formdata.json')
|
|
with open(filename) as fd:
|
|
payload = json.load(fd)
|
|
resp = app.post_json('/mdel/test/create', params=payload, status=200)
|
|
assert resp.json['data']['demand_id'] == '1-14-ILE-LA'
|
|
|
|
resp = app.get('/mdel/test/status', params={'demand_id': '1-14-ILE-LA'})
|
|
data = resp.json['data']
|
|
|
|
assert data['closed'] is True
|
|
assert data['status'] == 'accepted'
|
|
assert data['comment'] == 'dossier trait\xe9.'
|
|
|
|
records = [record for record in caplog.records if record.name == 'passerelle.resource.mdel.test']
|
|
for record in records:
|
|
assert record.levelname == 'INFO'
|
|
assert record.connector == 'mdel'
|
|
if record.connector_endpoint_method == 'POST':
|
|
assert 'endpoint POST /mdel/test/create' in record.message
|
|
assert record.connector_endpoint == 'create'
|
|
else:
|
|
assert 'endpoint GET /mdel/test/status?demand_id=1-14-ILE-LA' in record.message
|
|
assert record.connector_endpoint == 'status'
|
|
assert record.connector_endpoint_url == '/mdel/test/status?demand_id=1-14-ILE-LA'
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.get')
|
|
def test_proxy_logger(mocked_get, caplog, app, arcgis):
|
|
with open(os.path.join(os.path.dirname(__file__), 'data', 'nancy_arcgis', 'sigresponse.json')) as fd:
|
|
payload = fd.read()
|
|
mocked_get.return_value = tests.utils.FakedResponse(content=payload, status_code=200)
|
|
|
|
# simple logger
|
|
arcgis.log_evel = 'DEBUG'
|
|
logger = ProxyLogger(connector=arcgis)
|
|
logger.debug('this is a debug test')
|
|
logger.info('this is an info test')
|
|
|
|
assert ResourceLog.objects.count() == 2
|
|
for log in ResourceLog.objects.all():
|
|
if log.levelno == 10:
|
|
assert log.message == 'this is a debug test'
|
|
else:
|
|
assert log.message == 'this is an info test'
|
|
|
|
log = ResourceLog.objects.filter(appname='arcgis', slug='test').delete()
|
|
caplog.clear()
|
|
resp = app.get(
|
|
'/arcgis/test/mapservice-query',
|
|
params={
|
|
'lon': 6.172122,
|
|
'lat': 48.673836,
|
|
'service': 'test',
|
|
'template': '{{ attributes.NOM }}',
|
|
'id_template': '{{ attributes.NUMERO }}',
|
|
},
|
|
headers={'Publik-Caller-URL': 'https://wcs.invalid/backoffice/management/foo/1/'},
|
|
status=200,
|
|
)
|
|
|
|
# Resource Custom DB Logger
|
|
log = ResourceLog.objects.filter(appname='arcgis', slug='test').first()
|
|
assert log.appname == 'arcgis'
|
|
assert log.slug == 'test'
|
|
assert log.levelno == 20
|
|
assert log.sourceip == '127.0.0.1'
|
|
assert log.extra['connector'] == 'arcgis'
|
|
assert log.extra['connector_endpoint'] == 'mapservice-query'
|
|
assert log.extra['connector_endpoint_method'] == 'GET'
|
|
assert log.extra['publik_caller_url'] == 'https://wcs.invalid/backoffice/management/foo/1/'
|
|
assert '/arcgis/test/mapservice-query?' in log.extra['connector_endpoint_url']
|
|
|
|
# Resource Generic Logger
|
|
record = caplog.records[0]
|
|
assert record.levelno == 20
|
|
assert record.levelname == 'INFO'
|
|
assert record.name == 'passerelle.resource.arcgis.test'
|
|
assert 'endpoint GET /arcgis/test/mapservice-query?' in record.message
|
|
assert not hasattr(record, 'connector_result')
|
|
record = caplog.records[1]
|
|
assert record.levelno == 10
|
|
assert record.levelname == 'DEBUG'
|
|
assert record.name == 'passerelle.resource.arcgis.test'
|
|
assert 'endpoint GET /arcgis/test/mapservice-query?' in record.message
|
|
assert hasattr(record, 'connector_result')
|
|
|
|
data = resp.json['data']
|
|
assert data[0]['id'] == '4'
|
|
assert data[0]['text'] == 'HAUSSONVILLE / BLANDAN / MON DESERT / SAURUPT'
|
|
|
|
# when changing log level
|
|
ResourceLog.objects.all().delete()
|
|
arcgis.set_log_level('INFO')
|
|
arcgis.save()
|
|
app.get(
|
|
'/arcgis/test/mapservice-query',
|
|
params={'lon': 6.172122, 'lat': 48.673836, 'service': 'test'},
|
|
status=200,
|
|
)
|
|
assert ResourceLog.objects.count() == 1
|
|
|
|
arcgis.logger.warning('first warning')
|
|
assert ResourceLog.objects.count() == 2
|
|
assert ResourceLog.objects.last().message == 'first warning'
|
|
assert ResourceLog.objects.last().levelno == 30
|
|
|
|
|
|
@mock.patch('requests.Session.send')
|
|
def test_proxy_logger_transaction_id(mocked_send, app, arcgis):
|
|
with open(os.path.join(os.path.dirname(__file__), 'data', 'nancy_arcgis', 'sigresponse.json')) as fd:
|
|
payload = fd.read()
|
|
mocked_send.return_value = tests.utils.FakedResponse(content=payload, status_code=200)
|
|
arcgis.log_evel = 'DEBUG'
|
|
arcgis.base_url = 'https://example.net/'
|
|
arcgis.save()
|
|
app.get(
|
|
'/arcgis/test/mapservice-query',
|
|
params={'lon': 6.172122, 'lat': 48.673836, 'service': 'test'},
|
|
status=200,
|
|
)
|
|
|
|
log1, log2, log3 = ResourceLog.objects.filter(appname='arcgis', slug='test').all()
|
|
assert log1.extra['transaction_id'] == log2.extra['transaction_id'] == log3.extra['transaction_id']
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.patch')
|
|
def test_proxy_logger_on_405(mocked_patch, caplog, app, arcgis):
|
|
mocked_patch.return_value = tests.utils.FakedResponse(status_code=500)
|
|
|
|
# simple logger
|
|
arcgis.log_evel = 'WARNING'
|
|
log = ResourceLog.objects.filter(appname='arcgis', slug='test').delete()
|
|
caplog.clear()
|
|
resp = app.patch('/arcgis/test/mapservice-query', status=405)
|
|
assert not resp.text
|
|
|
|
# Resource Custom DB Logger
|
|
log = ResourceLog.objects.filter(appname='arcgis', slug='test').first()
|
|
assert log.levelno == 30
|
|
assert log.message == 'endpoint PATCH /arcgis/test/mapservice-query (=> 405)'
|
|
assert log.extra['connector_endpoint_method'] == ['GET']
|
|
|
|
# Resource Generic Logger
|
|
for record in caplog.records:
|
|
if record.name != 'passerelle.resource.arcgis.test':
|
|
continue
|
|
assert record.levelno == 30
|
|
assert record.levelname == 'WARNING'
|
|
assert record.message == 'endpoint PATCH /arcgis/test/mapservice-query (=> 405)'
|
|
|
|
|
|
class FakeConnectorBase:
|
|
slug = 'connector'
|
|
|
|
def get_connector_slug(self):
|
|
return 'fake'
|
|
|
|
@endpoint(perm='OPEN')
|
|
def foo1(self, request):
|
|
pass
|
|
|
|
@endpoint(name='bar', perm='OPEN')
|
|
def foo2(self, request, param1):
|
|
pass
|
|
|
|
@endpoint(perm='OPEN')
|
|
def foo3(self, request, param1, param2):
|
|
pass
|
|
|
|
@endpoint(perm='OPEN')
|
|
def foo4(self, request, param1, param2='a', param3='b'):
|
|
pass
|
|
|
|
@endpoint(pattern='^test/$', example_pattern='test/', perm='OPEN')
|
|
def foo5(self, request, param1='a', param2='b', param3='c'):
|
|
pass
|
|
|
|
@endpoint(
|
|
pattern=r'^(?P<param1>\w+)/?$',
|
|
example_pattern='{param1}/',
|
|
parameters={'param1': {'description': 'param 1', 'example_value': 'bar'}},
|
|
perm='OPEN',
|
|
)
|
|
def foo6(self, request, param1, param2='a'):
|
|
pass
|
|
|
|
@endpoint(
|
|
description_get='foo7 get',
|
|
description_post='foo7 post',
|
|
description_put='foo7 put',
|
|
methods=['get', 'post', 'put'],
|
|
perm='OPEN',
|
|
)
|
|
def foo7(self, request, param1='a', param2='b', param3='c'):
|
|
pass
|
|
|
|
@endpoint(
|
|
long_description_get='foo7 get',
|
|
long_description_post='foo7 post',
|
|
long_description_put='foo7 put',
|
|
methods=['get', 'post', 'put'],
|
|
perm='OPEN',
|
|
)
|
|
def foo7b(self, request, param1='a', param2='b', param3='c'):
|
|
pass
|
|
|
|
@endpoint(
|
|
parameters={
|
|
'test': {'description': 'test', 'example_value': 'test'},
|
|
'reg': {'description': 'test', 'example_value': 'test'},
|
|
},
|
|
perm='OPEN',
|
|
)
|
|
def foo8(self, request, test, reg):
|
|
pass
|
|
|
|
@endpoint(
|
|
post={
|
|
'long_description': 'foo9 post',
|
|
},
|
|
perm='OPEN',
|
|
)
|
|
def foo9(self, request):
|
|
pass
|
|
|
|
@endpoint(cache_duration=10, perm='OPEN')
|
|
def cached_endpoint(self, request):
|
|
pass
|
|
|
|
|
|
def test_endpoint_decorator():
|
|
connector = FakeConnectorBase()
|
|
for i in range(8):
|
|
getattr(connector, 'foo%d' % (i + 1)).endpoint_info.object = connector
|
|
|
|
assert connector.foo1.endpoint_info.name == 'foo1'
|
|
assert connector.foo2.endpoint_info.name == 'bar'
|
|
assert not connector.foo1.endpoint_info.has_params()
|
|
assert connector.foo2.endpoint_info.has_params()
|
|
assert connector.foo2.endpoint_info.get_params() == [{'name': 'param1'}]
|
|
assert connector.foo3.endpoint_info.get_params() == [{'name': 'param1'}, {'name': 'param2'}]
|
|
assert connector.foo4.endpoint_info.get_params() == [
|
|
{'name': 'param1'},
|
|
{'name': 'param2', 'optional': True, 'default_value': 'a'},
|
|
{'name': 'param3', 'optional': True, 'default_value': 'b'},
|
|
]
|
|
assert connector.foo5.endpoint_info.get_params() == [
|
|
{'name': 'param1', 'optional': True, 'default_value': 'a'},
|
|
{'name': 'param2', 'optional': True, 'default_value': 'b'},
|
|
{'name': 'param3', 'optional': True, 'default_value': 'c'},
|
|
]
|
|
|
|
assert connector.foo6.endpoint_info.get_params() == [
|
|
{'name': 'param1', 'description': 'param 1'},
|
|
{'name': 'param2', 'optional': True, 'default_value': 'a'},
|
|
]
|
|
|
|
assert connector.foo1.endpoint_info.example_url() == '/fake/connector/foo1'
|
|
assert connector.foo1.endpoint_info.example_url_as_html() == '/fake/connector/foo1'
|
|
assert connector.foo2.endpoint_info.example_url() == '/fake/connector/bar'
|
|
assert connector.foo3.endpoint_info.example_url() == '/fake/connector/foo3'
|
|
assert connector.foo5.endpoint_info.example_url() == '/fake/connector/foo5/test/'
|
|
assert connector.foo5.endpoint_info.example_url_as_html() == '/fake/connector/foo5/test/'
|
|
assert connector.foo6.endpoint_info.example_url() == '/fake/connector/foo6/bar/'
|
|
assert (
|
|
connector.foo6.endpoint_info.example_url_as_html()
|
|
== '/fake/connector/foo6/<i class="varname">param1</i>/'
|
|
)
|
|
assert '®' not in connector.foo8.endpoint_info.example_url_as_html()
|
|
|
|
connector.foo6.endpoint_info.pattern = None
|
|
connector.foo6.endpoint_info.example_pattern = None
|
|
assert connector.foo6.endpoint_info.example_url() == '/fake/connector/foo6?param1=bar'
|
|
assert (
|
|
connector.foo6.endpoint_info.example_url_as_html()
|
|
== '/fake/connector/foo6?param1=<i class="varname">param1</i>'
|
|
)
|
|
|
|
connector.foo7.endpoint_info.http_method = 'get'
|
|
assert connector.foo7.endpoint_info.description == 'foo7 get'
|
|
connector.foo7.endpoint_info.http_method = 'post'
|
|
assert connector.foo7.endpoint_info.description == 'foo7 post'
|
|
assert connector.foo7.endpoint_info.cache_duration is None
|
|
connector.foo7.endpoint_info.http_method = 'put'
|
|
assert connector.foo7.endpoint_info.description == 'foo7 put'
|
|
|
|
connector.foo7b.endpoint_info.http_method = 'get'
|
|
assert connector.foo7b.endpoint_info.long_description == 'foo7 get'
|
|
connector.foo7b.endpoint_info.http_method = 'post'
|
|
assert connector.foo7b.endpoint_info.long_description == 'foo7 post'
|
|
assert connector.foo7b.endpoint_info.cache_duration is None
|
|
connector.foo7b.endpoint_info.http_method = 'put'
|
|
assert connector.foo7b.endpoint_info.long_description == 'foo7 put'
|
|
|
|
assert connector.cached_endpoint.endpoint_info.cache_duration == 10
|
|
|
|
connector.foo9.endpoint_info.http_method = 'post'
|
|
assert connector.foo9.endpoint_info.long_description == 'foo9 post'
|
|
|
|
|
|
class FakeJSONConnector:
|
|
slug = 'connector-json'
|
|
log_level = 'DEBUG'
|
|
|
|
FOO_SCHEMA = {
|
|
'properties': {
|
|
'foo': {
|
|
'type': 'array',
|
|
'items': {
|
|
'properties': {'id': {'type': 'integer'}, 'bar': {'type': 'boolean'}},
|
|
'required': ['id', 'bar'],
|
|
},
|
|
}
|
|
}
|
|
}
|
|
BAR_SCHEMA = copy.deepcopy(FOO_SCHEMA)
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.logger = ProxyLogger(connector=self)
|
|
|
|
@property
|
|
def logging_parameters(self):
|
|
return LoggingParameters()
|
|
|
|
def get_connector_slug(self):
|
|
return 'connector-json'
|
|
|
|
def down(self):
|
|
return False
|
|
|
|
def pre_process(self, post_data):
|
|
foo_objects = copy.deepcopy(post_data.get('foo', []))
|
|
foo_objects.reverse()
|
|
foo_len = len(foo_objects)
|
|
# pylint: disable=disallowed-name
|
|
for i, foo in enumerate(foo_objects):
|
|
if not foo.get('id'):
|
|
# id is empty, remove foo object
|
|
post_data['foo'].pop(foo_len - i - 1)
|
|
return post_data
|
|
|
|
BAR_SCHEMA['pre_process'] = pre_process
|
|
|
|
@endpoint(perm='OPEN', post={'request_body': {'schema': {'application/json': FOO_SCHEMA}}})
|
|
# pylint: disable=disallowed-name
|
|
def foo(self, request, post_data):
|
|
return {'data': post_data}
|
|
|
|
@endpoint(perm='OPEN', post={'request_body': {'schema': {'application/json': BAR_SCHEMA}}})
|
|
# pylint: disable=disallowed-name
|
|
def bar(self, request, post_data):
|
|
return {'data': post_data}
|
|
|
|
|
|
def test_endpoint_decorator_pre_process(db, app):
|
|
connector = FakeJSONConnector()
|
|
|
|
patch_init = mock.patch('passerelle.views.GenericConnectorMixin.init_stuff')
|
|
patch_object = mock.patch('passerelle.views.GenericEndpointView.get_object', return_value=connector)
|
|
|
|
url_foo = reverse(
|
|
'generic-endpoint',
|
|
kwargs={
|
|
'connector': 'connector-json',
|
|
'slug': 'connector-json',
|
|
'endpoint': 'foo',
|
|
},
|
|
)
|
|
url_bar = reverse(
|
|
'generic-endpoint',
|
|
kwargs={
|
|
'connector': 'connector-json',
|
|
'slug': 'connector-json',
|
|
'endpoint': 'bar',
|
|
},
|
|
)
|
|
|
|
payload = {'foo': [{'id': 42, 'bar': True}]}
|
|
with patch_init, patch_object:
|
|
resp = app.post_json(url_foo, params=payload)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] == payload
|
|
with patch_init, patch_object:
|
|
resp = app.post_json(url_bar, params=payload)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] == payload
|
|
|
|
payload = {'foo': [{'id': 42, 'bar': True}, {'id': None, 'bar': False}]} # invalid object
|
|
with patch_init, patch_object:
|
|
resp = app.post_json(url_foo, params=payload, status=400)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'foo/1/id: None is not of type %s' % repr('integer')
|
|
with patch_init, patch_object:
|
|
resp = app.post_json(url_bar, params=payload)
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] == {'foo': [{'id': 42, 'bar': True}]}
|
|
|
|
|
|
class FakeConnectorDatasource:
|
|
slug = 'connector-datasource'
|
|
log_level = 'DEBUG'
|
|
|
|
payload = {
|
|
'data': [
|
|
{'id': '1', 'text': 'A'},
|
|
{'id': '2', 'text': 'aa'},
|
|
{'id': '3', 'text': 'aAa'},
|
|
{'id': '4', 'text': 'AaAA'},
|
|
{'id': '5', 'text': 'b'},
|
|
{'id': '6', 'text': 'Bb'},
|
|
{'id': '7', 'text': 'bbb'},
|
|
{'id': '8', 'text': 'c'},
|
|
{'id': '9', 'text': 'cC'},
|
|
{'id': '10', 'text': 'Ccc'},
|
|
{'id': '11'},
|
|
{'foo': 'bar'},
|
|
]
|
|
}
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.logger = ProxyLogger(connector=self)
|
|
|
|
@property
|
|
def logging_parameters(self):
|
|
return LoggingParameters()
|
|
|
|
def get_connector_slug(self):
|
|
return 'connector-datasource'
|
|
|
|
def down(self):
|
|
return False
|
|
|
|
@endpoint(perm='OPEN')
|
|
def a(self, request):
|
|
return copy.deepcopy(self.payload)
|
|
|
|
@endpoint(datasource=True, perm='OPEN')
|
|
def b(self, request):
|
|
return copy.deepcopy(self.payload)
|
|
|
|
@endpoint(datasource=True, cache_duration=10, perm='OPEN')
|
|
def cached_b(self, request):
|
|
return copy.deepcopy(self.payload)
|
|
|
|
@endpoint(datasource=True, perm='OPEN')
|
|
def bb(self, request, id=None, q=None):
|
|
return copy.deepcopy(self.payload)
|
|
|
|
@endpoint(datasource=True, perm='OPEN')
|
|
def c(self, request):
|
|
return {}
|
|
|
|
@endpoint(datasource=True, perm='OPEN')
|
|
def d(self, request):
|
|
return {'data': 'foobar'}
|
|
|
|
@endpoint(datasource=True, perm='OPEN')
|
|
def e(self, request):
|
|
return {'data': ['foobar']}
|
|
|
|
|
|
def test_datasource_endpoint(db, app):
|
|
connector = FakeConnectorDatasource()
|
|
|
|
patch_init = mock.patch('passerelle.views.GenericConnectorMixin.init_stuff')
|
|
patch_object = mock.patch('passerelle.views.GenericEndpointView.get_object', return_value=connector)
|
|
|
|
url_a = reverse(
|
|
'generic-endpoint',
|
|
kwargs={
|
|
'connector': 'connector-datasource',
|
|
'slug': 'connector-datasource',
|
|
'endpoint': 'a',
|
|
},
|
|
)
|
|
url_b = reverse(
|
|
'generic-endpoint',
|
|
kwargs={
|
|
'connector': 'connector-datasource',
|
|
'slug': 'connector-datasource',
|
|
'endpoint': 'b',
|
|
},
|
|
)
|
|
url_cached_b = reverse(
|
|
'generic-endpoint',
|
|
kwargs={
|
|
'connector': 'connector-datasource',
|
|
'slug': 'connector-datasource',
|
|
'endpoint': 'cached_b',
|
|
},
|
|
)
|
|
url_bb = reverse(
|
|
'generic-endpoint',
|
|
kwargs={
|
|
'connector': 'connector-datasource',
|
|
'slug': 'connector-datasource',
|
|
'endpoint': 'bb',
|
|
},
|
|
)
|
|
url_c = reverse(
|
|
'generic-endpoint',
|
|
kwargs={
|
|
'connector': 'connector-datasource',
|
|
'slug': 'connector-datasource',
|
|
'endpoint': 'c',
|
|
},
|
|
)
|
|
url_d = reverse(
|
|
'generic-endpoint',
|
|
kwargs={
|
|
'connector': 'connector-datasource',
|
|
'slug': 'connector-datasource',
|
|
'endpoint': 'd',
|
|
},
|
|
)
|
|
url_e = reverse(
|
|
'generic-endpoint',
|
|
kwargs={
|
|
'connector': 'connector-datasource',
|
|
'slug': 'connector-datasource',
|
|
'endpoint': 'e',
|
|
},
|
|
)
|
|
|
|
with patch_init, patch_object:
|
|
resp = app.get(url_a)
|
|
assert resp.json['data'] == connector.payload['data']
|
|
|
|
resp = app.get(url_b)
|
|
assert resp.json['data'] == connector.payload['data']
|
|
|
|
app.get(url_a, params={'id': '1'}, status=400)
|
|
app.get(url_a, params={'q': 'a'}, status=400)
|
|
|
|
for url in [url_b, url_cached_b]:
|
|
resp = app.get(url, params={'id': '1'})
|
|
assert [d['id'] for d in resp.json['data']] == ['1']
|
|
resp = app.get(url, params={'id': '5'})
|
|
assert [d['id'] for d in resp.json['data']] == ['5']
|
|
resp = app.get(url, params={'q': 'a'})
|
|
assert [d['id'] for d in resp.json['data']] == ['1', '2', '3', '4']
|
|
resp = app.get(url, params={'q': 'AA'})
|
|
assert [d['id'] for d in resp.json['data']] == ['2', '3', '4']
|
|
resp = app.get(url, params={'q': 'bb'})
|
|
assert [d['id'] for d in resp.json['data']] == ['6', '7']
|
|
resp = app.get(url, params={'q': 'C'})
|
|
assert [d['id'] for d in resp.json['data']] == ['8', '9', '10']
|
|
|
|
# wrong result format
|
|
resp = app.get(url_c, params={'id': '1'})
|
|
assert resp.json == {'err': 0}
|
|
resp = app.get(url_d, params={'id': '1'})
|
|
assert resp.json == {'err': 0, 'data': 'foobar'}
|
|
resp = app.get(url_e, params={'id': '1'})
|
|
assert resp.json == {'err': 0, 'data': ['foobar']}
|
|
|
|
# this endpoints accepts id and q, no automatic filter
|
|
resp = app.get(url_bb, params={'id': '1'})
|
|
assert resp.json['data'] == connector.payload['data']
|
|
resp = app.get(url_bb, params={'q': 'a'})
|
|
assert resp.json['data'] == connector.payload['data']
|
|
|
|
|
|
def test_endpoint_description_in_template(app, db):
|
|
StubInvoicesConnector(slug='fake').save()
|
|
resp = app.get('/stub-invoices/fake/')
|
|
assert 'Get invoice details' in resp.text
|
|
assert 'not yet implemented' in resp.text
|
|
|
|
|
|
def test_endpoint_cache(app, db, monkeypatch):
|
|
@endpoint(cache_duration=10, perm='OPEN', methods=['get', 'post'], pattern=r'^(?P<url_param>\w+)/$')
|
|
def randominvoice(obj, request, url_param, get_param=None):
|
|
return {'data': random.randint(0, pow(10, 12))}
|
|
|
|
monkeypatch.setattr(StubInvoicesConnector, 'randominvoice', randominvoice, raising=False)
|
|
|
|
connector = StubInvoicesConnector(slug='fake')
|
|
connector.save()
|
|
|
|
class TestCache:
|
|
def __init__(self):
|
|
self.d = {}
|
|
self.get_calls = 0
|
|
self.set_calls = 0
|
|
|
|
def get(self, key):
|
|
self.get_calls += 1
|
|
return self.d.get(key)
|
|
|
|
def set(self, key, value, timeout):
|
|
self.set_calls += 1
|
|
self.d[key] = value
|
|
|
|
def delete(self, key):
|
|
del self.d[key]
|
|
|
|
cache = TestCache()
|
|
import passerelle.views
|
|
|
|
monkeypatch.setattr(passerelle.views, 'cache', cache)
|
|
|
|
resp1 = app.get('/stub-invoices/fake/randominvoice/url_param_value/?get_param=get_param_value')
|
|
assert cache.get_calls == 1
|
|
assert cache.set_calls == 1
|
|
|
|
resp2 = app.get('/stub-invoices/fake/randominvoice/url_param_value/?get_param=get_param_value')
|
|
assert cache.get_calls == 2
|
|
assert cache.set_calls == 1
|
|
assert resp1.json_body == resp2.json_body
|
|
|
|
resp3 = app.get(
|
|
'/stub-invoices/fake/randominvoice/url_param_value/?get_param=get_param_value'
|
|
'&apikey=somekey×tamp=somestamp'
|
|
)
|
|
assert cache.get_calls == 3
|
|
assert cache.set_calls == 1
|
|
assert resp1.json_body == resp3.json_body
|
|
|
|
resp4 = app.get('/stub-invoices/fake/randominvoice/url_param_value/?get_param=other_value')
|
|
assert cache.get_calls == 4
|
|
assert cache.set_calls == 2
|
|
assert resp1.json_body != resp4.json_body
|
|
|
|
resp5 = app.get('/stub-invoices/fake/randominvoice/other_value/?get_param=get_param_value')
|
|
assert cache.get_calls == 5
|
|
assert cache.set_calls == 3
|
|
assert resp1.json_body != resp5.json_body
|
|
|
|
resp6 = app.post('/stub-invoices/fake/randominvoice/other_value/?get_param=get_param_value')
|
|
assert cache.get_calls == 5
|
|
assert cache.set_calls == 3
|
|
assert resp1.json_body != resp6.json_body
|
|
|
|
|
|
def test_endpoint_cookies(app, db, monkeypatch):
|
|
@endpoint(methods=['get'], perm='OPEN')
|
|
def httpcall(obj, request):
|
|
with responses.RequestsMock() as rsps:
|
|
rsps.get('https://foo.invalid/set-cookie', json={}, headers={'set-cookie': 'foo=bar;'})
|
|
rsps.get(
|
|
'https://foo.invalid/get',
|
|
json={},
|
|
)
|
|
response = obj.requests.get('https://foo.invalid/set-cookie', allow_redirects=False)
|
|
cookie1 = response.request.headers.get('Cookie')
|
|
response = obj.requests.get('https://foo.invalid/get')
|
|
cookie2 = response.request.headers.get('Cookie')
|
|
return {'cookie1': cookie1, 'cookie2': cookie2}
|
|
|
|
monkeypatch.setattr(StubInvoicesConnector, 'httpcall', httpcall, raising=False)
|
|
|
|
connector = StubInvoicesConnector(slug='fake')
|
|
connector.save()
|
|
|
|
json_res = app.get('/stub-invoices/fake/httpcall').json
|
|
assert json_res['cookie1'] is None
|
|
assert json_res['cookie2'] == 'foo=bar'
|
|
# Do it a second time to test that no cookies are leaking from one call
|
|
# to the other
|
|
json_res = app.get('/stub-invoices/fake/httpcall').json
|
|
assert json_res['cookie1'] is None
|
|
assert json_res['cookie2'] == 'foo=bar'
|
|
|
|
|
|
def test_https_warnings(app, db, monkeypatch, httpsserver, relax_openssl):
|
|
from requests.exceptions import SSLError
|
|
|
|
resource = tests.utils.make_resource(
|
|
ArcGIS, base_url='https://example.com/', slug='gis', verify_cert=True
|
|
)
|
|
with pytest.raises(SSLError):
|
|
resource.requests.get(httpsserver.url)
|
|
resource.verify_cert = False
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter('error')
|
|
resource.requests.get(httpsserver.url)
|
|
|
|
|
|
def test_endpoint_typed_params(app, db, monkeypatch):
|
|
@endpoint(
|
|
methods=['get'],
|
|
parameters={
|
|
'boolean': {
|
|
'type': 'bool',
|
|
},
|
|
'integer': {
|
|
'type': 'int',
|
|
},
|
|
'floating': {
|
|
'type': 'float',
|
|
},
|
|
'date': {
|
|
'type': 'date',
|
|
},
|
|
},
|
|
perm='OPEN',
|
|
)
|
|
def httpcall(obj, request, boolean=False, integer=1, floating=1.1, date=None):
|
|
return {'boolean': boolean, 'integer': integer, 'floating': floating, 'date': date}
|
|
|
|
monkeypatch.setattr(StubInvoicesConnector, 'httpcall', httpcall, raising=False)
|
|
|
|
connector = StubInvoicesConnector(slug='fake')
|
|
connector.save()
|
|
|
|
json_res = app.get('/stub-invoices/fake/httpcall').json
|
|
assert json_res == {'boolean': False, 'integer': 1, 'floating': 1.1, 'date': None, 'err': 0}
|
|
|
|
json_res = app.get('/stub-invoices/fake/httpcall?boolean=True').json
|
|
assert json_res['boolean'] is True
|
|
json_res = app.get('/stub-invoices/fake/httpcall?boolean=on').json
|
|
assert json_res['boolean'] is True
|
|
json_res = app.get('/stub-invoices/fake/httpcall?boolean=False').json
|
|
assert json_res['boolean'] is False
|
|
json_res = app.get('/stub-invoices/fake/httpcall?boolean=off').json
|
|
assert json_res['boolean'] is False
|
|
json_res = app.get('/stub-invoices/fake/httpcall?boolean=notabool', status=400).json
|
|
assert json_res['err'] == 1
|
|
assert json_res['err_desc'] == 'invalid value for parameter "boolean"'
|
|
json_res = app.get('/stub-invoices/fake/httpcall?boolean=', status=400).json
|
|
assert json_res['err'] == 1
|
|
assert json_res['err_desc'] == 'invalid value for parameter "boolean"'
|
|
|
|
json_res = app.get('/stub-invoices/fake/httpcall?integer=2').json
|
|
assert json_res['integer'] == 2
|
|
json_res = app.get('/stub-invoices/fake/httpcall?integer=notanint', status=400).json
|
|
assert json_res['err'] == 1
|
|
assert json_res['err_desc'] == 'invalid value for parameter "integer"'
|
|
json_res = app.get('/stub-invoices/fake/httpcall?integer=', status=400).json
|
|
assert json_res['err'] == 1
|
|
assert json_res['err_desc'] == 'invalid value for parameter "integer"'
|
|
|
|
json_res = app.get('/stub-invoices/fake/httpcall?floating=1.5').json
|
|
assert json_res['floating'] == 1.5
|
|
json_res = app.get('/stub-invoices/fake/httpcall?floating=1,5').json
|
|
assert json_res['floating'] == 1.5
|
|
json_res = app.get('/stub-invoices/fake/httpcall?floating=notafloat', status=400).json
|
|
assert json_res['err'] == 1
|
|
assert json_res['err_desc'] == 'invalid value for parameter "floating"'
|
|
json_res = app.get('/stub-invoices/fake/httpcall?floating=', status=400).json
|
|
assert json_res['err'] == 1
|
|
assert json_res['err_desc'] == 'invalid value for parameter "floating"'
|
|
|
|
json_res = app.get('/stub-invoices/fake/httpcall?date=1970-01-01').json
|
|
assert json_res['date'] == '1970-01-01'
|
|
json_res = app.get('/stub-invoices/fake/httpcall?date=nodate', status=400).json
|
|
assert json_res['err'] == 1
|
|
assert json_res['err_desc'] == 'invalid value for parameter "date (YYYY-MM-DD expected)"'
|
|
json_res = app.get('/stub-invoices/fake/httpcall?date=1970-02-31', status=400).json
|
|
assert json_res['err'] == 1
|
|
assert json_res['err_desc'] == 'invalid value for parameter "date (not a valid date)"'
|
|
json_res = app.get('/stub-invoices/fake/httpcall?date=').json
|
|
assert not json_res['err']
|
|
assert json_res['date'] is None
|
|
|
|
|
|
def test_endpoint_params_type_detection(app, db, monkeypatch):
|
|
@endpoint(
|
|
methods=['get'],
|
|
parameters={
|
|
'bool_by_example': {
|
|
'example_value': True,
|
|
},
|
|
'int_by_example': {
|
|
'example_value': 1,
|
|
},
|
|
'float_by_example': {
|
|
'example_value': 1.1,
|
|
},
|
|
'date_by_example': {
|
|
'example_value': '1970-01-01',
|
|
},
|
|
},
|
|
perm='OPEN',
|
|
)
|
|
def httpcall(
|
|
obj,
|
|
request,
|
|
boolean=False,
|
|
integer=1,
|
|
floating=1.1,
|
|
bool_by_example=None,
|
|
int_by_example=None,
|
|
float_by_example=None,
|
|
date_by_example=None,
|
|
):
|
|
return {
|
|
'boolean': boolean,
|
|
'integer': integer,
|
|
'floating': floating,
|
|
'bool_by_example': bool_by_example,
|
|
'int_by_example': int_by_example,
|
|
'float_by_example': float_by_example,
|
|
'date_by_example': date_by_example,
|
|
}
|
|
|
|
monkeypatch.setattr(StubInvoicesConnector, 'httpcall', httpcall, raising=False)
|
|
|
|
connector = StubInvoicesConnector(slug='fake')
|
|
connector.save()
|
|
|
|
json_res = app.get('/stub-invoices/fake/httpcall?boolean=True').json
|
|
assert json_res['boolean'] is True
|
|
json_res = app.get('/stub-invoices/fake/httpcall?bool_by_example=True').json
|
|
assert json_res['bool_by_example'] is True
|
|
|
|
json_res = app.get('/stub-invoices/fake/httpcall?integer=2').json
|
|
assert json_res['integer'] == 2
|
|
json_res = app.get('/stub-invoices/fake/httpcall?int_by_example=2').json
|
|
assert json_res['int_by_example'] == 2
|
|
|
|
json_res = app.get('/stub-invoices/fake/httpcall?floating=1.5').json
|
|
assert json_res['floating'] == 1.5
|
|
json_res = app.get('/stub-invoices/fake/httpcall?float_by_example=1.5').json
|
|
assert json_res['float_by_example'] == 1.5
|
|
|
|
json_res = app.get('/stub-invoices/fake/httpcall?date_by_example=1970-01-01').json
|
|
assert json_res['date_by_example'] == '1970-01-01'
|
|
|
|
res = app.get('/stub-invoices/fake/')
|
|
for param in res.pyquery('ul.get-params li'):
|
|
param_details = param.getchildren()
|
|
name = next(el for el in param_details if 'param-name' in el.attrib['class']).text
|
|
typ = next(el for el in param_details if 'type' in el.attrib['class']).text
|
|
typ = typ.strip('()')
|
|
if 'bool' in name:
|
|
assert typ == 'boolean'
|
|
elif 'int' in name:
|
|
assert typ == 'integer'
|
|
elif 'float' in name:
|
|
assert typ == 'float'
|
|
elif 'date' in name:
|
|
assert typ == 'date'
|
|
else:
|
|
assert typ == 'string'
|
|
|
|
|
|
class DummyConnectorBase(BaseResource):
|
|
def get_availability_status(self):
|
|
# naive get_availability_status method for testing
|
|
try:
|
|
self.check_status()
|
|
except Exception:
|
|
return ResourceStatus(status='down')
|
|
return ResourceStatus(status='up')
|
|
|
|
class Meta:
|
|
app_label = 'dummy'
|
|
abstract = True
|
|
|
|
|
|
class DummyConnectorWithCheckStatus(DummyConnectorBase):
|
|
def check_status(self):
|
|
return
|
|
|
|
|
|
class DummyConnectorWithCheckStatusFailure(DummyConnectorBase):
|
|
def check_status(self):
|
|
raise Exception('dummy reason')
|
|
|
|
|
|
class DummyConnectorWithoutCheckStatus(DummyConnectorBase):
|
|
pass
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'connector_class, expected_status, expected_response',
|
|
[
|
|
(DummyConnectorWithCheckStatus, 200, {'err': 0}),
|
|
(
|
|
DummyConnectorWithCheckStatusFailure,
|
|
200,
|
|
{
|
|
'err_class': 'passerelle.utils.jsonresponse.APIError',
|
|
'err_desc': 'service not available',
|
|
'data': None,
|
|
'err': 1,
|
|
},
|
|
),
|
|
(DummyConnectorWithoutCheckStatus, 404, None),
|
|
],
|
|
)
|
|
def test_generic_up_endpoint(db, app, connector_class, expected_status, expected_response):
|
|
connector = connector_class()
|
|
connector.id = 42
|
|
|
|
url = reverse(
|
|
'generic-endpoint',
|
|
kwargs={
|
|
'connector': 'foo',
|
|
'slug': 'foo',
|
|
'endpoint': 'up',
|
|
},
|
|
)
|
|
patch_init = mock.patch('passerelle.views.GenericConnectorMixin.init_stuff')
|
|
patch_object = mock.patch('passerelle.views.GenericEndpointView.get_object', return_value=connector)
|
|
with patch_init, patch_object:
|
|
response = app.get(url, status=expected_status)
|
|
if expected_response is not None:
|
|
assert response.json == expected_response
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'connector_class, expected',
|
|
[
|
|
(DummyConnectorWithCheckStatus, True),
|
|
(DummyConnectorWithCheckStatusFailure, True),
|
|
(DummyConnectorWithoutCheckStatus, False),
|
|
],
|
|
)
|
|
def test_generic_up_in_endpoints_infos(db, app, connector_class, expected):
|
|
connector = connector_class()
|
|
connector.id = 42
|
|
|
|
up_endpoints = [ep for ep in connector.get_endpoints_infos() if ep.name == 'up']
|
|
if expected:
|
|
assert len(up_endpoints) == 1
|
|
else:
|
|
assert up_endpoints == []
|
|
|
|
|
|
def test_generic_endpoint_superuser_access(db, app, admin_user, simple_user):
|
|
MDEL.objects.create(slug='test')
|
|
filename = os.path.join(os.path.dirname(__file__), 'data', 'mdel', 'formdata.json')
|
|
with open(filename) as fd:
|
|
payload = json.load(fd)
|
|
|
|
app = login(app, username='user', password='user')
|
|
resp = app.post_json('/mdel/test/create', params=payload, status=403)
|
|
|
|
app = login(app, username='admin', password='admin')
|
|
resp = app.post_json('/mdel/test/create', params=payload, status=200)
|
|
assert resp.json['data']['demand_id'] == '1-14-ILE-LA'
|
|
|
|
|
|
class DummyConnectorWithoutOrdering(DummyConnectorBase):
|
|
@endpoint()
|
|
def b(self, request):
|
|
pass
|
|
|
|
@endpoint(name='c', pattern='bb')
|
|
def cb(self, request):
|
|
pass
|
|
|
|
@endpoint(name='c', pattern='aa')
|
|
def ca(self, request):
|
|
pass
|
|
|
|
@endpoint()
|
|
def a(self, request):
|
|
pass
|
|
|
|
|
|
class DummyConnectorWithOrdering(DummyConnectorBase):
|
|
@endpoint()
|
|
def b(self, request):
|
|
pass
|
|
|
|
@endpoint(name='c', pattern='bb', display_order=1)
|
|
def cb(self, request):
|
|
pass
|
|
|
|
@endpoint(name='c', pattern='aa', display_order=1)
|
|
def ca(self, request):
|
|
pass
|
|
|
|
@endpoint(display_order=42)
|
|
def a(self, request):
|
|
pass
|
|
|
|
|
|
class DummyConnectorWithOrderingAndCategory(DummyConnectorBase):
|
|
_category_ordering = ['Foo', 'Bar']
|
|
|
|
@endpoint(display_category='Foo')
|
|
def a(self, request):
|
|
pass
|
|
|
|
@endpoint(display_category='Bar', display_order=2)
|
|
def b(self, request):
|
|
pass
|
|
|
|
@endpoint()
|
|
def c(self, request):
|
|
pass
|
|
|
|
@endpoint(display_category='Bar', display_order=1)
|
|
def d(self, request):
|
|
pass
|
|
|
|
@endpoint(display_category='Blah')
|
|
def e(self, request):
|
|
pass
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'connector_class, expected_ordering',
|
|
[
|
|
(DummyConnectorWithoutOrdering, ['a', 'b', 'caa', 'cbb']),
|
|
(DummyConnectorWithOrdering, ['caa', 'cbb', 'a', 'b']),
|
|
(DummyConnectorWithOrderingAndCategory, ['a', 'd', 'b', 'e', 'c']),
|
|
],
|
|
)
|
|
def test_generic_up_in_endpoints_ordering(db, app, connector_class, expected_ordering):
|
|
connector = connector_class()
|
|
connector.id = 42
|
|
|
|
assert [
|
|
'%s%s' % (ep.name, ep.pattern or '') for ep in connector.get_endpoints_infos()
|
|
] == expected_ordering
|
|
|
|
|
|
def test_response_schema(db, app):
|
|
tests.utils.make_resource(APIParticulier, slug='test', platform='test', api_key='xxx')
|
|
response = app.get('/api-particulier/test/')
|
|
assert 'nombrePersonnesCharge' in response
|
|
|
|
|
|
def test_view_connector(db, app, monkeypatch, admin_user):
|
|
connector = DummyConnectorWithoutOrdering()
|
|
connector.id = 42
|
|
connector.slug = 'foo'
|
|
monkeypatch.setattr('passerelle.views.GenericConnectorView.model', DummyConnectorWithoutOrdering)
|
|
patch_init = mock.patch('passerelle.views.GenericConnectorView.init_stuff')
|
|
patch_object = mock.patch('passerelle.views.GenericConnectorView.get_object', return_value=connector)
|
|
|
|
# check description is hidden when unlogged
|
|
url = reverse('view-connector', kwargs={'connector': 'dummy', 'slug': 'foo'})
|
|
with patch_init, patch_object:
|
|
response = app.get(url)
|
|
assert len(response.pyquery('#description')) == 0
|
|
|
|
login(app)
|
|
|
|
with patch_init, patch_object:
|
|
response = app.get(url)
|
|
assert len(response.pyquery('#description')) == 1
|
|
|
|
|
|
class DummyErrorConnector(BaseResource):
|
|
class Meta:
|
|
app_label = 'dummy-error'
|
|
|
|
@endpoint(perm='OPEN')
|
|
def error(self, request):
|
|
raise APIError('error')
|
|
|
|
@endpoint(perm='OPEN')
|
|
def noerror(self, request):
|
|
return {'data': 1}
|
|
|
|
|
|
def test_dummy_error_connector(db, app):
|
|
connector = DummyErrorConnector()
|
|
connector.id = 42
|
|
|
|
url_error = reverse(
|
|
'generic-endpoint',
|
|
kwargs={
|
|
'connector': 'foo',
|
|
'slug': 'foo',
|
|
'endpoint': 'error',
|
|
},
|
|
)
|
|
url_noerror = reverse(
|
|
'generic-endpoint',
|
|
kwargs={
|
|
'connector': 'foo',
|
|
'slug': 'foo',
|
|
'endpoint': 'noerror',
|
|
},
|
|
)
|
|
patch_init = mock.patch('passerelle.views.GenericConnectorMixin.init_stuff')
|
|
patch_object = mock.patch('passerelle.views.GenericEndpointView.get_object', return_value=connector)
|
|
with patch_init, patch_object:
|
|
response = app.get(url_error)
|
|
assert response.headers['x-error-code'] == '1'
|
|
response = app.get(url_noerror)
|
|
assert 'x-error-code' not in response.headers
|