336 lines
13 KiB
Python
336 lines
13 KiB
Python
import json
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
import responses
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
from passerelle.apps.matrix42.models import Matrix42
|
|
from passerelle.base.models import AccessRight, ApiUser
|
|
from tests.utils import FakedResponse, generic_endpoint_url
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
TOKEN = '{"RawToken": "token2","LifeTime":"2200-09-23T06:39:31.5285469Z"}'
|
|
USERS = (
|
|
'{"Result":[{"ID":"a9386c3e-cb7a-ed11-a3bb-000d3aaa0172","DisplayString":"User1, Leo",'
|
|
'"Expression-TypeCase":"46c86c68-42ae-4089-8398-6e4140fe8658",'
|
|
'"Expression-TypeID":"46c86c68-42ae-4089-8398-6e4140fe8658"},'
|
|
'{"ID":"12386c3e-cb7a-ed11-a3bb-00bd3aaa0111","DisplayString":"User2, Blah",'
|
|
'"Expression-TypeCase":"46c86c68-42ae-4089-8398-6e4140fe8658",'
|
|
'"Expression-TypeID":"46c86c68-42ae-4089-8398-6e4140fe8658"}],'
|
|
'"Schema":[{"ColumnName":"ID","ColumnType":"GuidType","Localizable":false},'
|
|
'{"ColumnName":"DisplayString","ColumnType":"StringType","Localizable":false}]}'
|
|
)
|
|
USER = '{"ID":"a9386c3e-cb7a-ed11-a3bb-000d3aaa0172","DisplayString":"User1, Leo"}'
|
|
OBJECT = '{"ID":"424242","SPSActivityClassBase":{"TicketNumber":"TCK0000153","TimeStamp":"AAAAAAHlWr4="}}'
|
|
|
|
|
|
@pytest.fixture
|
|
def matrix42():
|
|
return Matrix42.objects.create(slug='test', base_url='https://matrix42.example.net/api/', token='token1')
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.request')
|
|
def test_matrix42_fragment(mocked_request, app, matrix42):
|
|
endpoint = generic_endpoint_url('matrix42', 'fragment', slug=matrix42.slug)
|
|
assert endpoint == '/matrix42/test/fragment'
|
|
endpoint += '/SPSUserClassBase'
|
|
|
|
params = {}
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content=TOKEN, status_code=200),
|
|
FakedResponse(content=USERS, status_code=200),
|
|
]
|
|
|
|
resp = app.get(endpoint, params=params, status=403)
|
|
assert mocked_request.call_count == 0
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'django.core.exceptions.PermissionDenied'
|
|
|
|
# open access
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(matrix42)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=matrix42.pk
|
|
)
|
|
|
|
# get all users
|
|
resp = app.get(endpoint, params=params, status=200)
|
|
assert mocked_request.call_count == 2
|
|
get_token, get_users = mocked_request.call_args_list
|
|
assert get_token[0] == (
|
|
'POST',
|
|
'https://matrix42.example.net/api/ApiToken/GenerateAccessTokenFromApiToken',
|
|
)
|
|
assert get_token[1]['json'] == get_token[1]['params'] == None
|
|
assert get_token[1]['headers']['Authorization'] == 'Bearer token1'
|
|
assert get_users[0] == (
|
|
'GET',
|
|
'https://matrix42.example.net/api/data/fragments/SPSUserClassBase/schema-info',
|
|
)
|
|
assert get_users[1]['json'] is None
|
|
assert get_users[1]['params'] == {}
|
|
assert get_users[1]['headers']['Authorization'] == 'Bearer token2'
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 2
|
|
assert resp.json['data'][0]['id'] == resp.json['data'][0]['ID'] == 'a9386c3e-cb7a-ed11-a3bb-000d3aaa0172'
|
|
assert resp.json['data'][0]['text'] == resp.json['data'][0]['DisplayString'] == 'User1, Leo'
|
|
|
|
# get all users, with parameters
|
|
params['id_template'] = 'id:{{ID}}'
|
|
params['template'] = 'ds:{{DisplayString}}'
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content=TOKEN, status_code=200),
|
|
FakedResponse(content=USERS, status_code=200),
|
|
]
|
|
resp = app.get(endpoint, params=params, status=200)
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 2
|
|
assert resp.json['data'][0]['id'] == 'id:a9386c3e-cb7a-ed11-a3bb-000d3aaa0172'
|
|
assert resp.json['data'][0]['text'] == 'ds:User1, Leo'
|
|
|
|
# search user
|
|
params['q'] = 'User'
|
|
resp = app.get(endpoint, params=params, status=400)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_desc'] == 'q needs a search_column parameter'
|
|
params['search_column'] = 'DisplayString'
|
|
params['columns'] = 'DisplayString'
|
|
mocked_request.reset_mock()
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content=TOKEN, status_code=200),
|
|
FakedResponse(content=USERS, status_code=200),
|
|
]
|
|
resp = app.get(endpoint, params=params, status=200)
|
|
_, get_users = mocked_request.call_args_list
|
|
assert get_users[0] == (
|
|
'GET',
|
|
'https://matrix42.example.net/api/data/fragments/SPSUserClassBase/schema-info',
|
|
)
|
|
assert get_users[1]['params'] == {'columns': 'DisplayString', 'where': "DisplayString LIKE '%User%'"}
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 2
|
|
assert resp.json['data'][0]['id'] == 'id:a9386c3e-cb7a-ed11-a3bb-000d3aaa0172'
|
|
assert resp.json['data'][0]['text'] == 'ds:User1, Leo'
|
|
|
|
# filter
|
|
del params['q']
|
|
params['filter'] = '1=1'
|
|
mocked_request.reset_mock()
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content=TOKEN, status_code=200),
|
|
FakedResponse(content=USERS, status_code=200),
|
|
]
|
|
resp = app.get(endpoint, params=params, status=200)
|
|
_, get_users = mocked_request.call_args_list
|
|
assert get_users[0] == (
|
|
'GET',
|
|
'https://matrix42.example.net/api/data/fragments/SPSUserClassBase/schema-info',
|
|
)
|
|
assert get_users[1]['params'] == {
|
|
'columns': 'DisplayString',
|
|
'where': '1=1',
|
|
}
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 2
|
|
|
|
# filter & q
|
|
params['q'] = 'User'
|
|
params['filter'] = '1=1'
|
|
mocked_request.reset_mock()
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content=TOKEN, status_code=200),
|
|
FakedResponse(content=USERS, status_code=200),
|
|
]
|
|
resp = app.get(endpoint, params=params, status=200)
|
|
_, get_users = mocked_request.call_args_list
|
|
assert get_users[0] == (
|
|
'GET',
|
|
'https://matrix42.example.net/api/data/fragments/SPSUserClassBase/schema-info',
|
|
)
|
|
assert get_users[1]['params'] == {
|
|
'columns': 'DisplayString',
|
|
'where': "DisplayString LIKE '%User%' AND 1=1",
|
|
}
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 2
|
|
|
|
# get one user
|
|
del params['q']
|
|
params['id'] = 'a9386c3e-cb7a-ed11-a3bb-000d3aaa0172'
|
|
mocked_request.reset_mock()
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content=TOKEN, status_code=200),
|
|
FakedResponse(content=USER, status_code=200),
|
|
]
|
|
resp = app.get(endpoint, params=params, status=200)
|
|
_, get_users = mocked_request.call_args_list
|
|
assert get_users[0] == (
|
|
'GET',
|
|
'https://matrix42.example.net/api/data/fragments/SPSUserClassBase/a9386c3e-cb7a-ed11-a3bb-000d3aaa0172',
|
|
)
|
|
assert resp.json['err'] == 0
|
|
assert len(resp.json['data']) == 1
|
|
assert resp.json['data'][0]['id'] == 'id:a9386c3e-cb7a-ed11-a3bb-000d3aaa0172'
|
|
assert resp.json['data'][0]['text'] == 'ds:User1, Leo'
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.request')
|
|
def test_matrix42_bad_rawtoken(mocked_request, app, matrix42):
|
|
endpoint = generic_endpoint_url('matrix42', 'fragment', slug=matrix42.slug)
|
|
endpoint += '/SPSUserClassBase'
|
|
params = {}
|
|
# open access
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(matrix42)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=matrix42.pk
|
|
)
|
|
|
|
# no RawToken
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content='{}', status_code=200),
|
|
]
|
|
resp = app.get(endpoint, params=params, status=200)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'
|
|
assert resp.json['err_desc'] == 'Matrix42 not returned a RawToken: {}'
|
|
|
|
# bad JSON
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content='crashme', status_code=200),
|
|
]
|
|
resp = app.get(endpoint, params=params, status=200)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'
|
|
assert 'invalid JSON' in resp.json['err_desc']
|
|
|
|
# not a dict
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content='"crashme"', status_code=200),
|
|
]
|
|
resp = app.get(endpoint, params=params, status=200)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'
|
|
assert 'not returned a dict' in resp.json['err_desc']
|
|
|
|
# empty response: error when not allowed
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content='', status_code=204),
|
|
]
|
|
resp = app.get(endpoint, params=params, status=200)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'
|
|
assert 'returned an empty response' in resp.json['err_desc']
|
|
|
|
# Matrix42 error
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content='{"ExceptionName":"NotFound","Message":"4o4"}', status_code=404),
|
|
]
|
|
resp = app.get(endpoint, params=params, status=200)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'
|
|
assert resp.json['err_desc'] == 'Matrix42 returned 404 response, ExceptionName "NotFound": 4o4'
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content=TOKEN, status_code=500),
|
|
]
|
|
resp = app.get(endpoint, params=params, status=200)
|
|
assert resp.json['err'] == 1
|
|
assert resp.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'
|
|
assert resp.json['err_desc'] == 'Matrix42 returned status code 500'
|
|
|
|
|
|
@mock.patch('passerelle.utils.Request.request')
|
|
def test_matrix42_object(mocked_request, app, matrix42):
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(matrix42)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=matrix42.pk
|
|
)
|
|
|
|
# create-object
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content=TOKEN, status_code=200),
|
|
FakedResponse(content='"424242"', status_code=200),
|
|
]
|
|
|
|
endpoint = generic_endpoint_url('matrix42', 'create-object', slug=matrix42.slug)
|
|
endpoint += '/SPSActivityTypeTicket'
|
|
payload = {
|
|
'SPSActivityClassBase/Subject': 'incident subject',
|
|
'SPSActivityClassBase/Category': 'category-id',
|
|
}
|
|
resp = app.post_json(endpoint, params=payload, status=200)
|
|
assert mocked_request.call_count == 2
|
|
get_token, post_object = mocked_request.call_args_list
|
|
assert get_token[0] == (
|
|
'POST',
|
|
'https://matrix42.example.net/api/ApiToken/GenerateAccessTokenFromApiToken',
|
|
)
|
|
assert get_token[1]['json'] == get_token[1]['params'] == None
|
|
assert get_token[1]['headers']['Authorization'] == 'Bearer token1'
|
|
assert post_object[0] == (
|
|
'POST',
|
|
'https://matrix42.example.net/api/data/objects/SPSActivityTypeTicket',
|
|
)
|
|
assert post_object[1]['json'] == {
|
|
'SPSActivityClassBase': {
|
|
'Subject': 'incident subject',
|
|
'Category': 'category-id',
|
|
}
|
|
}
|
|
assert post_object[1]['params'] is None
|
|
assert post_object[1]['headers']['Authorization'] == 'Bearer token2'
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] == '424242'
|
|
|
|
# get-object
|
|
mocked_request.reset_mock()
|
|
mocked_request.side_effect = [
|
|
FakedResponse(content=TOKEN, status_code=200),
|
|
FakedResponse(content=OBJECT, status_code=200),
|
|
]
|
|
endpoint = generic_endpoint_url('matrix42', 'get-object', slug=matrix42.slug)
|
|
endpoint += '/SPSActivityTypeTicket/424242' # ciName + id
|
|
resp = app.get(endpoint, status=200)
|
|
assert mocked_request.call_count == 2
|
|
get_token, get_object = mocked_request.call_args_list
|
|
assert get_token[0] == (
|
|
'POST',
|
|
'https://matrix42.example.net/api/ApiToken/GenerateAccessTokenFromApiToken',
|
|
)
|
|
assert get_token[1]['json'] == get_token[1]['params'] == None
|
|
assert get_token[1]['headers']['Authorization'] == 'Bearer token1'
|
|
assert get_object[0] == (
|
|
'GET',
|
|
'https://matrix42.example.net/api/data/objects/SPSActivityTypeTicket/424242',
|
|
)
|
|
assert get_object[1]['json'] == get_object[1]['params'] == None
|
|
assert get_object[1]['headers']['Authorization'] == 'Bearer token2'
|
|
assert resp.json['err'] == 0
|
|
assert resp.json['data'] == {
|
|
'ID': '424242',
|
|
'SPSActivityClassBase': {'TicketNumber': 'TCK0000153', 'TimeStamp': 'AAAAAAHlWr4='},
|
|
}
|
|
|
|
|
|
def test_matrix42_generic(app, matrix42):
|
|
api = ApiUser.objects.create(username='all', keytype='', key='')
|
|
obj_type = ContentType.objects.get_for_model(matrix42)
|
|
AccessRight.objects.create(
|
|
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=matrix42.pk
|
|
)
|
|
endpoint = generic_endpoint_url('matrix42', 'generic', slug=matrix42.slug)
|
|
|
|
with responses.RequestsMock() as rsps:
|
|
rsps.post(
|
|
'https://matrix42.example.net/api/ApiToken/GenerateAccessTokenFromApiToken',
|
|
status=200,
|
|
body=TOKEN,
|
|
)
|
|
rsps.post('https://matrix42.example.net/api/ticket/Transform', status=204)
|
|
resp = app.post_json(endpoint + '/ticket/Transform', params={'foo/bar': 'coin'}, status=200)
|
|
assert json.loads(rsps.calls[1].request.body) == {'foo': {'bar': 'coin'}}
|
|
assert resp.json == {'err': 0, 'data': None} # empty response 204 is ok here
|