2020-03-24 15:50:58 +01:00
|
|
|
import os
|
|
|
|
import pytest
|
|
|
|
import mock
|
|
|
|
from requests import Response
|
|
|
|
import utils
|
|
|
|
|
2020-12-10 15:03:57 +01:00
|
|
|
from django.urls import reverse
|
2020-03-24 15:50:58 +01:00
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
|
|
|
|
from passerelle.base.models import ApiUser, AccessRight
|
|
|
|
from passerelle.contrib.greco.models import Greco
|
|
|
|
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
|
|
|
|
|
|
|
|
WSDL_FILENAME = os.path.join(os.path.dirname(__file__), 'data', 'greco.wsdl')
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.fixture
|
|
|
|
def conn():
|
|
|
|
api_user = ApiUser.objects.create(username='greco', keytype='API', key='grecokey')
|
|
|
|
connector = Greco.objects.create(
|
|
|
|
title='Greco',
|
|
|
|
slug='greco',
|
|
|
|
application='appid',
|
|
|
|
token_url='http://greco.example.net/token',
|
|
|
|
token_authorization='authtoken',
|
|
|
|
wsdl_url='file://%s' % WSDL_FILENAME,
|
|
|
|
verify_cert=False,
|
|
|
|
)
|
|
|
|
obj_type = ContentType.objects.get_for_model(Greco)
|
|
|
|
AccessRight.objects.create(
|
|
|
|
codename='can_access', apiuser=api_user, resource_type=obj_type, resource_pk=connector.pk
|
|
|
|
)
|
|
|
|
return connector
|
|
|
|
|
|
|
|
|
|
|
|
def post_effect_503(url, *args, **kwargs):
|
|
|
|
if url == 'http://greco.example.net/token':
|
|
|
|
return utils.FakedResponse(
|
|
|
|
content='{"token_type": "tt", "access_token": "at", "expires_in": "600"}', status_code=200
|
|
|
|
)
|
|
|
|
return utils.FakedResponse(content='Service Unavailable', status_code=503)
|
|
|
|
|
|
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.post', side_effect=post_effect_503)
|
|
|
|
def test_greco_ping(mocked_post, app, conn):
|
|
|
|
url = reverse('generic-endpoint', kwargs={'connector': 'greco', 'endpoint': 'ping', 'slug': conn.slug})
|
|
|
|
app.get(url, status=403)
|
|
|
|
assert mocked_post.call_count == 0
|
|
|
|
|
|
|
|
url += '?apikey=grecokey'
|
|
|
|
resp = app.get(url).json
|
|
|
|
assert mocked_post.call_count == 2 # token + communcationTest (ping -> 503)
|
|
|
|
assert resp['err'] == 'transport-error-503'
|