277 lines
9.1 KiB
Python
277 lines
9.1 KiB
Python
import pytest
|
||
from django.contrib.contenttypes.models import ContentType
|
||
from httmock import HTTMock
|
||
|
||
import tests.utils
|
||
from passerelle.base.models import AccessRight, ApiUser
|
||
from passerelle.contrib.strasbourg_eu.models import StrasbourgEu
|
||
|
||
INTERESTS_EXAMPLE = """{
|
||
"interests": [
|
||
{
|
||
"id": "275305",
|
||
"name": "Centre d'intérêt 2",
|
||
"description": "Centre d'intérêt 2",
|
||
"typeId": "275302",
|
||
"type": "Type 1",
|
||
"categories": [
|
||
"188274"
|
||
]
|
||
},
|
||
{
|
||
"id": "275303",
|
||
"name": "Centre d'intérêt 1",
|
||
"description": "Centre d'intérêt 1",
|
||
"typeId": "275302",
|
||
"type": "Type 1",
|
||
"categories": [
|
||
"188064",
|
||
"188274"
|
||
]
|
||
}
|
||
]
|
||
}"""
|
||
|
||
USER_INTERESTS_EXAMPLE = """{
|
||
"userId": "M0cG+MtX8/Zr+zSMuI+H7yTC7SeEHT/tvYr1BsdqZic=",
|
||
"interests": [
|
||
"275303"
|
||
]
|
||
}"""
|
||
|
||
NOTIFICATIONS_EXAMPLE = """{
|
||
"notifications": [
|
||
{
|
||
"id": "279201",
|
||
"title": "Notification de type 1",
|
||
"description": "Notification de type 1",
|
||
"url": "",
|
||
"type": "Type de notification 1",
|
||
"typeId": "TYPE_1",
|
||
"isRead": false,
|
||
"publicationDate": "2020-03-17 06:22:00",
|
||
"expirationDate": "2020-04-17 06:22:00"
|
||
},
|
||
{
|
||
"id": "279801",
|
||
"title": "Ma super notif",
|
||
"description": "Ma super notif",
|
||
"url": "",
|
||
"type": "Type de notification 1",
|
||
"typeId": "TYPE_1",
|
||
"isRead": true,
|
||
"publicationDate": "2020-03-17 06:22:00",
|
||
"expirationDate": "2020-04-17 06:22:00"
|
||
},
|
||
{
|
||
"id": "279802",
|
||
"title": "Ma super notif date invalide",
|
||
"description": "Ma super notif 2",
|
||
"url": "",
|
||
"type": "Type de notification 1",
|
||
"typeId": "TYPE_1",
|
||
"isRead": true,
|
||
"publicationDate": "whatever",
|
||
"expirationDate": "2020-04-17 06:22:00"
|
||
}
|
||
]
|
||
}"""
|
||
|
||
SUCCESS_EXAMPLE = """{
|
||
"success": "MESSAGE"
|
||
}"""
|
||
|
||
ERROR_EXAMPLE = """{
|
||
"error": "MESSAGE"
|
||
}"""
|
||
|
||
UNAUTHORIZED_EXAMPLE = """{"error":"not authorized"}"""
|
||
|
||
FAVORITES_EXAMPLE = """{
|
||
"favorites": [
|
||
{
|
||
"id": "281701",
|
||
"title": "Mon événement",
|
||
"url": "http://example.net",
|
||
"entityId": "190665",
|
||
"entityTitle": "Le ballet des ombres heureuses",
|
||
"typeId": "2"
|
||
},
|
||
{
|
||
"id": "281702",
|
||
"title": "Mon lieu",
|
||
"url": "http://example.org",
|
||
"entityId": "153866",
|
||
"entityTitle": "Crèche collective et halte garderie de la Maison de l’enfance",
|
||
"typeId": "1"
|
||
}
|
||
]
|
||
}"""
|
||
|
||
FAVORITE_TYPES_EXAMPLE = """{
|
||
"types": [
|
||
{
|
||
"id": "1",
|
||
"name": "PLACE"
|
||
},
|
||
{
|
||
"id": "2",
|
||
"name": "EVENT"
|
||
}
|
||
]
|
||
}"""
|
||
|
||
|
||
def interests_mock(url, request):
|
||
if url.path.endswith('/get-interests'):
|
||
return {'content': INTERESTS_EXAMPLE, 'request': request, 'status_code': 200}
|
||
elif url.path.endswith('/get-user-interests'):
|
||
return {'content': USER_INTERESTS_EXAMPLE, 'request': request, 'status_code': 200}
|
||
elif url.path.endswith('/set-user-interests'):
|
||
return {'content': USER_INTERESTS_EXAMPLE, 'request': request, 'status_code': 200}
|
||
|
||
|
||
def notifications_mock(url, request):
|
||
return {'content': NOTIFICATIONS_EXAMPLE, 'request': request, 'status_code': 200}
|
||
|
||
|
||
def notification_add_success_mock(url, request):
|
||
return {'content': SUCCESS_EXAMPLE, 'request': request, 'status_code': 200}
|
||
|
||
|
||
def notification_add_error_mock(url, request):
|
||
return {'content': ERROR_EXAMPLE, 'request': request, 'status_code': 200}
|
||
|
||
|
||
def unauthorized_mock(url, request):
|
||
return {'content': UNAUTHORIZED_EXAMPLE, 'request': request, 'status_code': 200}
|
||
|
||
|
||
def error_500_mock(url, request):
|
||
return {'content': '{}', 'request': request, 'status_code': 500}
|
||
|
||
|
||
def favorites_mock(url, request):
|
||
if url.path.endswith('/get-user-favorites'):
|
||
return {'content': FAVORITES_EXAMPLE, 'request': request, 'status_code': 200}
|
||
elif url.path.endswith('/get-types'):
|
||
return {'content': FAVORITE_TYPES_EXAMPLE, 'request': request, 'status_code': 200}
|
||
elif url.path.endswith('/add-favorite'):
|
||
assert 'typeId=1' in request.body
|
||
return {'content': SUCCESS_EXAMPLE, 'request': request, 'status_code': 200}
|
||
elif url.path.endswith('/delete-favorite'):
|
||
assert 'favoriteId=12' in request.body
|
||
return {'content': SUCCESS_EXAMPLE, 'request': request, 'status_code': 200}
|
||
|
||
|
||
@pytest.fixture
|
||
def strasbourg_eu(db):
|
||
connector = StrasbourgEu.objects.create(slug='foobar', liferay_api_url='http://example.net/api/')
|
||
api = ApiUser.objects.create(username='all', keytype='', key='')
|
||
obj_type = ContentType.objects.get_for_model(connector)
|
||
AccessRight.objects.create(
|
||
codename='can_access', apiuser=api, resource_type=obj_type, resource_pk=connector.pk
|
||
)
|
||
return connector
|
||
|
||
|
||
def test_availability(strasbourg_eu):
|
||
with HTTMock(interests_mock):
|
||
strasbourg_eu.check_status()
|
||
|
||
|
||
def test_interests(app, strasbourg_eu):
|
||
endpoint = tests.utils.generic_endpoint_url('strasbourg-eu', 'interests', slug=strasbourg_eu.slug)
|
||
with HTTMock(interests_mock):
|
||
resp = app.get(endpoint)
|
||
assert len(resp.json['data']) == 2
|
||
assert resp.json['data'][0]['id'] == '275303'
|
||
assert resp.json['data'][0]['text'] == "Type 1 / Centre d'intérêt 1"
|
||
assert resp.json['data'][1]['id'] == '275305'
|
||
assert resp.json['data'][1]['text'] == "Type 1 / Centre d'intérêt 2"
|
||
|
||
resp = app.get(endpoint, params={'name_id': 'xxx'})
|
||
assert len(resp.json['data']) == 1
|
||
assert resp.json['data'][0]['id'] == '275303'
|
||
assert resp.json['data'][0]['text'] == "Type 1 / Centre d'intérêt 1"
|
||
|
||
resp = app.post_json(endpoint, params={})
|
||
assert resp.json['err_desc'] == 'missing name_id'
|
||
|
||
resp = app.post_json(endpoint + '?name_id=xxx', params={'interests': ['275303']})
|
||
assert len(resp.json['data']) == 1
|
||
assert resp.json['data'][0]['id'] == '275303'
|
||
|
||
with HTTMock(unauthorized_mock):
|
||
resp = app.get(endpoint).json
|
||
assert resp['err_desc'] == 'not authorized'
|
||
|
||
with HTTMock(error_500_mock):
|
||
resp = app.get(endpoint).json
|
||
assert resp['err_desc'] == 'invalid service answer'
|
||
|
||
|
||
def test_notifications(app, strasbourg_eu, caplog):
|
||
endpoint = tests.utils.generic_endpoint_url('strasbourg-eu', 'notifications', slug=strasbourg_eu.slug)
|
||
|
||
with HTTMock(notifications_mock):
|
||
resp = app.get(endpoint, status=400)
|
||
records = [
|
||
record
|
||
for record in caplog.records
|
||
if record.msg.startswith('received invalid publicationDate for notification')
|
||
]
|
||
assert len(records) == 0
|
||
resp = app.get(endpoint + '?name_id=xxx')
|
||
assert len(resp.json['notifications']) == 2
|
||
records = [
|
||
record
|
||
for record in caplog.records
|
||
if record.msg.startswith('received invalid publicationDate for notification')
|
||
]
|
||
assert len(records) == 1
|
||
|
||
with HTTMock(notification_add_success_mock):
|
||
resp = app.post_json(endpoint, params={}, status=400)
|
||
assert resp.json['err_desc'] == "missing parameters: 'name_id'."
|
||
|
||
resp = app.post_json(endpoint + '?name_id=xxx', params={'title': 'title'})
|
||
assert resp.json['err'] == 0
|
||
|
||
with HTTMock(notification_add_error_mock):
|
||
resp = app.post_json(endpoint + '?name_id=xxx', params={'title': 'title'})
|
||
assert resp.json['err'] == 1
|
||
|
||
with HTTMock(unauthorized_mock):
|
||
resp = app.get(endpoint + '?name_id=xxx').json
|
||
assert resp['err_desc'] == 'not authorized'
|
||
|
||
with HTTMock(error_500_mock):
|
||
resp = app.get(endpoint + '?name_id=xxx').json
|
||
assert resp['err_desc'] == 'invalid service answer'
|
||
|
||
|
||
def test_favorites(app, strasbourg_eu):
|
||
endpoint = tests.utils.generic_endpoint_url('strasbourg-eu', 'favorites', slug=strasbourg_eu.slug)
|
||
with HTTMock(favorites_mock):
|
||
resp = app.get(endpoint, status=400)
|
||
resp = app.get(endpoint + '?name_id=xxx')
|
||
assert len(resp.json['favorites']) == 2
|
||
# URL filter
|
||
resp = app.get(endpoint + '?name_id=xxx&url=http://www.example.net')
|
||
assert len(resp.json['favorites']) == 2
|
||
|
||
resp = app.post_json(endpoint + '?name_id=xxx', params={'title': 'title', 'type': 'PLACE'})
|
||
assert resp.json['err'] == 0
|
||
|
||
resp = app.post_json(endpoint + '/12/delete?name_id=xxx', params={})
|
||
assert resp.json['err'] == 0
|
||
|
||
with HTTMock(unauthorized_mock):
|
||
resp = app.get(endpoint + '?name_id=xxx').json
|
||
assert resp['err_desc'] == 'not authorized'
|
||
|
||
with HTTMock(error_500_mock):
|
||
resp = app.get(endpoint + '?name_id=xxx').json
|
||
assert resp['err_desc'] == 'invalid service answer'
|