704 lines
25 KiB
Python
704 lines
25 KiB
Python
# -*- coding: utf-8 -*-
|
|
# 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/>.
|
|
|
|
from __future__ import unicode_literals
|
|
|
|
import copy
|
|
import os
|
|
import json
|
|
import random
|
|
import warnings
|
|
|
|
import mock
|
|
import pytest
|
|
|
|
import utils
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.core.urlresolvers import reverse
|
|
|
|
from passerelle.apps.arcgis.models import ArcGIS
|
|
from passerelle.base.models import ResourceLog, ProxyLogger, BaseResource, HTTPResource, LoggingParameters
|
|
from passerelle.base.models import ResourceStatus
|
|
from passerelle.apps.mdel.models import MDEL
|
|
from passerelle.contrib.stub_invoices.models import StubInvoicesConnector
|
|
from passerelle.utils.api import endpoint
|
|
|
|
from test_manager import login, admin_user, simple_user
|
|
|
|
|
|
@pytest.fixture
|
|
def mdel(db):
|
|
return utils.setup_access_rights(MDEL.objects.create(slug='test'))
|
|
|
|
|
|
@pytest.fixture
|
|
def arcgis(db):
|
|
instance = 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')
|
|
payload = json.load(open(filename))
|
|
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):
|
|
payload = open(os.path.join(os.path.dirname(__file__), 'data', 'nancy_arcgis', 'sigresponse.json')).read()
|
|
mocked_get.return_value = 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/district', params={'lon': 6.172122, 'lat': 48.673836}, 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'] == 'district'
|
|
assert log.extra['connector_endpoint_method'] == 'GET'
|
|
assert '/arcgis/test/district?' in log.extra['connector_endpoint_url']
|
|
|
|
# Resource Generic Logger
|
|
for record in caplog.records:
|
|
if record.name != 'passerelle.resource.arcgis.test':
|
|
continue
|
|
assert record.levelno == 20
|
|
assert record.levelname == 'INFO'
|
|
assert record.name == 'passerelle.resource.arcgis.test'
|
|
assert u"endpoint GET /arcgis/test/district?" in record.message
|
|
|
|
data = resp.json['data']
|
|
assert data['id'] == 4
|
|
assert data['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/district', params={'lon': 6.172122, 'lat': 48.673836}, 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
|
|
|
|
|
|
class FakeConnectorBase(object):
|
|
slug = 'connector'
|
|
|
|
def get_connector_slug(self):
|
|
return 'fake'
|
|
|
|
@endpoint()
|
|
def foo1(self, request):
|
|
pass
|
|
|
|
@endpoint(name='bar')
|
|
def foo2(self, request, param1):
|
|
pass
|
|
|
|
@endpoint()
|
|
def foo3(self, request, param1, param2):
|
|
pass
|
|
|
|
@endpoint()
|
|
def foo4(self, request, param1, param2='a', param3='b'):
|
|
pass
|
|
|
|
@endpoint(pattern='^test/$', example_pattern='test/')
|
|
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'
|
|
}})
|
|
def foo6(self, request, param1, param2='a'):
|
|
pass
|
|
|
|
@endpoint(description_get='foo7 get', description_post='foo7 post',
|
|
methods=['get', 'post'])
|
|
def foo7(self, request, param1='a', param2='b', param3='c'):
|
|
pass
|
|
|
|
@endpoint(long_description_get='foo7 get', long_description_post='foo7 post',
|
|
methods=['get', 'post'])
|
|
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'},
|
|
})
|
|
def foo8(self, request, test, reg):
|
|
pass
|
|
|
|
@endpoint(post={
|
|
'long_description': 'foo9 post',
|
|
})
|
|
def foo9(self, request):
|
|
pass
|
|
|
|
|
|
@endpoint(cache_duration=10)
|
|
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.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
|
|
|
|
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(object):
|
|
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(FakeJSONConnector, self).__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)
|
|
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(
|
|
post={
|
|
'request_body': {
|
|
'schema': {
|
|
'application/json': FOO_SCHEMA
|
|
}
|
|
}
|
|
})
|
|
def foo(self, request, post_data):
|
|
return {'data': post_data}
|
|
|
|
@endpoint(
|
|
post={
|
|
'request_body': {
|
|
'schema': {
|
|
'application/json': BAR_SCHEMA
|
|
}
|
|
}
|
|
})
|
|
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(u'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}
|
|
]
|
|
}
|
|
|
|
|
|
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, 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(object):
|
|
|
|
def __init__(self):
|
|
self.d = dict()
|
|
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, httpbin):
|
|
|
|
@endpoint(methods=['get'])
|
|
def httpcall(obj, request):
|
|
response = obj.requests.get(httpbin.url + '/cookies/set?foo=bar', allow_redirects=False)
|
|
cookie1 = response.request.headers.get('Cookie')
|
|
response = obj.requests.get(httpbin.url + '/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, httpbin_secure, relax_openssl):
|
|
from requests.exceptions import SSLError
|
|
from passerelle.contrib.maarch.models import Maarch
|
|
|
|
resource = utils.make_resource(Maarch, wsdl_url='https://example.com/', slug='slug', verify_cert=True)
|
|
with pytest.raises(SSLError):
|
|
resource.requests.get(httpbin_secure.join('/get/'))
|
|
resource.verify_cert = False
|
|
with warnings.catch_warnings():
|
|
warnings.simplefilter('error')
|
|
resource.requests.get(httpbin_secure.join('/get/'))
|
|
|
|
|
|
def test_endpoint_typed_params(app, db, monkeypatch):
|
|
|
|
@endpoint(methods=['get'],
|
|
parameters={
|
|
'boolean': {
|
|
'type': 'bool',
|
|
},
|
|
'integer': {
|
|
'type': 'int',
|
|
},
|
|
'floating': {
|
|
'type': 'float',
|
|
},
|
|
})
|
|
def httpcall(obj, request, boolean=False, integer=1, floating=1.1):
|
|
return {'boolean': boolean, 'integer': integer, 'floating': floating}
|
|
|
|
monkeypatch.setattr(StubInvoicesConnector, 'httpcall', httpcall, raising=False)
|
|
|
|
connector = StubInvoicesConnector(slug='fake')
|
|
connector.save()
|
|
|
|
json_res = app.get('/stub-invoices/fake/httpcall').json
|
|
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"'
|
|
|
|
|
|
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,
|
|
},
|
|
})
|
|
def httpcall(obj, request, boolean=False, integer=1, floating=1.1,
|
|
bool_by_example=None, int_by_example=None, float_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}
|
|
|
|
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
|
|
|
|
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'
|
|
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):
|
|
connector = MDEL.objects.create(slug='test')
|
|
filename = os.path.join(os.path.dirname(__file__), 'data', 'mdel', 'formdata.json')
|
|
payload = json.load(open(filename))
|
|
|
|
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'
|