686 lines
27 KiB
Python
686 lines
27 KiB
Python
# passerelle - uniform access to multiple data sources and services
|
|
# Copyright (C) 202 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 json
|
|
import os
|
|
|
|
import pytest
|
|
from requests.exceptions import ConnectionError, HTTPError
|
|
|
|
import tests.utils
|
|
from passerelle.apps.plone_restapi.models import PloneRestApi, Query
|
|
from passerelle.utils import import_site
|
|
from passerelle.utils.jsonresponse import APIError
|
|
from tests.test_manager import login
|
|
|
|
pytestmark = pytest.mark.django_db
|
|
|
|
TEST_BASE_DIR = os.path.join(os.path.dirname(__file__), 'data', 'plone_restapi')
|
|
|
|
TOKEN_RESPONSE = {
|
|
'access_token': 'd319258e-48b9-4853-88e8-7a2ad6883c7f',
|
|
'token_type': 'Bearer',
|
|
'expires_in': 28800,
|
|
'id_token': 'acd...def',
|
|
}
|
|
|
|
TOKEN_ERROR_RESPONSE = {
|
|
'error': 'access_denied',
|
|
'error_description': "Mauvaises informations de connexion de l'utilisateur",
|
|
}
|
|
|
|
|
|
def json_get_data(filename):
|
|
with open(os.path.join(TEST_BASE_DIR, '%s.json' % filename)) as fd:
|
|
return json.load(fd)
|
|
|
|
|
|
@pytest.fixture
|
|
def connector():
|
|
return tests.utils.setup_access_rights(
|
|
PloneRestApi.objects.create(
|
|
slug='my_connector',
|
|
service_url='http://www.example.net',
|
|
token_ws_url='http://www.example.net/idp/oidc/token/',
|
|
client_id='aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee',
|
|
client_secret='11111111-2222-3333-4444-555555555555',
|
|
username='jdoe',
|
|
password='secret',
|
|
)
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def token(connector):
|
|
with tests.utils.mock_url(url=connector.token_ws_url, response=TOKEN_RESPONSE) as mocked:
|
|
yield mocked
|
|
|
|
|
|
@pytest.fixture
|
|
def query(connector):
|
|
return Query.objects.create(
|
|
resource=connector,
|
|
name='demo query',
|
|
slug='my_query',
|
|
description="Annuaire de Braine-l'Alleud",
|
|
uri='braine-l-alleud',
|
|
text_template='{{ title }} ({{ PLONE_type }})',
|
|
filter_expression='''
|
|
portal_type=Document
|
|
review_state=published
|
|
''',
|
|
sort='UID',
|
|
order=False,
|
|
limit=3,
|
|
)
|
|
|
|
|
|
def test_views(db, admin_user, app, connector):
|
|
app = login(app)
|
|
resp = app.get('/plone-restapi/my_connector/', status=200)
|
|
resp = resp.click('New Query')
|
|
resp.form['name'] = 'my query'
|
|
resp.form['slug'] = 'my-query'
|
|
resp.form['uri'] = 'my-uri'
|
|
resp = resp.form.submit()
|
|
resp = resp.follow()
|
|
assert resp.html.find('div', {'id': 'panel-queries'}).ul.li.a.text == 'my query'
|
|
|
|
|
|
def test_views_query_unicity(admin_user, app, connector, query):
|
|
connector2 = PloneRestApi.objects.create(
|
|
slug='my_connector2',
|
|
)
|
|
Query.objects.create(
|
|
resource=connector2,
|
|
slug='foo-bar',
|
|
name='Foo Bar',
|
|
)
|
|
|
|
# create
|
|
app = login(app)
|
|
resp = app.get('/manage/plone-restapi/%s/query/new/' % connector.slug)
|
|
resp.form['slug'] = query.slug
|
|
resp.form['name'] = 'Foo Bar'
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 200
|
|
assert 'A query with this slug already exists' in resp.text
|
|
assert Query.objects.filter(resource=connector).count() == 1
|
|
resp.form['slug'] = 'foo-bar'
|
|
resp.form['name'] = query.name
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 200
|
|
assert 'A query with this name already exists' in resp.text
|
|
assert Query.objects.filter(resource=connector).count() == 1
|
|
resp.form['slug'] = 'foo-bar'
|
|
resp.form['name'] = 'Foo Bar'
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 302
|
|
assert Query.objects.filter(resource=connector).count() == 2
|
|
new_query = Query.objects.latest('pk')
|
|
assert new_query.resource == connector
|
|
assert new_query.slug == 'foo-bar'
|
|
assert new_query.name == 'Foo Bar'
|
|
|
|
# update
|
|
resp = app.get('/manage/plone-restapi/%s/query/%s/' % (connector.slug, new_query.pk))
|
|
resp.form['slug'] = query.slug
|
|
resp.form['name'] = 'Foo Bar'
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 200
|
|
assert 'A query with this slug already exists' in resp.text
|
|
resp.form['slug'] = 'foo-bar'
|
|
resp.form['name'] = query.name
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 200
|
|
assert 'A query with this name already exists' in resp.text
|
|
resp.form['slug'] = 'foo-bar'
|
|
resp.form['name'] = 'Foo Bar'
|
|
resp.form['uri'] = 'fr'
|
|
resp = resp.form.submit()
|
|
assert resp.status_code == 302
|
|
query = Query.objects.get(resource=connector, slug='foo-bar')
|
|
assert query.uri == 'fr'
|
|
|
|
|
|
def test_export_import(query):
|
|
assert PloneRestApi.objects.count() == 1
|
|
assert Query.objects.count() == 1
|
|
serialization = {'resources': [query.resource.export_json()]}
|
|
PloneRestApi.objects.all().delete()
|
|
assert PloneRestApi.objects.count() == 0
|
|
assert Query.objects.count() == 0
|
|
import_site(serialization)
|
|
assert PloneRestApi.objects.count() == 1
|
|
assert str(PloneRestApi.objects.get().client_id) == 'aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee'
|
|
assert Query.objects.count() == 1
|
|
|
|
|
|
def test_adapt_id_and_type_plone_attributes(connector):
|
|
plone_response = {
|
|
'@type': '@value',
|
|
'@dict': {
|
|
'@array': [
|
|
{
|
|
'@id': '123',
|
|
'@type': '@value',
|
|
}
|
|
]
|
|
},
|
|
}
|
|
connector.adapt_id_and_type_plone_attributes(plone_response)
|
|
assert plone_response == {
|
|
'PLONE_type': '@value',
|
|
'@dict': {'@array': [{'PLONE_id': '123', 'PLONE_type': '@value'}]},
|
|
}
|
|
|
|
|
|
def test_adapt_record(connector, token):
|
|
record = {
|
|
'@id': 'plone id',
|
|
'UID': 'plone uid',
|
|
'id': 'foo',
|
|
'text': 'bar',
|
|
}
|
|
template = '{{ PLONE_id }}, {{ id }}, {{original_id }}, {{ original_text }}'
|
|
connector.adapt_record(record, template)
|
|
assert record == {
|
|
'PLONE_id': 'plone id',
|
|
'UID': 'plone uid',
|
|
'id': 'plone uid',
|
|
'text': 'plone id, plone uid, foo, bar',
|
|
'original_id': 'foo',
|
|
'original_text': 'bar',
|
|
}
|
|
|
|
|
|
def test_get_token(app, connector):
|
|
with pytest.raises(APIError):
|
|
with tests.utils.mock_url(url=connector.token_ws_url, response=TOKEN_ERROR_RESPONSE, status_code=404):
|
|
connector.get_token()
|
|
with tests.utils.mock_url(url=connector.token_ws_url, response=TOKEN_RESPONSE) as mocked:
|
|
result = connector.get_token()
|
|
assert mocked.handlers[0].call['count'] == 1
|
|
assert 'secret' in mocked.handlers[0].call['requests'][0].body
|
|
assert result == 'acd...def'
|
|
|
|
# make sure the token from cache is used
|
|
connector.get_token()
|
|
assert mocked.handlers[0].call['count'] == 1
|
|
connector.get_token(True)
|
|
assert mocked.handlers[0].call['count'] == 2
|
|
|
|
|
|
def test_check_status(app, connector):
|
|
url = connector.service_url + '/ok'
|
|
with tests.utils.mock_url(url=url, response='OK'):
|
|
connector.check_status()
|
|
|
|
# plone not responding
|
|
with tests.utils.mock_url(url=url, response={}, status_code=500):
|
|
with pytest.raises(HTTPError):
|
|
connector.check_status()
|
|
with tests.utils.mock_url(url=url, exception=ConnectionError('plop')):
|
|
with pytest.raises(ConnectionError):
|
|
connector.check_status()
|
|
|
|
|
|
def test_fetch(app, connector, token):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'fetch', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/fetch'
|
|
url = connector.service_url + '/braine-l-alleud/dccd85d12cf54b6899dff41e5a56ee7f'
|
|
params = {
|
|
'uid': 'dccd85d12cf54b6899dff41e5a56ee7f',
|
|
'uri': 'braine-l-alleud',
|
|
'text_template': '{{ title }} ({{ topics.0.title }})',
|
|
}
|
|
with tests.utils.mock_url(url=url, response=json_get_data('fetch')):
|
|
resp = app.get(endpoint, params=params)
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['id'] == 'dccd85d12cf54b6899dff41e5a56ee7f'
|
|
assert resp.json['data']['text'] == 'Le Prisme (Activités et divertissement)'
|
|
assert token.handlers[0].call['count'] == 1
|
|
|
|
|
|
def test_basic_auth(app, connector, token):
|
|
connector.token_ws_url = ''
|
|
connector.basic_auth_username = 'jsmith'
|
|
connector.basic_auth_password = 'secret2'
|
|
connector.save()
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'fetch', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/fetch'
|
|
url = connector.service_url + '/braine-l-alleud/dccd85d12cf54b6899dff41e5a56ee7f'
|
|
params = {
|
|
'uid': 'dccd85d12cf54b6899dff41e5a56ee7f',
|
|
'uri': 'braine-l-alleud',
|
|
'text_template': '{{ title }} ({{ topics.0.title }})',
|
|
}
|
|
with tests.utils.mock_url(url=url, response=json_get_data('fetch')) as mocked_get:
|
|
resp = app.get(endpoint, params=params)
|
|
assert mocked_get.handlers[0].call['count'] == 1
|
|
assert 'Basic' in mocked_get.handlers[0].call['requests'][0].headers['Authorization']
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['id'] == 'dccd85d12cf54b6899dff41e5a56ee7f'
|
|
assert resp.json['data']['text'] == 'Le Prisme (Activités et divertissement)'
|
|
assert token.handlers[0].call['count'] == 0
|
|
|
|
|
|
def test_request_anonymously(app, connector, token):
|
|
connector.token_ws_url = ''
|
|
connector.save()
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'fetch', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/fetch'
|
|
url = connector.service_url + '/braine-l-alleud/dccd85d12cf54b6899dff41e5a56ee7f'
|
|
params = {
|
|
'uid': 'dccd85d12cf54b6899dff41e5a56ee7f',
|
|
'uri': 'braine-l-alleud',
|
|
'text_template': '{{ title }} ({{ topics.0.title }})',
|
|
}
|
|
with tests.utils.mock_url(url=url, response=json_get_data('fetch')):
|
|
resp = app.get(endpoint, params=params)
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['id'] == 'dccd85d12cf54b6899dff41e5a56ee7f'
|
|
assert resp.json['data']['text'] == 'Le Prisme (Activités et divertissement)'
|
|
assert token.handlers[0].call['count'] == 0
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'exception, status_code, response, err_desc',
|
|
[
|
|
[ConnectionError('plop'), None, None, 'plop'],
|
|
[None, 200, 'not json', 'bad JSON response'],
|
|
[None, 404, {'message': 'Resource not found: ...', 'type': 'NotFound'}, '404 Client Error'],
|
|
],
|
|
)
|
|
def test_request_error(app, connector, token, exception, status_code, response, err_desc):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'fetch', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/fetch'
|
|
url = connector.service_url + '/braine-l-alleud/plop'
|
|
params = {
|
|
'uid': 'plop',
|
|
'uri': 'braine-l-alleud',
|
|
'text_template': '{{ title }} ({{ PLONE_type }})',
|
|
}
|
|
with tests.utils.mock_url(url=url, response=response, status_code=status_code, exception=exception):
|
|
resp = app.get(endpoint, params=params)
|
|
assert resp.json['err']
|
|
assert err_desc in resp.json['err_desc']
|
|
|
|
|
|
def test_get_content_types(app, connector, token):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'get_content_types', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/get_content_types'
|
|
url = connector.service_url + '//@types'
|
|
with tests.utils.mock_url(url=url, response=json_get_data('get_content_types')):
|
|
resp = app.get(endpoint)
|
|
assert not resp.json['err']
|
|
assert len(resp.json['data']) == 10
|
|
assert resp.json['data'][2]['id'] == 'imio.directory.Contact'
|
|
assert resp.json['data'][2]['text'] == 'Contact'
|
|
|
|
url = connector.service_url + '/belleville/citoyens/@types'
|
|
with tests.utils.mock_url(url=url, response=json_get_data('get_content_types')):
|
|
resp = app.get(endpoint + '?uri=belleville/citoyens')
|
|
assert not resp.json['err']
|
|
|
|
|
|
def test_get_content_type(app, connector, token):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'get_content_type', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/get_content_type'
|
|
url = connector.service_url + '/@types/imio.directory.Contact'
|
|
params = {'id': 'imio.directory.Contact'}
|
|
with tests.utils.mock_url(url=url, response=json_get_data('get_content_type')):
|
|
resp = app.get(endpoint, params=params)
|
|
assert not resp.json['err']
|
|
assert resp.json['data']['title'] == 'Contact'
|
|
assert resp.json['data']['required'] == ['title', 'type']
|
|
assert len(resp.json['data']['properties']) == 28
|
|
assert (
|
|
resp.json['data']['properties']['topics']['items']['vocabulary']['@id']
|
|
== 'https://annuaire.preprod.imio.be/@vocabularies/imio.smartweb.vocabulary.Topics'
|
|
)
|
|
|
|
|
|
def test_get_field_choices(app, connector, token):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'get_field_choices', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/get_field_choices'
|
|
url = connector.service_url + '/@vocabularies/imio.smartweb.vocabulary.Topics'
|
|
params = {'id': 'imio.smartweb.vocabulary.Topics'}
|
|
with tests.utils.mock_url(url=url, response=json_get_data('get_field_choices')) as mocked_get:
|
|
resp = app.get(endpoint, params=params)
|
|
assert 'b_size=999999' in mocked_get.handlers[0].call['requests'][1].url
|
|
assert not resp.json['err']
|
|
assert len(resp.json['data']) == 17
|
|
assert resp.json['data'][16]['id'] == 'tourism'
|
|
assert resp.json['data'][16]['text'] == 'Tourisme'
|
|
|
|
|
|
def test_create(app, connector, token):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'create', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/create'
|
|
url = connector.service_url + '/braine-l-alleud'
|
|
payload = {
|
|
'@type': 'imio.directory.Contact',
|
|
'title': "Test Entr'ouvert",
|
|
'type': 'organization',
|
|
'schedule': {},
|
|
'topics/0/title': 'Tourisme',
|
|
'topics/0/token': 'tourism',
|
|
'image': {'filename': 'foo.jpg', 'content_type': 'image/jpeg', 'content': '...'},
|
|
'phones': [
|
|
{'label': 'numéro principal', 'number': '0123456789', 'type': 'work'},
|
|
{'label': '', 'number': '', 'type': ''},
|
|
],
|
|
}
|
|
with tests.utils.mock_url(url=url, response=json_get_data('fetch'), status_code=201) as mocked:
|
|
resp = app.post_json(endpoint + '?uri=braine-l-alleud&publish=false', params=payload)
|
|
body = json.loads(mocked.handlers[0].call['requests'][1].body)
|
|
assert body['topics'] == [{'title': 'Tourisme', 'token': 'tourism'}]
|
|
assert body['image'] == {
|
|
'filename': 'foo.jpg',
|
|
'content_type': 'image/jpeg',
|
|
'encoding': 'base64',
|
|
'data': '...',
|
|
'content-type': 'image/jpeg',
|
|
}
|
|
assert body['phones'] == [{'label': 'numéro principal', 'number': '0123456789', 'type': 'work'}]
|
|
assert not resp.json['err']
|
|
assert resp.json['data'] == {
|
|
'uid': 'dccd85d12cf54b6899dff41e5a56ee7f',
|
|
'created': True,
|
|
'review_state': None,
|
|
}
|
|
|
|
|
|
def test_create_and_publish(app, connector, token):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'create', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/create'
|
|
url = connector.service_url + '/braine-l-alleud'
|
|
payload = {
|
|
'@type': 'imio.directory.Contact',
|
|
'title': "Test Entr'ouvert",
|
|
'type': 'organization',
|
|
'schedule': {},
|
|
'topics/0/title': 'Tourisme',
|
|
'topics/0/token': 'tourism',
|
|
'image': {'filename': 'foo.jpg', 'content_type': 'image/jpeg', 'content': '...'},
|
|
}
|
|
publish_url = url + '/%s/@workflow/publish' % 'dccd85d12cf54b6899dff41e5a56ee7f'
|
|
with tests.utils.mock_url(url=url, response=json_get_data('fetch'), status_code=201) as mocked:
|
|
with tests.utils.mock_url(
|
|
url=publish_url, response=json_get_data('workflow_publish'), status_code=200
|
|
):
|
|
resp = app.post_json(endpoint + '?uri=braine-l-alleud&publish=true', params=payload)
|
|
body = json.loads(mocked.handlers[0].call['requests'][1].body)
|
|
assert body['topics'] == [{'title': 'Tourisme', 'token': 'tourism'}]
|
|
assert body['image'] == {
|
|
'filename': 'foo.jpg',
|
|
'content_type': 'image/jpeg',
|
|
'encoding': 'base64',
|
|
'data': '...',
|
|
'content-type': 'image/jpeg',
|
|
}
|
|
assert not resp.json['err']
|
|
assert resp.json['data'] == {
|
|
'uid': 'dccd85d12cf54b6899dff41e5a56ee7f',
|
|
'created': True,
|
|
'review_state': 'published',
|
|
}
|
|
|
|
|
|
def test_create_wrong_payload(app, connector, token):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'create', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/create'
|
|
payload = 'not json'
|
|
resp = app.post(endpoint + '?uri=braine-l-alleud', params=payload, status=400)
|
|
assert resp.json['err']
|
|
assert resp.json['err_desc'] == 'Expecting value: line 1 column 1 (char 0)'
|
|
assert resp.json['err_class'] == 'passerelle.apps.plone_restapi.models.ParameterTypeError'
|
|
|
|
|
|
def test_update(app, connector, token):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'update', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/update'
|
|
url = connector.service_url + '/braine-l-alleud/dccd85d12cf54b6899dff41e5a56ee7f'
|
|
query_string = '?uri=braine-l-alleud&uid=dccd85d12cf54b6899dff41e5a56ee7f'
|
|
payload = {
|
|
'title': 'Test update',
|
|
'topics/0/token': 'social',
|
|
'image': {'filename': 'foo.jpg', 'content_type': 'image/jpeg', 'content': '...'},
|
|
'taxonomy_contact_category': ['tfixs8rmb2', 'lugpblqfj3'],
|
|
'phones': [
|
|
{'label': '', 'number': '', 'type': ''},
|
|
{'label': '', 'number': '', 'type': ''},
|
|
],
|
|
}
|
|
with tests.utils.mock_url(url=url, response='', status_code=204) as mocked:
|
|
resp = app.post_json(endpoint + query_string, params=payload)
|
|
body = json.loads(mocked.handlers[0].call['requests'][1].body)
|
|
assert body['topics'] == [{'token': 'social'}]
|
|
assert body['taxonomy_contact_category'] == ['tfixs8rmb2', 'lugpblqfj3']
|
|
assert body['image'] == {
|
|
'filename': 'foo.jpg',
|
|
'content_type': 'image/jpeg',
|
|
'encoding': 'base64',
|
|
'data': '...',
|
|
'content-type': 'image/jpeg',
|
|
}
|
|
assert body['phones'] == []
|
|
|
|
assert not resp.json['err']
|
|
assert resp.json['data'] == {'uid': 'dccd85d12cf54b6899dff41e5a56ee7f', 'updated': True}
|
|
|
|
|
|
def test_update_wrong_payload(app, connector, token):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'update', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/update'
|
|
query_string = '?uri=braine-l-alleud&uid=dccd85d12cf54b6899dff41e5a56ee7f'
|
|
payload = 'not json'
|
|
resp = app.post(endpoint + query_string, params=payload, status=400)
|
|
assert resp.json['err']
|
|
assert resp.json['err_desc'] == 'Expecting value: line 1 column 1 (char 0)'
|
|
assert resp.json['err_class'] == 'passerelle.apps.plone_restapi.models.ParameterTypeError'
|
|
|
|
|
|
def test_remove(app, connector, token):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'remove', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/remove'
|
|
url = connector.service_url + '/braine-l-alleud/dccd85d12cf54b6899dff41e5a56ee7f'
|
|
query_string = '?uri=braine-l-alleud&uid=dccd85d12cf54b6899dff41e5a56ee7f'
|
|
with tests.utils.mock_url(url=url, response='', status_code=204):
|
|
resp = app.delete(endpoint + query_string)
|
|
assert resp.json['data'] == {'uid': 'dccd85d12cf54b6899dff41e5a56ee7f', 'removed': True}
|
|
assert not resp.json['err']
|
|
|
|
|
|
def test_search(app, connector, token):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'search', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/search'
|
|
url = connector.service_url + '/braine-l-alleud/@search'
|
|
params = {
|
|
'uri': 'braine-l-alleud',
|
|
'text_template': '{{ title }} ({{ PLONE_type }})',
|
|
'sort': 'UID',
|
|
'order': False,
|
|
'limit': 3,
|
|
}
|
|
qs = {}
|
|
with tests.utils.mock_url(url=url, response=json_get_data('q_search'), qs=qs):
|
|
resp = app.get(endpoint, params=params)
|
|
assert token.handlers[0].call['count'] == 1
|
|
assert qs == {'sort_on': 'UID', 'sort_order': 'descending', 'b_size': '3', 'fullobjects': 'y'}
|
|
assert not resp.json['err']
|
|
assert len(resp.json['data']) == 3
|
|
assert [(x['id'], x['text']) for x in resp.json['data']] == [
|
|
(
|
|
'dea9d26baab944beb7e54d4024d35a33',
|
|
"Cabinet du Bourgmestre de la Commune de Braine-l'Alleud (imio.directory.Contact)",
|
|
),
|
|
(
|
|
'23a32197d6c841259963b43b24747854',
|
|
"Académie de Musique de Braine-l'Alleud (imio.directory.Contact)",
|
|
),
|
|
(
|
|
'f82d2c079131433ea6ab20f9f7f49442',
|
|
'Accueil et Orientation Volontariat (A.O.V.) (imio.directory.Contact)',
|
|
),
|
|
]
|
|
|
|
|
|
def test_search_using_q(app, connector, token):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'search', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/search'
|
|
url = connector.service_url + '/braine-l-alleud/@search'
|
|
params = {
|
|
'uri': 'braine-l-alleud',
|
|
'text_template': '{{ title }} ({{ PLONE_type }})',
|
|
'sort': 'title',
|
|
'order': True,
|
|
'limit': '3',
|
|
'q': 'Página dentro',
|
|
}
|
|
qs = {}
|
|
with tests.utils.mock_url(url=url, response=json_get_data('q_search'), qs=qs):
|
|
resp = app.get(endpoint, params=params)
|
|
assert qs == {
|
|
'SearchableText': 'Página dentro',
|
|
'sort_on': 'title',
|
|
'sort_order': 'ascending',
|
|
'b_size': '3',
|
|
'fullobjects': 'y',
|
|
}
|
|
assert not resp.json['err']
|
|
assert len(resp.json['data']) == 3
|
|
assert [(x['id'], x['text']) for x in resp.json['data']] == [
|
|
(
|
|
'dea9d26baab944beb7e54d4024d35a33',
|
|
"Cabinet du Bourgmestre de la Commune de Braine-l'Alleud (imio.directory.Contact)",
|
|
),
|
|
(
|
|
'23a32197d6c841259963b43b24747854',
|
|
"Académie de Musique de Braine-l'Alleud (imio.directory.Contact)",
|
|
),
|
|
(
|
|
'f82d2c079131433ea6ab20f9f7f49442',
|
|
'Accueil et Orientation Volontariat (A.O.V.) (imio.directory.Contact)',
|
|
),
|
|
]
|
|
|
|
|
|
def test_search_using_id(app, connector, token):
|
|
endpoint = tests.utils.generic_endpoint_url('plone-restapi', 'search', slug=connector.slug)
|
|
assert endpoint == '/plone-restapi/my_connector/search'
|
|
url = connector.service_url + '/braine-l-alleud/@search'
|
|
params = {
|
|
'uri': 'braine-l-alleud',
|
|
'text_template': '{{ title }} ({{ PLONE_type }})',
|
|
'id': '9fbb2afd499e465983434f974fce8404',
|
|
}
|
|
qs = {}
|
|
with tests.utils.mock_url(url=url, response=json_get_data('id_search'), qs=qs):
|
|
resp = app.get(endpoint, params=params)
|
|
assert qs == {'UID': '9fbb2afd499e465983434f974fce8404', 'fullobjects': 'y'}
|
|
assert len(resp.json['data']) == 1
|
|
assert resp.json['data'][0]['text'] == "Académie de Musique de Braine-l'Alleud (imio.directory.Contact)"
|
|
|
|
|
|
def test_query_q(app, query, token):
|
|
endpoint = '/plone-restapi/my_connector/q/my_query/'
|
|
url = query.resource.service_url + '/braine-l-alleud/@search'
|
|
params = {
|
|
'limit': 3,
|
|
}
|
|
qs = {}
|
|
with tests.utils.mock_url(url=url, response=json_get_data('q_search'), qs=qs):
|
|
resp = app.get(endpoint, params=params)
|
|
assert qs == {
|
|
'sort_on': 'UID',
|
|
'sort_order': 'descending',
|
|
'b_size': '3',
|
|
'portal_type': 'Document',
|
|
'review_state': 'published',
|
|
'fullobjects': 'y',
|
|
}
|
|
assert not resp.json['err']
|
|
assert len(resp.json['data']) == 3
|
|
assert resp.json['meta'] == {'label': 'demo query', 'description': "Annuaire de Braine-l'Alleud"}
|
|
|
|
|
|
def test_query_q_using_q(app, query, token):
|
|
endpoint = '/plone-restapi/my_connector/q/my_query/'
|
|
url = query.resource.service_url + '/braine-l-alleud/@search'
|
|
params = {
|
|
'q': 'Página dentro',
|
|
}
|
|
qs = {}
|
|
with tests.utils.mock_url(url=url, response=json_get_data('q_search'), qs=qs):
|
|
resp = app.get(endpoint, params=params)
|
|
assert qs == {
|
|
'SearchableText': 'Página dentro',
|
|
'sort_on': 'UID',
|
|
'sort_order': 'descending',
|
|
'b_size': '3',
|
|
'portal_type': 'Document',
|
|
'review_state': 'published',
|
|
'fullobjects': 'y',
|
|
}
|
|
assert not resp.json['err']
|
|
assert len(resp.json['data']) == 3
|
|
assert resp.json['meta'] == {'label': 'demo query', 'description': "Annuaire de Braine-l'Alleud"}
|
|
|
|
|
|
def test_query_q_using_id(app, query, token):
|
|
endpoint = '/plone-restapi/my_connector/q/my_query/'
|
|
url = query.resource.service_url + '/braine-l-alleud/@search'
|
|
params = {
|
|
'id': '9fbb2afd499e465983434f974fce8404',
|
|
}
|
|
qs = {}
|
|
with tests.utils.mock_url(url=url, response=json_get_data('id_search'), qs=qs):
|
|
resp = app.get(endpoint, params=params)
|
|
assert qs == {
|
|
'UID': '9fbb2afd499e465983434f974fce8404',
|
|
'fullobjects': 'y',
|
|
}
|
|
assert len(resp.json['data']) == 1
|
|
assert resp.json['data'][0]['text'] == "Académie de Musique de Braine-l'Alleud (imio.directory.Contact)"
|
|
assert resp.json['meta'] == {'label': 'demo query', 'description': "Annuaire de Braine-l'Alleud"}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'in_data, out_data',
|
|
[
|
|
[
|
|
{
|
|
'phones': [
|
|
{'label': 'numéro principal', 'number': '0123456789', 'type': 'work'},
|
|
{'label': '', 'number': '', 'type': ''},
|
|
]
|
|
},
|
|
{'phones': [{'label': 'numéro principal', 'number': '0123456789', 'type': 'work'}]},
|
|
],
|
|
[{'taxonomy_contact_category': ['tfixs8rmb2', 'lugpblqfj3']}, 'same'],
|
|
[
|
|
{'phones': [{'label': '', 'number': '', 'type': ''}, ['tfixs8rmb2', 'lugpblqfj3']]},
|
|
{'phones': [['tfixs8rmb2', 'lugpblqfj3']]},
|
|
],
|
|
],
|
|
)
|
|
def test_remove_unvaluated_dict_from_list(connector, in_data, out_data):
|
|
if out_data == 'same':
|
|
out_data = in_data
|
|
assert connector.remove_unvaluated_dict_from_list(in_data) == out_data
|