entrouvert
/
passerelle
15
0
Fork 0
Code Issues Pull Requests 14 Releases Activity
passerelle/tests/test_mdph13.py

507 lines
21 KiB
Python
Raw Blame History

# passerelle - uniform access to multiple data sources and services
# Copyright (C) 2018 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 base64
import datetime
import json
import logging
import httmock
import pytest
import requests
import requests.exceptions
import tests.utils
from passerelle.contrib.mdph13.models import Link, MDPH13Resource
from passerelle.utils.jsonresponse import APIError
NAME_ID = 'xyz'
FILE_NUMBER = '1234'
DOB = datetime.date(1993, 5, 4)
DOB_ISOFORMAT = '1993-05-04'
EMAIL = 'john.doe@example.com'
SECRET = 'secret'
IP = '88.34.56.56'
VALID_RESPONSE = json.dumps(
{
'err': 0,
'data': {
'numero': FILE_NUMBER,
'beneficiaire': {
'nom': 'Martini',
'prenom': 'Alfonso',
'tel_mobile': '06 01 02 03 04',
'tel_fixe': '04.01.02.03.04',
'date_de_naissance': '1951-03-23',
'email': 'martini.a@free.fr',
'entourage': [
{
'role': 'Père',
'nom': 'DUPONT Henri',
'tel_mobile': '0123232323',
'tel_fixe': '0202020202',
'email': 'henri.dupont@xyz.com',
},
{
'role': 'Mère',
'nom': 'DUPONT Marie',
'tel_mobile': '0123232323',
'tel_fixe': '0202020202',
'email': 'marie.dupont@xyz.com',
},
{
'role': 'Aidant',
'nom': 'ROBERT Fanny',
'tel_mobile': '0123232323',
'tel_fixe': '0202020202',
'email': 'frobert@xyz.com',
},
],
'adresse': {
'adresse_2': 'Bliblibli',
'adresse_3': 'Bliblibli',
'adresse_4': 'CHEMIN DE LA CARRAIRE',
'adresse_5': 'Bliblibli',
'code_postal': '13500',
'ville': 'MARTIGUES',
},
'incapacite': {'taux': 'Taux >=80%', 'date_fin_effet': '2019-06-30'},
},
'demandes': [
{
'numero': '1544740',
'date_demande': '2015-11-26',
'type_demande': 'Renouvellement',
'prestation': "Carte d'invalidité (de priorité) pour personne handicapée",
'statut': 'Instruction administrative terminée en attente de passage en évaluation',
'typologie': 'Demande En Cours',
'date_decision': None,
},
{
'numero': '1210524',
'date_demande': '2014-06-13',
'type_demande': 'Renouvellement',
'prestation': "Carte d'invalidité (de priorité) pour personne handicapée",
'statut': 'Décision prononcée et expédition réalisée (traitement terminé)',
'typologie': 'Traitée non expédiée',
'date_decision': '2014-07-10',
'date_debut_effet': '2014-08-01',
'date_fin_effet': '2016-05-01',
},
{
'numero': '1231345',
'date_demande': '2014-07-22',
'type_demande': 'Recours Gracieux',
'prestation': "Carte d'invalidité (de priorité) pour personne handicapée",
'statut': 'Décision prononcée et expédition réalisée (traitement terminé)',
'typologie': 'Traitée et expédiée',
'date_decision': '2014-09-17',
'date_debut_effet': '2014-08-01',
'date_fin_effet': '2016-05-01',
},
{
'numero': '666660',
'date_demande': '2012-08-13',
'type_demande': 'Recours Gracieux',
'prestation': "Carte d'invalidité (de priorité) pour personne handicapée",
'statut': 'Décision prononcée et expédition réalisée (traitement terminé)',
'typologie': 'Traitée et expédiée',
'date_decision': '2012-09-26',
'date_debut_effet': '2012-07-19',
'date_fin_effet': '2014-08-01',
},
{
'numero': '605280',
'date_demande': '2012-04-05',
'type_demande': '1ère demande',
'prestation': "Carte d'invalidité (de priorité) pour personne handicapée",
'statut': 'Décision prononcée et expédition réalisée (traitement terminé)',
'typologie': 'Traitée et expédiée',
'date_decision': '2012-07-19',
'date_debut_effet': '2012-07-19',
'date_fin_effet': '2014-05-01',
},
{
'numero': '1544741',
'date_demande': '2015-11-26',
'type_demande': 'Renouvellement',
'prestation': "Carte d'invalidité (de priorité) pour personne handicapée",
'statut': 'Décision prononcée et expédition réalisée (traitement terminé)',
'typologie': 'Traitée et expédiée',
'date_decision': '2015-12-22',
'date_debut_effet': '2016-05-01',
'date_fin_effet': '2026-05-01',
},
{
'numero': '1210526',
'date_demande': '2014-06-13',
'type_demande': 'Renouvellement',
'prestation': 'Carte européenne de Stationnement',
'statut': 'Décision prononcée et expédition réalisée (traitement terminé)',
'typologie': 'Traitée et expédiée',
'date_decision': '2014-07-04',
'date_debut_effet': '2014-05-01',
'date_fin_effet': '2015-05-01',
},
{
'numero': '605281',
'date_demande': '2012-04-05',
'type_demande': '1ère demande',
'prestation': 'Carte européenne de Stationnement',
'statut': 'Décision prononcée et expédition réalisée (traitement terminé)',
'typologie': 'Traitée et expédiée',
'date_decision': '2012-07-04',
'date_debut_effet': '2012-05-01',
'date_fin_effet': '2014-05-01',
},
],
},
}
)
DOSSIER_INCONNU = {
'status_code': 404,
'content': json.dumps(
{
'err': 1,
'err_code': 'dossier-inconnu',
}
),
}
SECRET_INVALIDE = {
'status_code': 404,
'content': json.dumps(
{
'err': 1,
'err_code': 'secret-invalide',
}
),
}
@pytest.fixture
def mdph13(db):
return tests.utils.make_resource(
MDPH13Resource,
title='Test 1',
slug='test1',
description='Connecteur de test',
webservice_base_url='http://cd13.fr/',
)
@pytest.fixture
def mock_http():
class MockHttp:
def __init__(self):
self.requests = []
self.responses = []
def add_response(self, response):
self.responses.append(response)
@property
def last_request(self):
return self.requests[-1]
def request_handler(self, url, request):
idx = len(self.requests)
self.requests.append(request)
response = self.responses[idx]
if isinstance(response, Exception):
raise response
if hasattr(response, '__call__'):
response = response(url, request)
return response
mock_http = MockHttp()
with httmock.HTTMock(httmock.urlmatch()(mock_http.request_handler)):
yield mock_http
def test_situation_dossier_url(mdph13):
assert mdph13.situation_dossier_url(1234) == 'http://cd13.fr/situation/dossier/1234'
def test_call_situation_dossier(mdph13, mock_http):
mock_http.add_response(VALID_RESPONSE)
mdph13.call_situation_dossier(1234, SECRET, DOB)
request = mock_http.last_request
headers = request.headers
url = request.url
assert url == 'http://cd13.fr/situation/dossier/1234'
assert headers['X-CD13-Secret'] == base64.b64encode(SECRET.encode('utf-8')).decode('ascii')
assert headers['X-CD13-DateNaissBenef'] == '1993-05-04'
assert headers['X-CD13-Email'] == 'appel-sans-utilisateur@cd13.fr'
assert 'X-CD13-IP' not in headers
def test_call_situation_dossier_with_email_and_ip(mdph13, mock_http):
mock_http.add_response(VALID_RESPONSE)
mdph13.call_situation_dossier(1234, SECRET, DOB, email=EMAIL, ip=IP)
request = mock_http.last_request
headers = request.headers
url = request.url
assert url == 'http://cd13.fr/situation/dossier/1234'
assert headers['X-CD13-Secret'] == base64.b64encode(SECRET.encode('utf-8')).decode('ascii')
assert headers['X-CD13-DateNaissBenef'] == DOB_ISOFORMAT
assert headers['X-CD13-Email'] == EMAIL
assert headers['X-CD13-IP'] == IP
def test_link_bad_file_number(mdph13):
with pytest.raises(APIError) as e:
mdph13.link(
request=None, NameID=NAME_ID, numero_dossier='x', secret=None, date_de_naissance=None, email=None
)
assert str(e.value) == 'numero_dossier must be a number'
def test_link_bad_date_de_naissance(app, mdph13):
url = '/mdph13/test1/link/?NameID=%s&numero_dossier=%s&date_de_naissance=%s&secret=%s&email=%s'
url = url % (NAME_ID, FILE_NUMBER, '34-45-6', None, None)
response = app.post(url, status=400)
assert response.json['err_class'] == 'passerelle.views.InvalidParameterValue'
assert (
response.json['err_desc'] == 'invalid value for parameter "date_de_naissance (YYYY-MM-DD expected)"'
)
def test_link_bad_email(mdph13):
with pytest.raises(APIError) as e:
mdph13.link(
request=None,
NameID=NAME_ID,
numero_dossier=FILE_NUMBER,
secret=None,
date_de_naissance=DOB_ISOFORMAT,
email='xxx@@vvv',
)
assert str(e.value) == 'email is not valid'
def test_link_nok_dossier_inconnu(app, mdph13, mock_http):
mock_http.add_response(DOSSIER_INCONNU)
url = '/mdph13/test1/link/?NameID=%s&numero_dossier=%s&date_de_naissance=%s&secret=%s&email=%s'
url = url % (NAME_ID, FILE_NUMBER, DOB_ISOFORMAT, SECRET, EMAIL)
response = app.post(url)
assert response.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'
assert response.json['err_desc'] == 'dossier-inconnu'
def test_link_nok_secret_invalide(app, mdph13, mock_http):
mock_http.add_response(SECRET_INVALIDE)
url = '/mdph13/test1/link/?NameID=%s&numero_dossier=%s&date_de_naissance=%s&secret=%s&email=%s'
url = url % (NAME_ID, FILE_NUMBER, DOB_ISOFORMAT, SECRET, EMAIL)
response = app.post(url)
assert response.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'
assert response.json['err_desc'] == 'secret-invalide'
def test_link_numero_dont_match(app, mdph13, mock_http):
mock_http.add_response(json.dumps({'err': 0, 'data': {'numero': '456'}}))
url = '/mdph13/test1/link/?NameID=%s&numero_dossier=%s&date_de_naissance=%s&secret=%s&email=%s'
url = url % (NAME_ID, FILE_NUMBER, DOB_ISOFORMAT, SECRET, EMAIL)
response = app.post(url)
assert response.json['err_class'] == 'passerelle.utils.jsonresponse.APIError'
assert response.json['err_desc'] == 'numero-must-match-numero-dossier'
def test_link_ok(app, mdph13, mock_http):
# check first time link
mock_http.add_response(VALID_RESPONSE)
assert not Link.objects.count()
url = '/mdph13/test1/link/?NameID=%s&numero_dossier=%s&date_de_naissance=%s&secret=%s&email=%s&ip=%s'
url = url % (NAME_ID, FILE_NUMBER, DOB_ISOFORMAT, SECRET, EMAIL, IP)
response = app.post(url)
link = Link.objects.get()
assert response.json == {'err': 0, 'link_id': link.pk, 'created': True, 'updated': False}
# check relinking with update
mock_http.add_response(VALID_RESPONSE)
url = '/mdph13/test1/link/?NameID=%s&numero_dossier=%s&date_de_naissance=%s&secret=%s&email=%s&ip=%s'
url = url % (NAME_ID, FILE_NUMBER, DOB_ISOFORMAT, SECRET + 'a', EMAIL, IP)
response = app.post(url)
assert response.json == {
'err': 0,
'link_id': link.pk,
'created': False,
'updated': True,
}
def test_unlink_nok_bad_link_id(mdph13):
with pytest.raises(APIError) as e:
mdph13.unlink(None, None, 'e')
assert str(e.value) == 'link_id-must-be-a-number'
def test_unlink_ok(mdph13):
link = Link.objects.create(
resource=mdph13, name_id=NAME_ID, file_number=FILE_NUMBER, secret=SECRET, dob=DOB
)
result = mdph13.unlink(None, NAME_ID, str(link.pk))
assert result['deleted'] == 1
result = mdph13.unlink(None, NAME_ID, str(link.pk))
assert result['deleted'] == 0
def test_unlink_with_delete(mdph13, app):
link = Link.objects.create(
resource=mdph13, name_id=NAME_ID, file_number=FILE_NUMBER, secret=SECRET, dob=DOB
)
response = app.delete('/mdph13/%s/unlink/?NameID=%s&link_id=%s' % (mdph13.slug, NAME_ID, link.pk))
assert response.json['deleted'] == 1
def test_unlink_all_ok(mdph13):
Link.objects.create(resource=mdph13, name_id=NAME_ID, file_number=FILE_NUMBER, secret=SECRET, dob=DOB)
Link.objects.create(resource=mdph13, name_id=NAME_ID, file_number='12345', secret=SECRET, dob=DOB)
result = mdph13.unlink(None, NAME_ID, 'all')
assert result['deleted'] == 2
result = mdph13.unlink(None, NAME_ID, 'all')
assert result['deleted'] == 0
def test_dossier_ok(mdph13, mock_http):
link = Link.objects.create(
resource=mdph13, name_id=NAME_ID, file_number=FILE_NUMBER, secret=SECRET, dob=DOB
)
mock_http.add_response(VALID_RESPONSE)
response = mdph13.dossiers(None, NAME_ID, EMAIL, ip=IP)
assert response['data']
assert response['data'][0]['id'] == str(link.pk)
assert response['data'][0]['numero_dossier'] == FILE_NUMBER
assert response['data'][0]['date_de_naissance'] == DOB.isoformat()
assert response['data'][0]['dossier']['numero'] == FILE_NUMBER
assert response['data'][0]['text'] == 'Alfonso Martini #%s' % FILE_NUMBER
assert len(response['data'][0]['dossier']['beneficiaire']['entourage']) == 2
assert len(response['data'][0]['dossier']['beneficiaire']['entourage']['parents']) == 2
assert len(response['data'][0]['dossier']['beneficiaire']['entourage']['aidants']) == 1
assert len(response['data'][0]['dossier']['demandes']) == 2
assert len(response['data'][0]['dossier']['demandes']['en_cours']) == 1
assert len(response['data'][0]['dossier']['demandes']['historique']) == 7
# check demands are ordered by date
assert response['data'][0]['dossier']['demandes']['historique'][0]['date_demande'] == '2015-11-26'
dates = [demande['date_demande'] for demande in response['data'][0]['dossier']['demandes']['historique']]
assert sorted(dates, reverse=True) == dates
# check no demandes
valid_response = json.loads(VALID_RESPONSE)
del valid_response['data']['demandes']
mock_http.add_response(json.dumps(valid_response))
response = mdph13.dossiers(None, NAME_ID, EMAIL, ip=IP)
assert response['data']
assert response['data'][0]['err'] == 0
assert 'demandes' not in response['data'][0]['dossier']
def test_dossier_with_link_id_ok(mdph13, mock_http):
link = Link.objects.create(
resource=mdph13, name_id=NAME_ID, file_number=FILE_NUMBER, secret=SECRET, dob=DOB
)
mock_http.add_response(VALID_RESPONSE)
response = mdph13.dossiers(None, NAME_ID, EMAIL, link_id=str(link.pk), ip=IP)
assert response['data']
assert response['data']['id'] == str(link.pk)
assert response['data']['numero_dossier'] == FILE_NUMBER
assert response['data']['date_de_naissance'] == DOB.isoformat()
assert response['data']['dossier']['numero'] == FILE_NUMBER
assert response['data']['text'] == 'Alfonso Martini #%s' % FILE_NUMBER
assert len(response['data']['dossier']['beneficiaire']['entourage']) == 2
assert len(response['data']['dossier']['beneficiaire']['entourage']['parents']) == 2
assert len(response['data']['dossier']['beneficiaire']['entourage']['aidants']) == 1
assert len(response['data']['dossier']['demandes']) == 2
assert len(response['data']['dossier']['demandes']['en_cours']) == 1
assert len(response['data']['dossier']['demandes']['historique']) == 7
def test_dossier_partial_failure(mdph13, mock_http):
link1 = Link.objects.create(
resource=mdph13, name_id=NAME_ID, file_number=FILE_NUMBER, secret=SECRET, dob=DOB
)
link2 = Link.objects.create(
resource=mdph13, name_id=NAME_ID, file_number=FILE_NUMBER + '2', secret=SECRET, dob=DOB
)
mock_http.add_response(VALID_RESPONSE)
VALID_RESPONSE2 = json.loads(VALID_RESPONSE).copy()
VALID_RESPONSE2['data']['numero'] = FILE_NUMBER + '2'
mock_http.add_response(json.dumps(VALID_RESPONSE2))
# test that display_name is updated if changed after any access
link2.refresh_from_db()
assert not link2.display_name
response = mdph13.dossiers(None, NAME_ID, EMAIL)
link2.refresh_from_db()
assert link2.display_name
mock_http.add_response(VALID_RESPONSE)
mock_http.add_response({'status_code': 500, 'content': ''})
response = mdph13.dossiers(None, NAME_ID, EMAIL, ip=IP)
assert response['data']
assert response['data'][0]['id'] == str(link1.pk)
assert response['data'][0]['err'] == 0
assert response['data'][0]['text'] == 'Alfonso Martini #' + FILE_NUMBER
assert response['data'][1]['id'] == str(link2.pk)
assert response['data'][1]['err'] == 1
# verify display_name is returned even in case of failure
assert response['data'][1]['text'] == 'Alfonso Martini #' + FILE_NUMBER + '2'
def test_dossier_bad_date(mdph13, mock_http):
link = Link.objects.create(
resource=mdph13, name_id=NAME_ID, file_number=FILE_NUMBER, secret=SECRET, dob=DOB
)
INVALID_RESPONSE = json.loads(VALID_RESPONSE)
INVALID_RESPONSE['data']['demandes'][0]['date_demande'] = 'xxx'
mock_http.add_response(json.dumps(INVALID_RESPONSE))
with pytest.raises(APIError) as exc_info:
mdph13.dossiers(None, NAME_ID, EMAIL, link_id=str(link.pk))
assert str(exc_info.value) == 'invalid-response-format'
def test_dossier_http_error(app, mdph13, mock_http, caplog):
url = '/mdph13/test1/link/?NameID=%s&numero_dossier=%s&date_de_naissance=%s&secret=%s&email=%s&ip=%s'
url = url % (NAME_ID, FILE_NUMBER, DOB, SECRET, EMAIL, IP)
mock_http.add_response({'status_code': 401, 'content': 'wtf', 'reason': 'Authentication required'})
response = app.post(url, status=500)
assert response.json['err_class'] == 'requests.exceptions.HTTPError'
for record in caplog.records:
if (
record.getMessage() == 'GET http://cd13.fr/situation/dossier/1234 (=> 401)'
and record.levelno == logging.ERROR
):
break
else:
assert False, '401 caplog error message expected'
assert hasattr(record.request, 'META')
def raise_ssl_error(url, request):
raise requests.exceptions.SSLError(request=request)
caplog.clear()
mock_http.add_response(raise_ssl_error)
response = app.post(url, status=500)
assert response.json['err_class'] == 'requests.exceptions.SSLError'
for record in caplog.records:
if (
record.getMessage() == 'GET http://cd13.fr/situation/dossier/1234 (=> SSLError())'
and record.levelno == logging.ERROR
):
break
else:
assert False, 'SSLError caplog error message expected'
assert hasattr(record.request, 'META')
View Git Blame Copy Permalink