1258 lines
42 KiB
Python
1258 lines
42 KiB
Python
import base64
|
|
import json
|
|
import urllib
|
|
import uuid
|
|
|
|
import pytest
|
|
import requests
|
|
import responses
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from requests.exceptions import ConnectTimeout, ReadTimeout
|
|
|
|
import tests.utils
|
|
from passerelle.apps.carl.models import Carl
|
|
from passerelle.base.models import AccessRight, ApiUser
|
|
from passerelle.utils.jsonresponse import APIError
|
|
|
|
APIERROR_CLS = 'passerelle.utils.jsonresponse.APIError'
|
|
SERVICE_URL = 'http://carl.example.org/'
|
|
|
|
|
|
@pytest.fixture()
|
|
def carl_token_conn(db):
|
|
'''Usefull to test token authentication
|
|
For other tests it will be anoying to mock the 1st call to fetch the token
|
|
'''
|
|
user = ApiUser.objects.create(username='all', keytype='', key='')
|
|
carl = Carl.objects.create(
|
|
service_url=SERVICE_URL, slug='test', carl_username='foo', carl_password='hackmeplz'
|
|
)
|
|
content_type = ContentType.objects.get_for_model(carl)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=user, resource_type=content_type, resource_pk=carl.pk
|
|
)
|
|
return carl
|
|
|
|
|
|
@pytest.fixture()
|
|
def carl_conn(db):
|
|
user = ApiUser.objects.create(username='all', keytype='', key='')
|
|
carl = Carl.objects.create(
|
|
service_url=SERVICE_URL,
|
|
slug='test',
|
|
basic_auth_username='foo',
|
|
basic_auth_password='hackmeplz',
|
|
)
|
|
content_type = ContentType.objects.get_for_model(carl)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=user, resource_type=content_type, resource_pk=carl.pk
|
|
)
|
|
return carl
|
|
|
|
|
|
def fake_relationship(e_type, e_id, fieldname):
|
|
return {
|
|
fieldname: {
|
|
'links': {
|
|
'self': '%sapi/entities/v1/%s/%s/relationships/%s' % (SERVICE_URL, e_type, e_id, fieldname),
|
|
'related': '%sapi/entities/v1/%s/%s/%s' % (SERVICE_URL, e_type, e_id, fieldname),
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
def fake_carl_entity(e_id=None, e_type=None, relations=None, attributes=None, **kwargs):
|
|
uuid4 = uuid.uuid4()
|
|
if e_id is None:
|
|
e_id = '%x-%x' % (uuid4.fields[0], uuid4.fields[1])
|
|
if e_type is None:
|
|
e_type = 'site'
|
|
if relations is None:
|
|
relations = ['NMaddress', 'site', 'NMaddressD', 'WOMeasure']
|
|
|
|
relationships = {}
|
|
for rel in [fake_relationship(e_type, e_id, fname) for fname in relations]:
|
|
relationships.update(rel)
|
|
|
|
entity = {
|
|
'data': {
|
|
'type': e_type,
|
|
'id': e_id,
|
|
'attributes': {
|
|
'code': 1,
|
|
'description': 'Super site',
|
|
'UOwner': 'ROOT',
|
|
},
|
|
'links': {
|
|
'self': SERVICE_URL + '/api/entities/v1/' + e_id,
|
|
},
|
|
'relationships': relationships,
|
|
},
|
|
}
|
|
|
|
if attributes is not None:
|
|
entity['data']['attributes'] = attributes
|
|
entity['data']['attributes'].update(kwargs)
|
|
|
|
return entity
|
|
|
|
|
|
def filters_to_params(filters):
|
|
return {'filter[%s]' % k: v for k, v in filters.items()}
|
|
|
|
|
|
#
|
|
# Tests
|
|
#
|
|
|
|
|
|
@responses.activate
|
|
def test_carl_check_status_ok(app, carl_conn):
|
|
responses.add(responses.GET, f'{SERVICE_URL}public/status', body='status: ok\n')
|
|
carl_conn.check_status()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'body,status',
|
|
[
|
|
('status: error\n', 200),
|
|
('status: ok\n', 500),
|
|
('', 201),
|
|
(ConnectTimeout('timeout'), 200),
|
|
(ReadTimeout('timeout'), 200),
|
|
],
|
|
)
|
|
@responses.activate
|
|
def test_carl_check_status_fail(app, carl_conn, body, status):
|
|
status_url = f'{SERVICE_URL}public/status'
|
|
responses.add(responses.GET, status_url, body=body, status=status)
|
|
with pytest.raises((APIError, requests.RequestException)):
|
|
carl_conn.check_status()
|
|
|
|
|
|
@responses.activate
|
|
def test_carl_token_authentication_success(app, carl_token_conn):
|
|
e_type = 't1'
|
|
e_id = '1234'
|
|
url = tests.utils.generic_endpoint_url('carl', f'entities/{e_type}')
|
|
expt_token = '0123456789abcdef'
|
|
token_expires = 3600000
|
|
responses.add(
|
|
responses.POST,
|
|
f'{carl_token_conn.service_url}api/auth/v1/authenticate',
|
|
json={carl_token_conn.token_header_name: expt_token, 'expires_in': token_expires},
|
|
status=200,
|
|
)
|
|
responses.add(
|
|
responses.GET,
|
|
f'{carl_token_conn.service_url}api/entities/v1/{e_type}/{e_id}',
|
|
json={
|
|
'data': {'id': 'foo', 'attributes': {'description': 'toto', 'code': 1}},
|
|
},
|
|
status=200,
|
|
)
|
|
response = app.get(url, params={'id': e_id})
|
|
assert len(responses.calls) == 2
|
|
auth_call = responses.calls[0]
|
|
auth_data = urllib.parse.parse_qs(auth_call.request.body)
|
|
assert auth_data['login'][0] == carl_token_conn.carl_username
|
|
assert auth_data['password'][0] == carl_token_conn.carl_password
|
|
assert auth_data['origin'][0] == 'publik'
|
|
|
|
call = responses.calls[1]
|
|
assert call.request.headers[carl_token_conn.token_header_name] == expt_token
|
|
|
|
assert response.json['err'] == 0
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'http_status,err_desc',
|
|
[
|
|
(401, 'Not authorized (HTTP 401) when trying to authenticate: bad credentials ?'),
|
|
(500, 'Got an HTTP error code (500) when trying to authenticate'),
|
|
],
|
|
)
|
|
@responses.activate
|
|
def test_carl_token_authentication_fails(app, carl_token_conn, http_status, err_desc):
|
|
url = tests.utils.generic_endpoint_url('carl', 'entities/t1')
|
|
auth_url = f'{carl_token_conn.service_url}api/auth/v1/authenticate'
|
|
err = '<body>some body</body> not JSON'
|
|
|
|
responses.add(
|
|
responses.POST,
|
|
auth_url,
|
|
status=http_status,
|
|
body=err,
|
|
)
|
|
response = app.get(url)
|
|
|
|
assert len(responses.calls) == 1
|
|
assert response.json['err'] != 0
|
|
assert response.json['err_class'] == APIERROR_CLS
|
|
assert response.json['err_desc'] == err_desc
|
|
assert response.json['data']['raw-response'] == err
|
|
|
|
|
|
@responses.activate
|
|
def test_carl_token_expiration(freezer, app, carl_token_conn):
|
|
e_type = 't1'
|
|
e_id = '1234'
|
|
url = tests.utils.generic_endpoint_url('carl', f'entities/{e_type}')
|
|
|
|
freezer.move_to('2024-12-13T22:43')
|
|
expt_token = '0123456789abcdef'
|
|
expt_token2 = 'fedcba9876543210'
|
|
|
|
def init_responses(expt_token):
|
|
responses.reset()
|
|
responses.add(
|
|
responses.POST,
|
|
f'{carl_token_conn.service_url}api/auth/v1/authenticate',
|
|
json={carl_token_conn.token_header_name: expt_token, 'expires_in': 3600 * 1000},
|
|
status=200,
|
|
)
|
|
responses.add(
|
|
responses.GET,
|
|
f'{carl_token_conn.service_url}api/entities/v1/{e_type}/{e_id}',
|
|
json={
|
|
'data': {'id': 'bar', 'attributes': {'description': 'titi', 'code': 2}},
|
|
},
|
|
status=200,
|
|
)
|
|
|
|
# auth calls: auth + request
|
|
init_responses(expt_token)
|
|
response = app.get(url, params={'id': e_id})
|
|
assert len(responses.calls) == 2
|
|
assert responses.calls[1].request.headers[carl_token_conn.token_header_name] == expt_token
|
|
assert response.json['err'] == 0
|
|
|
|
# call with token: request only
|
|
response = app.get(url, params={'id': e_id})
|
|
assert len(responses.calls) == 3
|
|
assert response.json['err'] == 0
|
|
|
|
# token invalidation
|
|
init_responses(expt_token2)
|
|
freezer.move_to('2024-12-13T23:59')
|
|
|
|
# auth calls again
|
|
response = app.get(url, params={'id': e_id})
|
|
assert len(responses.calls) == 2
|
|
auth_call = responses.calls[0]
|
|
auth_data = urllib.parse.parse_qs(auth_call.request.body)
|
|
assert auth_data['login'][0] == carl_token_conn.carl_username
|
|
assert auth_data['password'][0] == carl_token_conn.carl_password
|
|
assert auth_data['origin'][0] == 'publik'
|
|
|
|
call = responses.calls[1]
|
|
assert call.request.headers[carl_token_conn.token_header_name] == expt_token2
|
|
assert response.json['err'] == 0
|
|
|
|
|
|
@responses.activate
|
|
def test_carl_token_unexpected_expiration(freezer, app, carl_token_conn):
|
|
e_type = 'toto'
|
|
e_id = '42'
|
|
url = tests.utils.generic_endpoint_url('carl', f'entities/{e_type}')
|
|
|
|
freezer.move_to('2024-12-13T22:43')
|
|
expt_token = '0123456789abcdef'
|
|
|
|
responses.add(
|
|
responses.POST,
|
|
f'{carl_token_conn.service_url}api/auth/v1/authenticate',
|
|
json={carl_token_conn.token_header_name: expt_token, 'expires_in': 3600 * 1000},
|
|
status=200,
|
|
)
|
|
responses.add(
|
|
responses.GET,
|
|
f'{carl_token_conn.service_url}api/entities/v1/{e_type}/{e_id}',
|
|
json={
|
|
'errors': [{'title': 'Unauthorized', 'detail': 'Please provide valid authentication credentials'}]
|
|
},
|
|
status=401,
|
|
)
|
|
|
|
response = app.get(url, params={'id': e_id})
|
|
|
|
assert len(responses.calls) == 4
|
|
# no token: authenticate
|
|
assert responses.calls[0].request.method == 'POST'
|
|
assert responses.calls[0].request.url == f'{carl_token_conn.service_url}api/auth/v1/authenticate'
|
|
|
|
# send request to carl => unexpected token invalidation
|
|
assert responses.calls[1].request.method == 'GET'
|
|
|
|
# auth again to renew the token
|
|
assert responses.calls[2].request.method == 'POST'
|
|
assert responses.calls[2].request.url == f'{carl_token_conn.service_url}api/auth/v1/authenticate'
|
|
|
|
# retry the failed request
|
|
assert responses.calls[3].request.method == responses.calls[1].request.method
|
|
assert responses.calls[3].request.url == responses.calls[1].request.url
|
|
|
|
# but it fails again, checking returned error
|
|
assert response.json['err'] == 'authentication-error'
|
|
assert response.json['err_class'] == APIERROR_CLS
|
|
assert (
|
|
response.json['err_desc']
|
|
== 'Got an HTTP error code (401) from Carl: Unauthorized Please provide valid authentication credentials'
|
|
)
|
|
assert response.json['data']['status-code'] == 401
|
|
|
|
|
|
@pytest.mark.parametrize('timeout_expt', [ConnectTimeout, ReadTimeout])
|
|
@responses.activate
|
|
def test_carl_token_authentication_timeout(app, carl_token_conn, timeout_expt):
|
|
url = tests.utils.generic_endpoint_url('carl', 'entities/t1')
|
|
responses.add(
|
|
responses.POST, f'{carl_token_conn.service_url}api/auth/v1/authenticate', body=timeout_expt('timeout')
|
|
)
|
|
response = app.get(url)
|
|
assert response.json['err'] != 0
|
|
assert response.json['err_class'] == APIERROR_CLS
|
|
assert response.json['err_desc'] == 'Failed to fetch token: timeout'
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'bad_resp,err_desc',
|
|
[
|
|
(
|
|
'{"X-CS-Access-Token": "foobar", "expires_in": "not a number"}',
|
|
'Got an invalid token when authentication on Carl: Expected "expires_in" to be an int but got \'not a number\'',
|
|
),
|
|
(
|
|
'{"errors":[{"title": "some error", "detail": "some detail"}]}',
|
|
'Got an invalid token when authentication on Carl: \
|
|
Missing information: X-CS-Access-Token',
|
|
),
|
|
(
|
|
'<html><head><title>ERROR</title></head><body></body></html>',
|
|
'Unexpected reply from Carl when authenticating. \
|
|
JSON invalid: Expecting value: line 1 column 1 (char 0)',
|
|
),
|
|
],
|
|
)
|
|
@responses.activate
|
|
def test_carl_token_authentication_badresponse(app, carl_token_conn, bad_resp, err_desc):
|
|
url = tests.utils.generic_endpoint_url('carl', 'entities/t1')
|
|
responses.add(responses.POST, f'{carl_token_conn.service_url}api/auth/v1/authenticate', bad_resp)
|
|
response = app.get(url)
|
|
assert response.json['err'] != 0
|
|
assert response.json['err_class'] == APIERROR_CLS
|
|
assert response.json['err_desc'] == err_desc
|
|
|
|
|
|
@responses.activate
|
|
def test_carl_token_authentication_failure(app, carl_token_conn):
|
|
e_type = 't1'
|
|
e_id = '12345'
|
|
url = tests.utils.generic_endpoint_url('carl', f'entities/{e_type}')
|
|
responses.add(
|
|
responses.POST,
|
|
f'{carl_token_conn.service_url}api/auth/v1/authenticate',
|
|
body='',
|
|
status=401,
|
|
)
|
|
responses.add(
|
|
responses.GET,
|
|
f'{carl_token_conn.service_url}api/entities/v1/{e_type}/{e_id}',
|
|
json={
|
|
'data': {'id': 'foo', 'attributes': {'description': 'toto', 'code': 1}},
|
|
},
|
|
status=200,
|
|
)
|
|
response = app.get(url, params={'id': e_id})
|
|
assert len(responses.calls) == 1
|
|
auth_call = responses.calls[0]
|
|
auth_data = urllib.parse.parse_qs(auth_call.request.body)
|
|
assert auth_data['login'][0] == carl_token_conn.carl_username
|
|
assert auth_data['password'][0] == carl_token_conn.carl_password
|
|
assert auth_data['origin'][0] == 'publik'
|
|
|
|
assert response.json['err'] == 'authentication-error[token]'
|
|
assert response.json['data']['status-code'] == 401
|
|
assert response.json['err_class'] == APIERROR_CLS
|
|
assert (
|
|
response.json['err_desc']
|
|
== 'Not authorized (HTTP 401) when trying to authenticate: bad credentials ?'
|
|
)
|
|
|
|
|
|
@responses.activate
|
|
def test_carl_basic_authentication(app, carl_conn):
|
|
e_type = 'toto'
|
|
e_id = 'f00-b42'
|
|
url = tests.utils.generic_endpoint_url('carl', f'entities/{e_type}')
|
|
responses.add(
|
|
responses.POST, # should not be called !
|
|
f'{carl_conn.service_url}api/auth/v1/authenticate',
|
|
json={carl_conn.token_header_name: 'foobar', 'expires_in': 60 * 60 * 1000},
|
|
status=200,
|
|
)
|
|
responses.add(
|
|
responses.GET,
|
|
f'{carl_conn.service_url}api/entities/v1/{e_type}/{e_id}',
|
|
json={
|
|
'data': {'id': 'bar', 'attributes': {'description': 'titi', 'code': 2}},
|
|
},
|
|
status=200,
|
|
)
|
|
|
|
assert not carl_conn.authentication_token()
|
|
|
|
with pytest.raises(RuntimeError):
|
|
carl_conn.fetch_token()
|
|
|
|
response = app.get(url, params={'id': e_id})
|
|
|
|
assert len(responses.calls) == 1
|
|
call = responses.calls[0]
|
|
assert call.request.method == 'GET'
|
|
assert call.request.url == f'{carl_conn.service_url}api/entities/v1/{e_type}/{e_id}'
|
|
auth = call.request.headers['Authorization'].split(' ')[1]
|
|
|
|
assert base64.b64decode(auth).decode('utf-8') == '%s:%s' % (
|
|
carl_conn.basic_auth_username,
|
|
carl_conn.basic_auth_password,
|
|
)
|
|
|
|
assert response.json['err'] == 0
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'mock_kwargs,err_desc',
|
|
[
|
|
(
|
|
{'status': 200, 'json': {'errors': [{'title': 'Error', 'detail': 'Some error'}]}},
|
|
'Carl replies with HTTP 200 but errors set: Error Some error',
|
|
),
|
|
(
|
|
{'status': 200, 'json': 42},
|
|
'Unexpected JSON received , without data (HTTP 200): no details',
|
|
),
|
|
(
|
|
{'status': 200, 'body': '<html><head><title>Hello world!</title></head><body/></html>'},
|
|
'Invalid reply from Carl, unable to parse JSON: Expecting value: line 1 column 1 (char 0)',
|
|
),
|
|
],
|
|
)
|
|
@responses.activate
|
|
def test_carl_api_failure(app, carl_conn, mock_kwargs, err_desc):
|
|
e_type = 't1'
|
|
url = tests.utils.generic_endpoint_url('carl', f'entities/{e_type}')
|
|
|
|
responses.add(responses.GET, carl_conn.entity_url(e_type), **mock_kwargs)
|
|
|
|
resp = app.get(url)
|
|
assert resp.json['err'] != 0
|
|
assert resp.json['err_class'] == APIERROR_CLS
|
|
assert resp.json['err_desc'] == err_desc
|
|
|
|
if 'raw-response' in resp.json:
|
|
assert resp.json['raw-response'] == mock_kwargs.get('body', mock_kwargs.get('json'))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'kwargs,expt',
|
|
[
|
|
(
|
|
{'entity_type': 'foo', 'entity_id': 'bar', 'related_fieldname': 'toto'},
|
|
'%sapi/entities/v1/foo/bar/toto',
|
|
),
|
|
({'entity_type': 13}, '%sapi/entities/v1/13'),
|
|
({'entity_type': 13, 'entity_id': 12}, '%sapi/entities/v1/13/12'),
|
|
(
|
|
{'entity_type': 'foo bar', 'entity_id': '?/" &foo=42', 'related_fieldname': ' '},
|
|
'%sapi/entities/v1/foo%%20bar/%%3F/%%22%%20%%26foo%%3D42/%%20',
|
|
),
|
|
({'entity_id': 'foobar', 'related_fieldname': 'toto'}, '%sapi/entities/v1/'),
|
|
({}, '%sapi/entities/v1/'),
|
|
],
|
|
)
|
|
@responses.activate
|
|
def test_carl_entity_url(app, carl_conn, kwargs, expt):
|
|
expt_url = expt % carl_conn.service_url
|
|
assert carl_conn.entity_url(**kwargs) == expt_url
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'e_type,params',
|
|
[
|
|
('t1', {'id': '12345678-abcde'}),
|
|
('toto', {'id': '12345678-abcde', 'text_attrname': 'titi'}),
|
|
('t1', {'id': '1234', 'filter_foo': 'bar'}),
|
|
('t1', {'id': '1234', 'tata': 'titi', 'filter_foo': 'bar', 'text_attrname': 'tatata', 'q': 'foobar'}),
|
|
],
|
|
)
|
|
@responses.activate
|
|
def test_carl_endpoint_entities_by_id_ok(app, carl_conn, e_type, params):
|
|
entity = fake_carl_entity(e_type=e_type, e_id=params['id'])
|
|
|
|
url = tests.utils.generic_endpoint_url('carl', f'entities/{e_type}')
|
|
|
|
mock = responses.add( # pylint: disable=assignment-from-none
|
|
responses.GET, carl_conn.entity_url(e_type, params['id']), json=entity, status=200
|
|
)
|
|
|
|
response = app.get(url, params=params)
|
|
|
|
assert len(responses.calls) == 1
|
|
assert len(mock.calls) == 1
|
|
text = {'text': entity['data']['attributes'].get(params.get('text_attrname', 'description'))}
|
|
assert response.json['data'] == [{**entity['data'], **text}]
|
|
|
|
|
|
@responses.activate
|
|
def test_carl_endpoint_entities_by_id_fail(app, carl_conn):
|
|
e_type = 't1'
|
|
e_id = 'idontexists'
|
|
|
|
url = tests.utils.generic_endpoint_url('carl', f'entities/{e_type}')
|
|
|
|
json_404 = {
|
|
'errors': [
|
|
{
|
|
'status': '404',
|
|
'title': f'resource not found: {e_id}',
|
|
'detail': f'resource not found: {e_id}',
|
|
}
|
|
],
|
|
}
|
|
e_url = carl_conn.entity_url(e_type, e_id)
|
|
responses.add(responses.GET, e_url, json=json_404, status=404)
|
|
|
|
resp = app.get(url, params={'id': e_id}).json
|
|
|
|
assert resp['err'] == 'not-found'
|
|
assert resp['err_class'] == APIERROR_CLS
|
|
assert resp['err_desc'] == f'Got an HTTP error code (404) from Carl: {json_404["errors"][0]["title"]}'
|
|
assert resp['data']['url'] == e_url
|
|
assert resp['data']['status-code'] == 404
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'e_type,params',
|
|
[
|
|
('t1', {'id': '12345678-abcde', 'relationships': 'NMaddress, site, fake,'}),
|
|
('t1', {'id': '12345678-abcde', 'relationships': 'NMaddress ,,'}),
|
|
('t1', {'id': '12345678-abcde', 'relationships': ',,idontexists,,,site'}),
|
|
('t1', {'id': '12345678-abcde', 'relationships': ',,,,,site'}),
|
|
('t1', {'id': '12345678-abcde'}),
|
|
],
|
|
)
|
|
@responses.activate
|
|
def test_carl_endpoint_entities_by_id_rel_ok(app, carl_conn, e_type, params):
|
|
e_id = params['id']
|
|
|
|
entities = {}
|
|
url_site = f'{carl_conn.service_url}api/entities/v1/{e_type}/{e_id}/site'
|
|
attr_site = {'a': 'b', 'c': 'd'}
|
|
entities['site'] = fake_carl_entity(e_type='site', attributes=attr_site)
|
|
|
|
url_addr = f'{carl_conn.service_url}api/entities/v1/{e_type}/{e_id}/NMaddress'
|
|
attr_addr = {'q': 42, 'foo': 'bar'}
|
|
entities['NMaddress'] = fake_carl_entity(e_type='addressees', attributes=attr_addr)
|
|
|
|
relations = ['NMaddress', 'site', 'toto', 'titi', 'tutu']
|
|
url_ent = f'{carl_conn.service_url}api/entities/v1/{e_type}/{e_id}'
|
|
attr_ent = {'foo': 'bar', 'toto': 'tata'}
|
|
entitie = fake_carl_entity(e_id, e_type, relations=relations, attributes=attr_ent)
|
|
|
|
mock = {}
|
|
mock['ent'] = responses.add( # pylint: disable=assignment-from-none
|
|
responses.GET, url_ent, json=entitie, status=200
|
|
)
|
|
mock['site'] = responses.add( # pylint: disable=assignment-from-none
|
|
responses.GET, url_site, json=entities['site'], status=200
|
|
)
|
|
mock['NMaddress'] = responses.add( # pylint: disable=assignment-from-none
|
|
responses.GET, url_addr, json=entities['NMaddress'], status=200
|
|
)
|
|
|
|
url = tests.utils.generic_endpoint_url('carl', f'entities/{e_type}')
|
|
|
|
resp = app.get(url, params=params)
|
|
|
|
assert resp.json['err'] == 0
|
|
|
|
assert len(mock['ent'].calls) == 1
|
|
if 'relationships' in params:
|
|
for rel in ('NMaddress', 'site'):
|
|
if rel in params['relationships']:
|
|
assert len(mock[rel].calls) == 1
|
|
assert rel in resp.json['data'][0]['relationships']
|
|
assert 'data' in resp.json['data'][0]['relationships'][rel]
|
|
assert (
|
|
resp.json['data'][0]['relationships'][rel]['data']['attributes']
|
|
== entities[rel]['data']['attributes']
|
|
)
|
|
assert set(resp.json['data'][0]['relationships'].keys()) == set(relations)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'e_type,params',
|
|
[
|
|
('t1', {'q': 'toto'}),
|
|
('site', {'q': 'toto', 'text_attrname': 'foobar'}),
|
|
('wo', {'q': 'toto', 'text_attrname': 'foobar', 'filter_foo': 'bar'}),
|
|
('wo', {'q': 'toto', 'text_attrname': 'foobar', 'filter_foo': 'bar', 'filter_': 'test'}),
|
|
('t2', {'q': 'toto', 'relationships': 'toto, titi, tutu'}),
|
|
('t2', {'q': 'toto', 'relationships': 'toto, titi, tutu', 'id': ''}),
|
|
],
|
|
)
|
|
@responses.activate
|
|
def test_carl_endpoint_entities_by_q_ok(app, carl_conn, e_type, params):
|
|
entities = [fake_carl_entity(e_type=e_type) for _ in range(5)]
|
|
|
|
url = tests.utils.generic_endpoint_url('carl', f'entities/{e_type}')
|
|
|
|
text_attrname = params.get('text_attrname', 'description')
|
|
qmatch = {'filter[%s][LIKE]' % text_attrname: params['q']}
|
|
qmatch.update(
|
|
{
|
|
'filter[%s]' % k[len('filter_') :]: v
|
|
for k, v in params.items()
|
|
if k.startswith('filter_') and len(k) > len('filter_')
|
|
}
|
|
)
|
|
mock = responses.add( # pylint: disable=assignment-from-none
|
|
responses.GET,
|
|
carl_conn.entity_url(e_type),
|
|
json={'data': [ent['data'] for ent in entities]},
|
|
match=[responses.matchers.query_param_matcher(qmatch)],
|
|
status=200,
|
|
)
|
|
|
|
resp = app.get(url, params=params)
|
|
|
|
assert len(responses.calls) == 1
|
|
assert len(mock.calls) == 1
|
|
|
|
assert resp.json['err'] == 0
|
|
|
|
for ret in resp.json['data']:
|
|
for ent in entities:
|
|
if ent['data']['id'] == ret['id']:
|
|
assert ent['data']['attributes'] == ret['attributes']
|
|
assert ent['data']['type'] == ret['type']
|
|
assert ent['data']['attributes'].get(text_attrname) == ret['text']
|
|
|
|
|
|
@responses.activate
|
|
def test_carl_endpoint_entities_by_filter_fail(app, carl_conn):
|
|
e_type = 'foo'
|
|
bad_fieldname = 'titi'
|
|
|
|
# Invalid filter response
|
|
error = {
|
|
'errors': [
|
|
{
|
|
'status': '400',
|
|
'code': 'UNKNOWN_PARAMETER',
|
|
'title': 'unknown parameter',
|
|
'detail': f"Failed to resolve path to field '{bad_fieldname}' from {e_type}",
|
|
'source': {'parameter': f'filter[{bad_fieldname}]'},
|
|
}
|
|
]
|
|
}
|
|
responses.add(
|
|
responses.GET,
|
|
carl_conn.entity_url(e_type),
|
|
match=[responses.matchers.query_param_matcher({f'filter[{bad_fieldname}]': '42'})],
|
|
json=error,
|
|
status=400,
|
|
)
|
|
|
|
url = tests.utils.generic_endpoint_url('carl', f'entities/{e_type}')
|
|
|
|
resp = app.get(url, params={f'filter_{bad_fieldname}': 42})
|
|
|
|
assert resp.json['err'] != 0
|
|
assert (
|
|
resp.json['err_desc']
|
|
== f'Got an HTTP error code (400) from Carl: \
|
|
unknown parameter(parameter filter[{bad_fieldname}]) \
|
|
Failed to resolve path to field \'{bad_fieldname}\' from {e_type}'
|
|
)
|
|
|
|
qmock = responses.add( # pylint: disable=assignment-from-none
|
|
responses.GET,
|
|
carl_conn.entity_url(e_type),
|
|
match=[responses.matchers.query_param_matcher({f'filter[{bad_fieldname}][LIKE]': '42'})],
|
|
json=error,
|
|
status=400,
|
|
)
|
|
|
|
resp = app.get(url, params={'text_attrname': bad_fieldname, 'q': 42})
|
|
assert resp.json['err'] != 0
|
|
assert len(qmock.calls) == 1
|
|
assert (
|
|
resp.json['err_desc']
|
|
== f'Got an HTTP error code (400) from Carl: \
|
|
unknown parameter(parameter filter[{bad_fieldname}]) \
|
|
Failed to resolve path to field \'{bad_fieldname}\' from {e_type}'
|
|
)
|
|
|
|
|
|
@responses.activate
|
|
def test_carl_endpoint_create_entity_ok(app, carl_conn):
|
|
want = {
|
|
'attrs': {
|
|
'priority': 'HIGH',
|
|
'libelle': 'Probleme',
|
|
'owner': 'Toto',
|
|
},
|
|
'linked': {
|
|
'site_pb': {'type': 'site', 'filter': {'code': '10'}},
|
|
'address': {'type': 'addressees', 'id': '1234-56'},
|
|
'testfield': {'type': 'test', 'filter': {'a': 'b', 'c': 'd'}},
|
|
},
|
|
'related': {
|
|
'details': {'type': 'metadata', 'attrs': {'prop1': 'val1', 'prop2': 42}},
|
|
'infos': {'type': 'information', 'attrs': {'prop1': 'value', 'prop2': 'foobar'}},
|
|
},
|
|
}
|
|
|
|
site10 = fake_carl_entity(e_type='site', code=10)
|
|
testfield = fake_carl_entity()
|
|
metadata = fake_carl_entity(e_type='addressees', attributes=want['related']['details']['attrs'])
|
|
infos = fake_carl_entity()
|
|
entitie = fake_carl_entity(e_type='wo', attributes=want['attrs'])
|
|
|
|
responses.add(
|
|
responses.GET,
|
|
carl_conn.entity_url('site'),
|
|
match=[responses.matchers.query_param_matcher({'filter[code]': '10'})],
|
|
json={'data': [site10['data']]},
|
|
status=200,
|
|
)
|
|
responses.add(
|
|
responses.GET,
|
|
carl_conn.entity_url('test'),
|
|
match=[responses.matchers.query_param_matcher({'filter[a]': 'b', 'filter[c]': 'd'})],
|
|
json={'data': [testfield['data']]},
|
|
status=200,
|
|
)
|
|
responses.add(
|
|
responses.POST,
|
|
carl_conn.entity_url('metadata'),
|
|
json=metadata,
|
|
status=201,
|
|
)
|
|
responses.add(
|
|
responses.POST,
|
|
carl_conn.entity_url('information'),
|
|
json=infos,
|
|
status=201,
|
|
)
|
|
responses.add(
|
|
responses.POST,
|
|
carl_conn.entity_url('wo'),
|
|
json=entitie,
|
|
status=201,
|
|
)
|
|
|
|
url = tests.utils.generic_endpoint_url('carl', 'entity/wo')
|
|
resp = app.post(url, json.dumps(want))
|
|
|
|
assert len(responses.calls) == 5
|
|
|
|
call_site = responses.calls[0]
|
|
assert call_site.request.method == 'GET'
|
|
spl = urllib.parse.urlsplit(call_site.request.url)
|
|
url = urllib.parse.urlunsplit(spl[:3] + ('', ''))
|
|
qs = {k: v[0] for k, v in urllib.parse.parse_qs(spl[3]).items()}
|
|
assert url == carl_conn.entity_url('site')
|
|
assert qs == filters_to_params(want['linked']['site_pb']['filter'])
|
|
|
|
call_testfield = responses.calls[1]
|
|
assert call_testfield.request.method == 'GET'
|
|
spl = urllib.parse.urlsplit(call_testfield.request.url)
|
|
url = urllib.parse.urlunsplit(spl[:3] + ('', ''))
|
|
qs = {k: v[0] for k, v in urllib.parse.parse_qs(spl[3]).items()}
|
|
assert url == carl_conn.entity_url('test')
|
|
assert qs == filters_to_params(want['linked']['testfield']['filter'])
|
|
|
|
call_meta = responses.calls[2]
|
|
assert call_meta.request.method == 'POST'
|
|
assert call_meta.request.headers['Content-Type'] == 'application/vnd.api+json'
|
|
assert call_meta.request.url == carl_conn.entity_url('metadata')
|
|
assert json.loads(call_meta.request.body) == {
|
|
'data': {'type': 'metadata', 'attributes': want['related']['details']['attrs']}
|
|
}
|
|
|
|
call_info = responses.calls[3]
|
|
assert call_info.request.method == 'POST'
|
|
assert call_meta.request.headers['Content-Type'] == 'application/vnd.api+json'
|
|
assert call_info.request.url == carl_conn.entity_url('information')
|
|
assert json.loads(call_info.request.body) == {
|
|
'data': {'type': 'information', 'attributes': want['related']['infos']['attrs']}
|
|
}
|
|
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data']['id'] == entitie['data']['id']
|
|
assert resp.json['data']['type'] == entitie['data']['type']
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'badwant',
|
|
[
|
|
{ # id & filter not allowed together
|
|
'attrs': {},
|
|
'linked': {
|
|
'toto': {
|
|
'type': 'foobar',
|
|
'id': '42',
|
|
'filter': {'should_not': 'be_there'},
|
|
},
|
|
},
|
|
},
|
|
{ # Must specified one of id or filter in linked
|
|
'attrs': {},
|
|
'linked': {'toto': {'type': 'titi'}},
|
|
},
|
|
{ # "Double declaration"
|
|
'attrs': {},
|
|
'linked': {'t1': {'type': 'titi', 'id': 'foo'}},
|
|
'related': {'t1': {'type': 'titi', 'attrs': {'toto': 'tata'}}},
|
|
},
|
|
{ # attrs must be an object
|
|
'attrs': 'tata',
|
|
},
|
|
{ # linked/.*/ must be an object
|
|
'attrs': {},
|
|
'linked': {'foo': 'bar'},
|
|
},
|
|
{ # related/.*/ must be an object
|
|
'attrs': {},
|
|
'related': {'toto': 'titi'},
|
|
},
|
|
{ # type is mandatory for linked fields
|
|
'attrs': {},
|
|
'linked': {'rel': {'id': 12}},
|
|
},
|
|
{ # type is mandatory for related fields
|
|
'attrs': {},
|
|
'related': {'rel': {'attrs': {'toto': 'tata'}}},
|
|
},
|
|
],
|
|
)
|
|
@responses.activate
|
|
def test_carl_endpoint_create_entity_badfmt(app, carl_conn, badwant):
|
|
carl_type = 'toto'
|
|
url = tests.utils.generic_endpoint_url('carl', f'entity/{carl_type}')
|
|
resp = app.post(url, json.dumps(badwant), expect_errors=True)
|
|
|
|
assert len(responses.calls) == 0
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@responses.activate
|
|
def test_carl_endpoint_create_entity_bad_linked_filter(app, carl_conn):
|
|
want = {
|
|
'attrs': {},
|
|
'linked': {'toto': {'type': 't2', 'filter': {'titi': 'tutu'}}},
|
|
}
|
|
|
|
error = {
|
|
'errors': [
|
|
{
|
|
'status': '400',
|
|
'code': 'UNKNOWN_PARAMETER',
|
|
'title': 'unknown parameter',
|
|
'detail': "Failed to resolve path to field 'titi' from t2",
|
|
'source': {'parameter': 'filter[titi]'},
|
|
}
|
|
]
|
|
}
|
|
|
|
# Invalid filter response
|
|
responses.add(
|
|
responses.GET,
|
|
carl_conn.entity_url('t2'),
|
|
match=[responses.matchers.query_param_matcher({'filter[titi]': 'tutu'})],
|
|
json=error,
|
|
status=400,
|
|
)
|
|
|
|
url = tests.utils.generic_endpoint_url('carl', 'entity/t1')
|
|
resp = app.post(url, json.dumps(want))
|
|
|
|
assert len(responses.calls) == 1
|
|
|
|
assert resp.json['err'] != 0
|
|
assert resp.json['err_class'] == APIERROR_CLS
|
|
assert (
|
|
resp.json['err_desc']
|
|
== 'Error looking for linked fields: \
|
|
\'toto\' (Got an HTTP error code (400) from Carl: unknown parameter(parameter filter[titi]) Failed to resolve path to field \'titi\' from t2)'
|
|
)
|
|
|
|
# Filter returns no data
|
|
responses.reset()
|
|
responses.add(
|
|
responses.GET,
|
|
carl_conn.entity_url('t2'),
|
|
match=[responses.matchers.query_param_matcher({'filter[titi]': 'tutu'})],
|
|
json={'data': []},
|
|
status=200,
|
|
)
|
|
|
|
resp = app.post(url, json.dumps(want))
|
|
|
|
assert len(responses.calls) == 1
|
|
|
|
assert resp.json['err'] != 0
|
|
assert resp.json['err_class'] == APIERROR_CLS
|
|
assert (
|
|
resp.json['err_desc']
|
|
== 'Error looking for linked fields: \
|
|
\'toto\' (No entities returned with given filters (titi=\'tutu\') for field \'toto\')'
|
|
)
|
|
|
|
# Filter returnes multiple data
|
|
responses.reset()
|
|
responses.add(
|
|
responses.GET,
|
|
carl_conn.entity_url('t2'),
|
|
match=[responses.matchers.query_param_matcher({'filter[titi]': 'tutu'})],
|
|
json={'data': [fake_carl_entity()['data'], fake_carl_entity()['data']]},
|
|
status=200,
|
|
)
|
|
|
|
resp = app.post(url, json.dumps(want))
|
|
|
|
assert len(responses.calls) == 1
|
|
|
|
assert resp.json['err'] != 0
|
|
assert resp.json['err_class'] == APIERROR_CLS
|
|
assert (
|
|
resp.json['err_desc']
|
|
== 'Error looking for linked fields: \
|
|
\'toto\' (More than one (2) entities returned with given filters (titi=\'tutu\') for field \'toto\')'
|
|
)
|
|
|
|
|
|
@responses.activate
|
|
def test_carl_endpoint_create_entity_badrequest(app, carl_conn):
|
|
bad_reply = {
|
|
'errors': [{'status': '400', 'title': 'BAD_REQUEST', 'detail': 'attribute TestField not found'}]
|
|
}
|
|
|
|
want = {
|
|
'attrs': {'TestField': 'toto'},
|
|
'linked': {'l1': {'type': 't2', 'id': '2'}, 'l2': {'type': 't3', 'filter': {'code': 10}}},
|
|
'related': {
|
|
'rel1': {
|
|
'type': 't4',
|
|
'attrs': {
|
|
'TestField': 'tata',
|
|
},
|
|
},
|
|
},
|
|
}
|
|
|
|
fake_t2 = fake_carl_entity(e_type='t2', attributes={'code': 10})
|
|
fake_t4 = fake_carl_entity(e_type='t4', attributes={'TestField': 'tata'})
|
|
fake_t1 = fake_carl_entity(e_type='t1', attributes={'TestField': 'tata'})
|
|
|
|
def init_responses(t4_fail=False):
|
|
'''Init responses, if t4_fail is True, fail on rel1 related entitie
|
|
creation (else fail on t1)
|
|
'''
|
|
responses.reset()
|
|
responses.add(
|
|
responses.GET,
|
|
carl_conn.entity_url('t3'),
|
|
match=[responses.matchers.query_param_matcher({'filter[code]': 10})],
|
|
json={'data': [fake_t2['data']]},
|
|
status=200,
|
|
)
|
|
responses.add(
|
|
responses.POST,
|
|
carl_conn.entity_url('t4'),
|
|
json=bad_reply if t4_fail else {'data': fake_t4['data']},
|
|
status=400 if t4_fail else 200,
|
|
)
|
|
responses.add(
|
|
responses.POST,
|
|
carl_conn.entity_url('t1'),
|
|
json=bad_reply if not t4_fail else {'data': fake_t1['data']},
|
|
status=400 if not t4_fail else 200,
|
|
)
|
|
responses.add(
|
|
responses.DELETE,
|
|
carl_conn.entity_url('t4', fake_t4['data']['id']),
|
|
json={'errors': [{'title': 'Error', 'detail': 'Details'}]},
|
|
status=400,
|
|
)
|
|
|
|
url = tests.utils.generic_endpoint_url('carl', 'entity/t1')
|
|
|
|
init_responses()
|
|
resp = app.post(url, json.dumps(want))
|
|
|
|
# Get l2 create rel1 fail on creation, delete rel1
|
|
assert len(responses.calls) == 4
|
|
|
|
assert resp.json['err'] == 'creation-error[rollback]'
|
|
assert resp.json['err_class'] == APIERROR_CLS
|
|
assert (
|
|
resp.json['err_desc']
|
|
== 'Error creating entity: \
|
|
Got an HTTP error code (400) from Carl: BAD_REQUEST attribute TestField not found. \
|
|
Then error during related entities rollback: \
|
|
Error during entities cleanup: \
|
|
HTTP error (400) during entity (%s/%s) deletion: Error Details'
|
|
% ('t4', fake_t4['data']['id'])
|
|
)
|
|
|
|
init_responses(True)
|
|
resp = app.post(url, json.dumps(want))
|
|
|
|
# Get l2 and fail on rel1 creation
|
|
assert len(responses.calls) == 2
|
|
|
|
assert resp.json['err'] != 0
|
|
assert resp.json['err_class'] == APIERROR_CLS
|
|
assert resp.json['err_desc'] == 'Error with related fields rel1'
|
|
assert (
|
|
resp.json['data']['errors']
|
|
== 'Error with field rel1 (Got an HTTP error code (400) from Carl: BAD_REQUEST attribute TestField not found)'
|
|
)
|
|
|
|
|
|
@responses.activate
|
|
def test_carl_endpoint_create_entity_failfast(app, carl_conn):
|
|
want = {
|
|
'attrs': {},
|
|
'linked': {
|
|
'l1': {'type': 't2', 'filter': {'foo': 'bar'}},
|
|
'l2': {'type': 't3', 'filter': {'toto': 'titi'}},
|
|
},
|
|
'related': {
|
|
'r1': {'type': 't4', 'attrs': {'tata': 'tutu'}},
|
|
'r2': {'type': 't5', 'attrs': {'foobar': '42'}},
|
|
},
|
|
}
|
|
|
|
some_fake = fake_carl_entity()
|
|
some_fake2 = fake_carl_entity()
|
|
fake_error = {'errors': [{'title': 'Some error', 'detail': 'Some detail'}]}
|
|
|
|
def init_responses(fails_on=0, cleanup_err=True):
|
|
mock_args = [
|
|
(
|
|
[responses.GET, carl_conn.entity_url('t2')],
|
|
{
|
|
'match': [responses.matchers.query_param_matcher({'filter[foo]': 'bar'})],
|
|
'json': {'data': [some_fake['data']]},
|
|
'status': 200,
|
|
},
|
|
),
|
|
(
|
|
[responses.GET, carl_conn.entity_url('t3')],
|
|
{
|
|
'match': [responses.matchers.query_param_matcher({'filter[toto]': 'titi'})],
|
|
'json': {'data': [some_fake['data']]},
|
|
'status': 200,
|
|
},
|
|
),
|
|
(
|
|
[responses.POST, carl_conn.entity_url('t4')],
|
|
{'json': {'data': some_fake['data']}, 'status': 200},
|
|
),
|
|
(
|
|
[responses.POST, carl_conn.entity_url('t5')],
|
|
{'json': {'data': some_fake2['data']}, 'status': 200},
|
|
),
|
|
(
|
|
[responses.POST, carl_conn.entity_url('t1')],
|
|
{'json': {'data': some_fake['data']}, 'status': 200},
|
|
),
|
|
]
|
|
responses.reset()
|
|
mocks = []
|
|
for i, (args, kwargs) in enumerate(mock_args):
|
|
kwargs = kwargs.copy()
|
|
if i == fails_on:
|
|
kwargs['status'] = 400
|
|
kwargs['json'] = fake_error
|
|
mocks.append(responses.add(*args, **kwargs))
|
|
|
|
for fake in (some_fake, some_fake2):
|
|
if cleanup_err:
|
|
responses.add(
|
|
responses.DELETE,
|
|
carl_conn.entity_url(fake['data']['type'], fake['data']['id']),
|
|
json=fake_error,
|
|
status=400,
|
|
)
|
|
else:
|
|
responses.add(
|
|
responses.DELETE,
|
|
carl_conn.entity_url(fake['data']['type'], fake['data']['id']),
|
|
status=204,
|
|
)
|
|
|
|
return mocks
|
|
|
|
url = tests.utils.generic_endpoint_url('carl', 'entity/t1')
|
|
|
|
mocks = init_responses(0)
|
|
resp = app.post(url, json.dumps(want))
|
|
assert len(mocks[0].calls) == 1
|
|
assert len(mocks[1].calls) == 1
|
|
assert len(mocks[2].calls) == 0
|
|
assert len(mocks[3].calls) == 0
|
|
assert len(responses.calls) == 2
|
|
|
|
assert resp.json['err'] != 0
|
|
assert resp.json['err_class'] == APIERROR_CLS
|
|
assert (
|
|
resp.json['err_desc']
|
|
== "Error looking for linked fields: \
|
|
'l1' (Got an HTTP error code (400) from Carl: Some error Some detail)"
|
|
)
|
|
|
|
mocks = init_responses(1)
|
|
resp = app.post(url, json.dumps(want))
|
|
assert len(mocks[0].calls) == 1
|
|
assert len(mocks[1].calls) == 1
|
|
assert len(mocks[2].calls) == 0
|
|
assert len(mocks[3].calls) == 0
|
|
assert len(mocks[4].calls) == 0
|
|
assert len(responses.calls) == 2
|
|
|
|
assert resp.json['err'] != 0
|
|
assert resp.json['err_class'] == APIERROR_CLS
|
|
assert (
|
|
resp.json['err_desc']
|
|
== "Error looking for linked fields: \
|
|
'l2' (Got an HTTP error code (400) from Carl: Some error Some detail)"
|
|
)
|
|
|
|
mocks = init_responses(2)
|
|
resp = app.post(url, json.dumps(want))
|
|
assert len(mocks[0].calls) == 1
|
|
assert len(mocks[1].calls) == 1
|
|
assert len(mocks[2].calls) == 1
|
|
assert len(mocks[3].calls) == 0
|
|
assert len(mocks[4].calls) == 0
|
|
assert len(responses.calls) == 3
|
|
|
|
assert resp.json['err'] != 0
|
|
assert resp.json['err_class'] == APIERROR_CLS
|
|
assert resp.json['err_desc'] == 'Error with related fields r1'
|
|
|
|
mocks = init_responses(3)
|
|
resp = app.post(url, json.dumps(want))
|
|
assert len(mocks[0].calls) == 1
|
|
assert len(mocks[1].calls) == 1
|
|
assert len(mocks[2].calls) == 1
|
|
assert len(mocks[3].calls) == 1
|
|
assert len(mocks[4].calls) == 0
|
|
assert len(responses.calls) == 4 + 1 # 4 + cleanup
|
|
|
|
assert resp.json['err'] == 'creation-error[rollback]'
|
|
assert resp.json['err_class'] == APIERROR_CLS
|
|
assert (
|
|
resp.json['err_desc']
|
|
== 'Error with related fields r2. \
|
|
On rollback: Error during entities cleanup: HTTP error (400) during entity (%s/%s) deletion: Some error Some detail'
|
|
% (
|
|
some_fake['data']['type'],
|
|
some_fake['data']['id'],
|
|
)
|
|
)
|
|
|
|
mocks = init_responses(4)
|
|
resp = app.post(url, json.dumps(want))
|
|
assert len(mocks[0].calls) == 1
|
|
assert len(mocks[1].calls) == 1
|
|
assert len(mocks[2].calls) == 1
|
|
assert len(mocks[3].calls) == 1
|
|
assert len(mocks[4].calls) == 1
|
|
assert len(responses.calls) == 5 + 2 # 5 + cleanups
|
|
|
|
assert resp.json['err'] == 'creation-error[rollback]'
|
|
assert resp.json['err_class'] == APIERROR_CLS
|
|
assert (
|
|
resp.json['err_desc']
|
|
== 'Error creating entity: \
|
|
Got an HTTP error code (400) from Carl: \
|
|
Some error Some detail. \
|
|
Then error during related entities rollback: \
|
|
Error during entities cleanup: \
|
|
HTTP error (400) during entity (%s/%s) deletion: Some error Some detail, \
|
|
HTTP error (400) during entity (%s/%s) deletion: Some error Some detail'
|
|
% (
|
|
some_fake['data']['type'],
|
|
some_fake['data']['id'],
|
|
some_fake2['data']['type'],
|
|
some_fake2['data']['id'],
|
|
)
|
|
)
|
|
|
|
mocks = init_responses(4, False)
|
|
resp = app.post(url, json.dumps(want))
|
|
assert len(mocks[0].calls) == 1
|
|
assert len(mocks[1].calls) == 1
|
|
assert len(mocks[2].calls) == 1
|
|
assert len(mocks[3].calls) == 1
|
|
assert len(mocks[4].calls) == 1
|
|
assert len(responses.calls) == 5 + 2 # 5 + cleanups
|
|
|
|
assert resp.json['err'] != 0
|
|
assert resp.json['err_class'] == APIERROR_CLS
|
|
assert resp.json['err_desc'] == 'Got an HTTP error code (400) from Carl: Some error Some detail'
|
|
|
|
|
|
@pytest.mark.parametrize('timeout_expt', [ConnectTimeout, ReadTimeout])
|
|
@responses.activate
|
|
def test_carl_timeout(app, carl_conn, timeout_expt):
|
|
want = {
|
|
'attrs': {},
|
|
'related': {
|
|
'r1': {'type': 't2', 'attrs': {'tata': 'tutu'}},
|
|
},
|
|
}
|
|
|
|
fake = fake_carl_entity(e_type='t2')
|
|
|
|
# Allow creation of related entity
|
|
responses.add(
|
|
responses.POST,
|
|
carl_conn.entity_url('t2'),
|
|
json=fake,
|
|
status=200,
|
|
)
|
|
# Timeout on "global" entity creation
|
|
responses.add(
|
|
responses.POST,
|
|
carl_conn.entity_url('t1'),
|
|
body=timeout_expt('timeout'),
|
|
)
|
|
# Timeout on entity cleanup
|
|
responses.add(
|
|
responses.DELETE,
|
|
carl_conn.entity_url('t2', fake['data']['id']),
|
|
body=timeout_expt('timeout'),
|
|
)
|
|
|
|
url = tests.utils.generic_endpoint_url('carl', 'entity/t1')
|
|
resp = app.post(url, json.dumps(want))
|
|
|
|
assert resp.json['err'] == 'creation-error[rollback]'
|
|
assert resp.json['err_class'] == APIERROR_CLS
|
|
assert (
|
|
resp.json['err_desc']
|
|
== 'Error creating entity: \
|
|
Carl interaction error: timeout. \
|
|
Then error during related entities rollback: \
|
|
Error during entities cleanup: Error sending deletion request (%s/%s) to Carl: timeout'
|
|
% (fake['data']['type'], fake['data']['id'])
|
|
)
|