add MDPH13 connector (#30692)
This commit is contained in:
parent
6884d84f8c
commit
ad953198e9
|
|
@ -0,0 +1,61 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# Generated by Django 1.11.10 on 2019-02-15 09:57
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from django.db import migrations, models
|
||||
import django.db.models.deletion
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
initial = True
|
||||
|
||||
dependencies = [
|
||||
('base', '0010_loggingparameters_trace_emails'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='Link',
|
||||
fields=[
|
||||
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('name_id', models.CharField(max_length=256, verbose_name='NameID')),
|
||||
('file_number', models.CharField(max_length=64, verbose_name='MDPH beneficiary file number')),
|
||||
('secret', models.CharField(max_length=64, verbose_name='MDPH beneficiary secret')),
|
||||
('dob', models.DateField(verbose_name='MDPH beneficiary date of birth')),
|
||||
('created', models.DateTimeField(auto_now_add=True, verbose_name='Creation date')),
|
||||
],
|
||||
options={
|
||||
'ordering': ['file_number'],
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='MDPH13Resource',
|
||||
fields=[
|
||||
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('title', models.CharField(max_length=50, verbose_name='Title')),
|
||||
('description', models.TextField(verbose_name='Description')),
|
||||
('slug', models.SlugField(unique=True, verbose_name='Identifier')),
|
||||
('basic_auth_username', models.CharField(blank=True, max_length=128, verbose_name='Basic authentication username')),
|
||||
('basic_auth_password', models.CharField(blank=True, max_length=128, verbose_name='Basic authentication password')),
|
||||
('client_certificate', models.FileField(blank=True, null=True, upload_to=b'', verbose_name='TLS client certificate')),
|
||||
('trusted_certificate_authorities', models.FileField(blank=True, null=True, upload_to=b'', verbose_name='TLS trusted CAs')),
|
||||
('verify_cert', models.BooleanField(default=True, verbose_name='TLS verify certificates')),
|
||||
('http_proxy', models.CharField(blank=True, max_length=128, verbose_name='HTTP and HTTPS proxy')),
|
||||
('webservice_base_url', models.URLField(verbose_name='Webservice Base URL')),
|
||||
('users', models.ManyToManyField(blank=True, to='base.ApiUser')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': 'MDPH CD13',
|
||||
},
|
||||
),
|
||||
migrations.AddField(
|
||||
model_name='link',
|
||||
name='resource',
|
||||
field=models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='mdph13.MDPH13Resource'),
|
||||
),
|
||||
migrations.AlterUniqueTogether(
|
||||
name='link',
|
||||
unique_together=set([('resource', 'name_id', 'file_number')]),
|
||||
),
|
||||
]
|
||||
|
|
@ -0,0 +1,340 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# passerelle - uniform access to multiple data sources and services
|
||||
# Copyright (C) 2019 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 urlparse
|
||||
import datetime
|
||||
import re
|
||||
|
||||
import requests
|
||||
|
||||
from django.db import models, transaction
|
||||
from django.utils import six
|
||||
from django.utils.translation import ugettext_lazy as _
|
||||
|
||||
from passerelle.utils.jsonresponse import APIError, to_json
|
||||
from passerelle.utils.api import endpoint
|
||||
from passerelle.base.models import BaseResource, HTTPResource
|
||||
|
||||
|
||||
class MDPH13Resource(BaseResource, HTTPResource):
|
||||
category = _('Business Process Connectors')
|
||||
|
||||
webservice_base_url = models.URLField(_('Webservice Base URL'))
|
||||
|
||||
EMAIL_RE = re.compile(r'^[^@\s]+@[^@\s]+\.[^@\s]+$')
|
||||
|
||||
class Meta:
|
||||
verbose_name = _('MDPH CD13')
|
||||
|
||||
def situation_dossier_url(self, file_number):
|
||||
return urlparse.urljoin(self.webservice_base_url, 'situation/dossier/%s' % file_number)
|
||||
|
||||
def url_get(self, *args, **kwargs):
|
||||
try:
|
||||
response = None
|
||||
response = self.requests.get(*args, **kwargs)
|
||||
response.raise_for_status()
|
||||
except requests.RequestException as e:
|
||||
data = {'exception': six.text_type(e)}
|
||||
if response is not None:
|
||||
try:
|
||||
if response.json():
|
||||
data['body'] = response.json()
|
||||
except ValueError:
|
||||
body = response.content[:1000]
|
||||
if len(response.content) > 1000:
|
||||
body += b'<SNIPPED>'
|
||||
data['body'] = repr(body)
|
||||
data['status_code'] = response.status_code
|
||||
raise APIError('HTTP request failed', data=data)
|
||||
try:
|
||||
content = response.json()
|
||||
except ValueError as e:
|
||||
data = {'exception': six.text_type(e)}
|
||||
data['body'] = '%r<SNIPPED>' % response.content[:1000]
|
||||
raise APIError('HTTP request did not respond with JSON', data=data)
|
||||
return content
|
||||
|
||||
def call_situation_dossier(self, file_number, secret, dob, email=None):
|
||||
url = self.situation_dossier_url(file_number)
|
||||
email = email or 'appel-sans-utilisateur@cd13.fr'
|
||||
error_mapping = {
|
||||
'dossier-inconnu': 'dossier-inconnu',
|
||||
'secret-invalide': 'secret-invalide',
|
||||
'dateNaissance-erronee': 'date-de-naissance-erronee',
|
||||
}
|
||||
try:
|
||||
content = self.url_get(
|
||||
url,
|
||||
headers={
|
||||
'X-CD13-Secret': base64.b64encode(secret.encode('utf-8')).decode('ascii'),
|
||||
'X-CD13-Email': email,
|
||||
'X-CD13-DateNaissBenef': dob.isoformat(),
|
||||
})
|
||||
except APIError as e:
|
||||
if e.data and e.data.get('status_code') == 404 and isinstance(e.data.get('body'), dict):
|
||||
content = e.data['body']
|
||||
if content.get('err') != 0:
|
||||
error_code = content.get('err_code')
|
||||
err_desc = error_mapping.get(error_code, 'err != 0')
|
||||
raise APIError(err_desc, data=e.data)
|
||||
raise
|
||||
|
||||
if content.get('err') != 0:
|
||||
error_code = content.get('err_code')
|
||||
err_desc = error_mapping.get(error_code, 'err != 0')
|
||||
raise APIError(err_desc, data=content)
|
||||
|
||||
data = content.get('data')
|
||||
if not isinstance(data, dict):
|
||||
raise APIError('data-must-be-a-dict', data=content)
|
||||
|
||||
if str(data.get('numero')) != str(file_number):
|
||||
raise APIError('numero-must-match-numero-dossier', date=content)
|
||||
|
||||
# Reorganize entourage
|
||||
beneficiaire = data.get('beneficiaire', {})
|
||||
entourage = beneficiaire.get('entourage')
|
||||
if entourage is not None:
|
||||
if not isinstance(entourage, list):
|
||||
raise APIError('entourage-must-be-a-list', data=content)
|
||||
if not all(isinstance(person, dict) for person in entourage):
|
||||
raise APIError('demandes-content-must-be-dicts', data=content)
|
||||
new_entourage = {}
|
||||
for person in entourage:
|
||||
if not isinstance(person, dict):
|
||||
raise APIError('entourage-content-must-be-dicts', data=content)
|
||||
if person.get('role') in [u'Père', u'Mère']:
|
||||
new_entourage.setdefault('parents', []).append(person)
|
||||
else:
|
||||
new_entourage.setdefault('aidants', []).append(person)
|
||||
beneficiaire['entourage'] = new_entourage
|
||||
|
||||
# Reorganize demandes
|
||||
demandes = data.get('demandes')
|
||||
if demandes:
|
||||
if not isinstance(demandes, list):
|
||||
raise APIError('demandes-must-be-a-list', data=content)
|
||||
if not all(isinstance(demande, dict) for demande in demandes):
|
||||
raise APIError('demandes-content-must-be-dicts', data=content)
|
||||
new_demandes = {}
|
||||
typologies = {
|
||||
u'demande en cours': 'en_cours',
|
||||
u'traitée et expédiée': 'historique',
|
||||
u'traitée non expédiée': 'historique',
|
||||
}
|
||||
if not all(isinstance(demande.get('typologie'), six.text_type) for demande in demandes):
|
||||
raise APIError('typologie-must-be-a-string', data=content)
|
||||
if not all(demande['typologie'].lower() in typologies for demande in demandes):
|
||||
unknowns = set([demande['typologie'].lower() for demande in demandes]) - set(typologies.keys())
|
||||
raise APIError('typologie-is-unknown',
|
||||
data={
|
||||
'unknowns': list(unknowns),
|
||||
'choices': typologies.keys()
|
||||
})
|
||||
for demande in demandes:
|
||||
new_demandes.setdefault(
|
||||
typologies[demande['typologie'].lower()],
|
||||
[]
|
||||
).append(demande)
|
||||
data['demandes'] = new_demandes
|
||||
return data
|
||||
|
||||
def check_status(self):
|
||||
try:
|
||||
link = Link.objects.latest('created')
|
||||
except Link.DoesNotExist:
|
||||
return
|
||||
# no email passed, it's a background check
|
||||
link.get_file()
|
||||
|
||||
@endpoint(name='link',
|
||||
methods=['post'],
|
||||
description=_('Create link with an extranet account'),
|
||||
perm='can_access',
|
||||
parameters={
|
||||
'NameID': {
|
||||
'description': _('Publik NameID'),
|
||||
'example_value': 'xyz24d934',
|
||||
},
|
||||
'numero_dossier': {
|
||||
'description': _('MDPH13 beneficiary file number'),
|
||||
'example_value': '1234',
|
||||
},
|
||||
'secret': {
|
||||
'description': _('MDPH13 beneficiary secret'),
|
||||
'example_value': 'secret',
|
||||
},
|
||||
'date_de_naissance': {
|
||||
'description': _('MDPH13 beneficiary date of birth'),
|
||||
'example_value': '1992-03-05',
|
||||
},
|
||||
'email': {
|
||||
'description': _('Publik known email'),
|
||||
'example_value': 'john.doe@example.com',
|
||||
},
|
||||
})
|
||||
def link(self, request, NameID, numero_dossier, secret, date_de_naissance, email):
|
||||
file_number = numero_dossier.strip()
|
||||
try:
|
||||
int(file_number)
|
||||
except ValueError:
|
||||
raise APIError('numero_dossier must be a number', http_status=400)
|
||||
try:
|
||||
dob = datetime.datetime.strptime(date_de_naissance.strip(), '%Y-%m-%d').date()
|
||||
except ValueError:
|
||||
raise APIError('date_de_naissance must be a date YYYY-MM-DD', http_status=400)
|
||||
email = email.strip()
|
||||
if not self.EMAIL_RE.match(email):
|
||||
raise APIError('email is not valid', http_status=400)
|
||||
with transaction.atomic():
|
||||
link, created = Link.objects.get_or_create(
|
||||
resource=self,
|
||||
name_id=NameID,
|
||||
file_number=file_number,
|
||||
defaults={
|
||||
'secret': secret,
|
||||
'dob': dob
|
||||
})
|
||||
updated = False
|
||||
if not created:
|
||||
if link.secret != secret or link.dob != dob:
|
||||
link.secret = secret
|
||||
link.dob = dob
|
||||
updated = True
|
||||
# email is necessary for audit purpose
|
||||
link.get_file(email=email)
|
||||
if updated:
|
||||
link.save()
|
||||
return {'link_id': link.pk, 'created': created, 'updated': updated}
|
||||
|
||||
@endpoint(name='unlink',
|
||||
methods=['post', 'delete'],
|
||||
description=_('Delete link with an extranet account'),
|
||||
perm='can_access',
|
||||
parameters={
|
||||
'NameID': {
|
||||
'description': _('Publik NameID'),
|
||||
'example_value': 'xyz24d934',
|
||||
},
|
||||
'link_id': {
|
||||
'description': _('Identifier of the link'),
|
||||
'example_value': '1',
|
||||
},
|
||||
})
|
||||
def unlink(self, request, NameID, link_id):
|
||||
qs = Link.objects.filter(resource=self, name_id=NameID)
|
||||
if link_id == 'all':
|
||||
pass # unlink of all links
|
||||
else:
|
||||
try:
|
||||
link_id = int(link_id.strip())
|
||||
except ValueError:
|
||||
raise APIError('link_id-must-be-a-number')
|
||||
qs = qs.filter(pk=link_id)
|
||||
count = qs.count()
|
||||
qs.delete()
|
||||
return {'deleted': count}
|
||||
|
||||
@endpoint(name='dossiers',
|
||||
description=_('Get datas for all links, or for a specified one'),
|
||||
perm='can_access',
|
||||
parameters={
|
||||
'NameID': {
|
||||
'description': _('Publik NameID'),
|
||||
'example_value': 'xyz24d934',
|
||||
},
|
||||
'email': {
|
||||
'description': _('Publik known email'),
|
||||
'example_value': 'john.doe@example.com',
|
||||
},
|
||||
'link_id': {
|
||||
'description': _('Link identifier'),
|
||||
'example_value': '1',
|
||||
}
|
||||
})
|
||||
def dossiers(self, request, NameID, email, link_id=None):
|
||||
email = email.strip()
|
||||
if not self.EMAIL_RE.match(email):
|
||||
raise APIError('email is not valid', http_status=400)
|
||||
|
||||
qs = Link.objects.filter(
|
||||
resource=self,
|
||||
name_id=NameID)
|
||||
if link_id:
|
||||
try:
|
||||
link_id = int(link_id)
|
||||
except ValueError:
|
||||
raise APIError('invalid-link-id', http_status=400)
|
||||
qs = qs.filter(id=link_id)
|
||||
data = []
|
||||
for link in qs:
|
||||
file_data = {
|
||||
'id': str(link.id),
|
||||
'err': 0,
|
||||
}
|
||||
try:
|
||||
mdph_file = link.get_file(email=email)
|
||||
except Exception as e:
|
||||
if link_id:
|
||||
raise
|
||||
file_data.update(to_json().err_to_response(e))
|
||||
else:
|
||||
file_data.update({
|
||||
'numero_dossier': link.file_number,
|
||||
'date_de_naissance': link.dob.isoformat(),
|
||||
'dossier': mdph_file,
|
||||
})
|
||||
data.append(file_data)
|
||||
if link_id:
|
||||
return {'data': data[0] if data else None}
|
||||
return {'data': data}
|
||||
|
||||
|
||||
class Link(models.Model):
|
||||
resource = models.ForeignKey(
|
||||
MDPH13Resource,
|
||||
on_delete=models.CASCADE)
|
||||
name_id = models.CharField(
|
||||
verbose_name=_('NameID'),
|
||||
max_length=256)
|
||||
file_number = models.CharField(
|
||||
max_length=64,
|
||||
verbose_name=_('MDPH beneficiary file number'))
|
||||
secret = models.CharField(
|
||||
verbose_name=_('MDPH beneficiary secret'),
|
||||
max_length=64)
|
||||
dob = models.DateField(
|
||||
verbose_name=_('MDPH beneficiary date of birth'))
|
||||
created = models.DateTimeField(
|
||||
verbose_name=_('Creation date'),
|
||||
auto_now_add=True)
|
||||
|
||||
def get_file(self, email=None):
|
||||
# email is necessary for audit purpose
|
||||
return self.resource.call_situation_dossier(
|
||||
file_number=self.file_number,
|
||||
secret=self.secret,
|
||||
dob=self.dob,
|
||||
email=email)
|
||||
|
||||
class Meta:
|
||||
unique_together = (
|
||||
'resource', 'name_id', 'file_number',
|
||||
)
|
||||
ordering = ['file_number']
|
||||
|
|
@ -27,6 +27,7 @@ INSTALLED_APPS += (
|
|||
'passerelle.contrib.iws',
|
||||
'passerelle.contrib.maarch',
|
||||
'passerelle.contrib.mdel',
|
||||
'passerelle.contrib.mdph13',
|
||||
'passerelle.contrib.meyzieu_newsletters',
|
||||
'passerelle.contrib.nancypoll',
|
||||
'passerelle.contrib.planitech',
|
||||
|
|
|
|||
|
|
@ -0,0 +1,462 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# 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 json
|
||||
import base64
|
||||
import datetime
|
||||
|
||||
import httmock
|
||||
import pytest
|
||||
|
||||
import utils
|
||||
|
||||
from passerelle.utils.jsonresponse import APIError
|
||||
from passerelle.contrib.mdph13.models import MDPH13Resource, Link
|
||||
|
||||
NAME_ID = 'xyz'
|
||||
FILE_NUMBER = '1234'
|
||||
DOB = datetime.date(1993, 5, 4)
|
||||
DOB_ISOFORMAT = '1993-05-04'
|
||||
EMAIL = 'john.doe@example.com'
|
||||
SECRET = 'secret'
|
||||
|
||||
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": "2015-01-16"
|
||||
},
|
||||
{
|
||||
"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 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(object):
|
||||
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)
|
||||
return self.responses[idx]
|
||||
|
||||
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'
|
||||
|
||||
|
||||
def test_call_situation_dossier_with_email(mdph13, mock_http):
|
||||
mock_http.add_response(VALID_RESPONSE)
|
||||
mdph13.call_situation_dossier(1234, SECRET, DOB, email=EMAIL)
|
||||
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
|
||||
|
||||
|
||||
def test_check_status_no_link(mdph13):
|
||||
assert Link.objects.count() == 0
|
||||
try:
|
||||
mdph13.check_status()
|
||||
except Exception as e:
|
||||
pytest.fail('check_status() should not raise')
|
||||
|
||||
|
||||
def test_check_status_with_link_nok(mdph13, mock_http):
|
||||
mock_http.add_response({'status_code': 500})
|
||||
Link.objects.create(
|
||||
resource=mdph13,
|
||||
name_id=NAME_ID,
|
||||
file_number=FILE_NUMBER,
|
||||
secret=SECRET,
|
||||
dob=DOB)
|
||||
assert Link.objects.count() == 1
|
||||
with pytest.raises(Exception):
|
||||
mdph13.check_status()
|
||||
|
||||
|
||||
def test_check_status_with_link_ok(mdph13, mock_http):
|
||||
mock_http.add_response(VALID_RESPONSE)
|
||||
Link.objects.create(
|
||||
resource=mdph13,
|
||||
name_id=NAME_ID,
|
||||
file_number=FILE_NUMBER,
|
||||
secret=SECRET,
|
||||
dob=DOB)
|
||||
assert Link.objects.count() == 1
|
||||
try:
|
||||
mdph13.check_status()
|
||||
except Exception:
|
||||
pytest.fail('check_status() should not raise')
|
||||
|
||||
|
||||
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(mdph13):
|
||||
with pytest.raises(APIError) as e:
|
||||
mdph13.link(request=None, NameID=NAME_ID, numero_dossier=FILE_NUMBER, secret=None,
|
||||
date_de_naissance='34-45-6', email=None)
|
||||
assert str(e.value) == 'date_de_naissance must be a date YYYY-MM-DD'
|
||||
|
||||
|
||||
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(mdph13, mock_http):
|
||||
mock_http.add_response(DOSSIER_INCONNU)
|
||||
with pytest.raises(APIError) as e:
|
||||
mdph13.link(request=None, NameID=NAME_ID, numero_dossier=FILE_NUMBER, secret=SECRET,
|
||||
date_de_naissance=DOB_ISOFORMAT, email=EMAIL)
|
||||
assert str(e.value) == 'dossier-inconnu'
|
||||
|
||||
|
||||
def test_link_nok_secret_invalide(mdph13, mock_http):
|
||||
mock_http.add_response(SECRET_INVALIDE)
|
||||
with pytest.raises(APIError) as e:
|
||||
mdph13.link(request=None, NameID=NAME_ID, numero_dossier=FILE_NUMBER, secret=SECRET,
|
||||
date_de_naissance=DOB_ISOFORMAT, email=EMAIL)
|
||||
assert str(e.value) == 'secret-invalide'
|
||||
|
||||
|
||||
def test_link_numero_dont_match(mdph13, mock_http):
|
||||
mock_http.add_response(json.dumps({'err': 0, 'data': {'numero': '456'}}))
|
||||
with pytest.raises(APIError) as e:
|
||||
mdph13.link(request=None, NameID=NAME_ID, numero_dossier=FILE_NUMBER, secret=SECRET,
|
||||
date_de_naissance=DOB_ISOFORMAT, email=EMAIL)
|
||||
assert str(e.value) == 'numero-must-match-numero-dossier'
|
||||
|
||||
|
||||
def test_link_ok(mdph13, mock_http):
|
||||
# check first time link
|
||||
mock_http.add_response(VALID_RESPONSE)
|
||||
assert not Link.objects.count()
|
||||
response = mdph13.link(request=None, NameID=NAME_ID,
|
||||
numero_dossier=FILE_NUMBER, secret=SECRET,
|
||||
date_de_naissance=DOB_ISOFORMAT, email=EMAIL)
|
||||
link = Link.objects.get()
|
||||
assert response == {
|
||||
'link_id': link.pk,
|
||||
'created': True,
|
||||
'updated': False
|
||||
}
|
||||
# check relinking with update
|
||||
mock_http.add_response(VALID_RESPONSE)
|
||||
response = mdph13.link(request=None, NameID=NAME_ID,
|
||||
numero_dossier=FILE_NUMBER, secret=SECRET+'a',
|
||||
date_de_naissance=DOB_ISOFORMAT, email=EMAIL)
|
||||
assert response == {
|
||||
'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_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)
|
||||
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 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
|
||||
|
||||
|
||||
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))
|
||||
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 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)
|
||||
mock_http.add_response({'status_code': 500, 'content': ''})
|
||||
response = mdph13.dossiers(None, NAME_ID, EMAIL)
|
||||
assert response['data']
|
||||
assert response['data'][0]['id'] == str(link1.pk)
|
||||
assert response['data'][0]['err'] == 0
|
||||
assert response['data'][1]['id'] == str(link2.pk)
|
||||
assert response['data'][1]['err'] == 1
|
||||
1
tox.ini
1
tox.ini
|
|
@ -29,6 +29,7 @@ deps =
|
|||
mohawk
|
||||
pytest-freezegun
|
||||
pytest-httpbin
|
||||
pytest-localserver
|
||||
commands =
|
||||
django18: py.test {posargs: {env:FAST:} --junitxml=test_{envname}_results.xml --cov-report xml --cov-report html --cov=passerelle/ --cov-config .coveragerc tests/}
|
||||
django18: ./pylint.sh passerelle/
|
||||
|
|
|
|||
Loading…
Reference in New Issue