add connector for Cart@DS CS (#27144)
This commit is contained in:
parent
8fc099421f
commit
81690434be
|
|
@ -0,0 +1,69 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
# Generated by Django 1.11.12 on 2019-02-20 10:38
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from django.db import migrations, models
|
||||
import passerelle.apps.cartads_cs.models
|
||||
|
||||
|
||||
class Migration(migrations.Migration):
|
||||
|
||||
initial = True
|
||||
|
||||
dependencies = [
|
||||
('base', '0012_job'),
|
||||
]
|
||||
|
||||
operations = [
|
||||
migrations.CreateModel(
|
||||
name='CartaDSCS',
|
||||
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')),
|
||||
('wsdl_base_url', models.URLField(help_text='ex: https://example.net/adscs/webservices/', verbose_name='WSDL Base URL')),
|
||||
('username', models.CharField(max_length=64, verbose_name='Username')),
|
||||
('password', models.CharField(max_length=64, verbose_name='Password')),
|
||||
('iv', models.CharField(max_length=16, verbose_name='Initialisation Vector')),
|
||||
('secret_key', models.CharField(max_length=16, verbose_name='Secret Key')),
|
||||
('ftp_server', models.CharField(max_length=128, verbose_name='FTP Server')),
|
||||
('ftp_username', models.CharField(max_length=64, verbose_name='FTP Username')),
|
||||
('ftp_password', models.CharField(max_length=64, verbose_name='FTP Password')),
|
||||
('ftp_client_name', models.CharField(max_length=64, verbose_name='FTP Client Name')),
|
||||
('users', models.ManyToManyField(blank=True, to='base.ApiUser')),
|
||||
],
|
||||
options={
|
||||
'verbose_name': 'Cart@DS CS',
|
||||
},
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='CartaDSDossier',
|
||||
fields=[
|
||||
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('email', models.CharField(max_length=256)),
|
||||
('tracking_code', models.CharField(max_length=20)),
|
||||
('commune_id', models.CharField(max_length=20)),
|
||||
('type_dossier_id', models.CharField(max_length=20)),
|
||||
('objet_demande_id', models.CharField(max_length=20, null=True)),
|
||||
('zip_ready', models.BooleanField(default=False)),
|
||||
('zip_sent', models.BooleanField(default=False)),
|
||||
('zip_ack_response', models.CharField(max_length=20, null=True)),
|
||||
('notification_url', models.URLField(null=True)),
|
||||
('notification_message', models.TextField(null=True)),
|
||||
('cartads_id_dossier', models.CharField(max_length=50, null=True)),
|
||||
('cartads_numero_dossier', models.CharField(max_length=50, null=True)),
|
||||
('last_update_datetime', models.DateTimeField(auto_now=True)),
|
||||
],
|
||||
),
|
||||
migrations.CreateModel(
|
||||
name='CartaDSFile',
|
||||
fields=[
|
||||
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
|
||||
('tracking_code', models.CharField(max_length=20)),
|
||||
('id_piece', models.CharField(max_length=20)),
|
||||
('uploaded_file', models.FileField(upload_to=passerelle.apps.cartads_cs.models.cartads_file_location)),
|
||||
('last_update_datetime', models.DateTimeField(auto_now=True)),
|
||||
],
|
||||
),
|
||||
]
|
||||
|
|
@ -0,0 +1,445 @@
|
|||
# -*- 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 base64
|
||||
import datetime
|
||||
from ftplib import FTP
|
||||
import json
|
||||
import os
|
||||
from xml.etree import ElementTree as etree
|
||||
import zipfile
|
||||
|
||||
from Crypto.Cipher import AES
|
||||
|
||||
from django.core.files.storage import default_storage
|
||||
from django.core.signing import Signer
|
||||
from django.core.urlresolvers import reverse
|
||||
from django.db import models
|
||||
from django.http import HttpResponse
|
||||
from django.utils.translation import ugettext_lazy as _
|
||||
from django.utils.six.moves.urllib import parse as urlparse
|
||||
|
||||
from passerelle.base.models import BaseResource
|
||||
from passerelle.utils.api import endpoint
|
||||
|
||||
|
||||
def cartads_file_location(instance, filename):
|
||||
return 'cartads_cs/%s/%s' % (instance.tracking_code, filename)
|
||||
|
||||
|
||||
class CartaDSFile(models.Model):
|
||||
tracking_code = models.CharField(max_length=20)
|
||||
id_piece = models.CharField(max_length=20)
|
||||
uploaded_file = models.FileField(upload_to=cartads_file_location)
|
||||
last_update_datetime = models.DateTimeField(auto_now=True)
|
||||
|
||||
|
||||
class CartaDSDossier(models.Model):
|
||||
email = models.CharField(max_length=256)
|
||||
tracking_code = models.CharField(max_length=20)
|
||||
commune_id = models.CharField(max_length=20)
|
||||
type_dossier_id = models.CharField(max_length=20)
|
||||
objet_demande_id = models.CharField(max_length=20, null=True)
|
||||
zip_ready = models.BooleanField(default=False)
|
||||
zip_sent = models.BooleanField(default=False)
|
||||
zip_ack_response = models.CharField(null=True, max_length=20)
|
||||
notification_url = models.URLField(null=True)
|
||||
notification_message = models.TextField(null=True)
|
||||
cartads_id_dossier = models.CharField(max_length=50, null=True)
|
||||
cartads_numero_dossier = models.CharField(max_length=50, null=True)
|
||||
last_update_datetime = models.DateTimeField(auto_now=True)
|
||||
|
||||
|
||||
class CartaDSCS(BaseResource):
|
||||
category = _('Misc')
|
||||
|
||||
wsdl_base_url = models.URLField(_('WSDL Base URL'),
|
||||
help_text=_('ex: https://example.net/adscs/webservices/'))
|
||||
username = models.CharField(_('Username'), max_length=64)
|
||||
password = models.CharField(_('Password'), max_length=64)
|
||||
iv = models.CharField(_('Initialisation Vector'), max_length=16)
|
||||
secret_key = models.CharField(_('Secret Key'), max_length=16)
|
||||
ftp_server = models.CharField(_('FTP Server'), max_length=128)
|
||||
ftp_username = models.CharField(_('FTP Username'), max_length=64)
|
||||
ftp_password = models.CharField(_('FTP Password'), max_length=64)
|
||||
ftp_client_name = models.CharField(_('FTP Client Name'), max_length=64)
|
||||
|
||||
class Meta:
|
||||
verbose_name = 'Cart@DS CS'
|
||||
|
||||
@property
|
||||
def wsdl_url(self):
|
||||
return self.get_wsdl_url()
|
||||
|
||||
def get_wsdl_url(self, service_type='ServicePortail'):
|
||||
return self.wsdl_base_url + service_type + '.svc?singleWsdl'
|
||||
|
||||
def soap_client(self, **kwargs):
|
||||
client = super(CartaDSCS, self).soap_client(**kwargs)
|
||||
# fix URL that should have been changed by reverse proxy
|
||||
parsed_wsdl_address = urlparse.urlparse(client.service._binding_options['address'])
|
||||
parsed_real_address = urlparse.urlparse(self.wsdl_base_url)
|
||||
client.service._binding_options['address'] = urlparse.urlunparse(
|
||||
parsed_real_address[:2] + parsed_wsdl_address[2:])
|
||||
return client
|
||||
|
||||
def get_token(self):
|
||||
token_data = {
|
||||
'date': datetime.datetime.now().strftime('%d/%m/%Y %H:%M:%S'),
|
||||
'login': self.username,
|
||||
'password': self.password,
|
||||
}
|
||||
token_data_str = json.dumps(token_data)
|
||||
data_pad = AES.block_size - len(token_data_str) % AES.block_size
|
||||
aes = AES.new(self.secret_key, AES.MODE_CBC, self.iv)
|
||||
token = aes.encrypt(token_data_str + (chr(data_pad)*data_pad))
|
||||
return base64.encodestring(token).replace('\n', '').rstrip('=')
|
||||
|
||||
def check_status(self):
|
||||
self.soap_client().service.GetCommunes(self.get_token())
|
||||
|
||||
# description of common endpoint parameters
|
||||
COMMUNE_ID_PARAM = {
|
||||
'description': _('Identifier of collectivity'),
|
||||
'example_value': '2'
|
||||
}
|
||||
TYPE_DOSSIER_ID_PARAM = {
|
||||
'description': _('Identifier of file type'),
|
||||
'example_value': 'CU',
|
||||
}
|
||||
OBJET_DEMANDE_ID_PARAM = {
|
||||
'description': _('Identifier of demand subject'),
|
||||
'example_value': '1',
|
||||
}
|
||||
TRACKING_CODE_PARAM = {
|
||||
'description': _('Unique identifier (ex: tracking code)'),
|
||||
'example_value': 'XCBTFRML',
|
||||
}
|
||||
PIECE_ID_PARAM = {
|
||||
'description': _('Identifier of single file item'),
|
||||
}
|
||||
UPLOAD_TOKEN_PARAM = {
|
||||
'description': _('Token for upload file'),
|
||||
}
|
||||
|
||||
@endpoint(description=_('Get list of collectivities'))
|
||||
def communes(self, request):
|
||||
client = self.soap_client()
|
||||
resp = client.service.GetCommunes(self.get_token())
|
||||
return {'data': [{'id': str(x['Key']), 'text': x['Value']} for x in resp]}
|
||||
|
||||
@endpoint(description=_('Get lisf of file types'),
|
||||
parameters={
|
||||
'commune_id': COMMUNE_ID_PARAM
|
||||
})
|
||||
def types_dossier(self, request, commune_id):
|
||||
client = self.soap_client()
|
||||
resp = client.service.GetTypesDossier(self.get_token(), int(commune_id))
|
||||
return {'data': [{'id': x['Key'], 'text': x['Value']} for x in resp]}
|
||||
|
||||
@endpoint(description=_('Get list of demand subjects'),
|
||||
parameters={'type_dossier_id': TYPE_DOSSIER_ID_PARAM},
|
||||
)
|
||||
def objets_demande(self, request, type_dossier_id):
|
||||
client = self.soap_client()
|
||||
resp = client.service.GetObjetsDemande(self.get_token(), type_dossier_id)
|
||||
return {'data': [{'id': str(x['Key']), 'text': x['Value']} for x in resp]}
|
||||
|
||||
@endpoint(description=_('Get list of CERFA documents'),
|
||||
parameters={
|
||||
'type_dossier_id': TYPE_DOSSIER_ID_PARAM,
|
||||
'type_compte': {'description': _('Type of account')},
|
||||
})
|
||||
def liste_pdf(self, request, type_dossier_id, type_compte=1):
|
||||
client = self.soap_client()
|
||||
resp = client.service.GetListePdf(self.get_token(), type_dossier_id,
|
||||
{'TypeCompteUtilisateur': type_compte})
|
||||
return {'data': [
|
||||
{'id': x['Identifiant'],
|
||||
'text': u'%s: %s' % (x['Nom'], x['Description']),
|
||||
'url': x['UrlTelechargement'],
|
||||
} for x in resp]}
|
||||
|
||||
@endpoint(perm='can_access',
|
||||
description=_('Get list of file items'),
|
||||
parameters={
|
||||
'type_dossier_id': TYPE_DOSSIER_ID_PARAM,
|
||||
'objet_demande_id': OBJET_DEMANDE_ID_PARAM,
|
||||
'tracking_code': TRACKING_CODE_PARAM,
|
||||
})
|
||||
def pieces(self, request, type_dossier_id, objet_demande_id, tracking_code):
|
||||
client = self.soap_client()
|
||||
resp = client.service.GetPieces(self.get_token(), type_dossier_id,
|
||||
objet_demande_id)
|
||||
signer = Signer(salt='cart@ds_cs')
|
||||
upload_token = signer.sign(tracking_code)
|
||||
cerfa_pieces = [
|
||||
{'id': 'cerfa-%s-%s' % (type_dossier_id, objet_demande_id),
|
||||
'text': 'Cerfa rempli',
|
||||
'description': '',
|
||||
'codePiece': '',
|
||||
'reglementaire': True,
|
||||
'files': [],
|
||||
'max_files': 1,
|
||||
'section_start': 'Cerfa',
|
||||
},
|
||||
{'id': 'cerfa-autres-%s-%s' % (type_dossier_id, objet_demande_id),
|
||||
'text': 'Cerfa demandeurs complémentaires',
|
||||
'description': '',
|
||||
'codePiece': '',
|
||||
'reglementaire': False,
|
||||
'files': [],
|
||||
'max_files': 6,
|
||||
}
|
||||
]
|
||||
pieces = [
|
||||
{'id': str(x['IdPiece']),
|
||||
'text': x['Libelle'],
|
||||
'description': x['Descriptif'],
|
||||
'codePiece': x['CodePiece'],
|
||||
'reglementaire': x['Reglementaire'],
|
||||
'files': [],
|
||||
'max_files': 6,
|
||||
} for x in resp]
|
||||
required_pieces = [x for x in pieces if x['reglementaire']]
|
||||
if required_pieces:
|
||||
required_pieces[0]['section_start'] = 'Pièces réglementaires'
|
||||
optional_pieces = [x for x in pieces if not x['reglementaire']]
|
||||
if optional_pieces:
|
||||
optional_pieces[0]['section_start'] = 'Pièces spécifiques'
|
||||
pieces = cerfa_pieces + required_pieces + optional_pieces
|
||||
known_files = CartaDSFile.objects.filter(tracking_code=tracking_code)
|
||||
for piece in pieces:
|
||||
if request:
|
||||
upload_url = request.build_absolute_uri('%supload/%s/%s/' % (
|
||||
self.get_absolute_url(),
|
||||
piece['id'],
|
||||
upload_token))
|
||||
else:
|
||||
upload_url = None
|
||||
piece['files'] = [
|
||||
{
|
||||
'url': upload_url,
|
||||
'name': os.path.basename(x.uploaded_file.name),
|
||||
'token': signer.sign(str(x.id)),
|
||||
'id': x.id,
|
||||
} for x in known_files if x.id_piece == str(piece['id'])]
|
||||
if len(piece['files']) < piece['max_files']:
|
||||
piece['files'].append({'url': upload_url})
|
||||
return {'data': pieces}
|
||||
|
||||
@endpoint(perm='can_access',
|
||||
description=_('Check list of file items'),
|
||||
parameters={
|
||||
'type_dossier_id': TYPE_DOSSIER_ID_PARAM,
|
||||
'objet_demande_id': OBJET_DEMANDE_ID_PARAM,
|
||||
'tracking_code': TRACKING_CODE_PARAM,
|
||||
})
|
||||
def check_pieces(self, request, type_dossier_id, objet_demande_id, tracking_code):
|
||||
pieces = self.pieces(request, type_dossier_id, objet_demande_id, tracking_code)
|
||||
result = True
|
||||
for piece in pieces['data']:
|
||||
if not piece['reglementaire']:
|
||||
continue
|
||||
if not [x for x in piece['files'] if x.get('name')]:
|
||||
result = False
|
||||
break
|
||||
return {'result': result}
|
||||
|
||||
@endpoint(methods=['post'],
|
||||
pattern='^(?P<id_piece>[\w-]+)/(?P<token>[\w:_-]+)/$',
|
||||
description=_('Upload a single document file'),
|
||||
parameters={
|
||||
'id_piece': PIECE_ID_PARAM,
|
||||
'token': UPLOAD_TOKEN_PARAM,
|
||||
})
|
||||
def upload(self, request, id_piece, token, **kwargs):
|
||||
signer = Signer(salt='cart@ds_cs')
|
||||
tracking_code = signer.unsign(token)
|
||||
file_upload = CartaDSFile(
|
||||
tracking_code=tracking_code,
|
||||
id_piece=id_piece,
|
||||
uploaded_file=request.FILES['files[]'])
|
||||
file_upload.save()
|
||||
return [{'name': os.path.basename(file_upload.uploaded_file.name),
|
||||
'token': signer.sign(str(file_upload.id))}]
|
||||
|
||||
@endpoint(methods=['post'],
|
||||
name='upload',
|
||||
pattern='^(?P<id_piece>[\w-]+)/(?P<token>[\w:_-]+)/(?P<file_upload>[\w:_-]+)/delete/$',
|
||||
description=_('Delete a single document file'),
|
||||
parameters={
|
||||
'id_piece': PIECE_ID_PARAM,
|
||||
'token': UPLOAD_TOKEN_PARAM,
|
||||
'file_upload': {
|
||||
'description': _('Signed identifier of single document upload'),
|
||||
},
|
||||
})
|
||||
def upload_delete(self, request, id_piece, token, file_upload, **kwargs):
|
||||
# this cannot be verb DELETE as we have no way to set
|
||||
# Access-Control-Allow-Methods
|
||||
signer = Signer(salt='cart@ds_cs')
|
||||
tracking_code = signer.unsign(token)
|
||||
CartaDSFile.objects.filter(id=signer.unsign(file_upload)).delete()
|
||||
return {'err': 0}
|
||||
|
||||
@endpoint(perm='can_access',
|
||||
description=_('Validate and send a file'),
|
||||
parameters={
|
||||
'commune_id': COMMUNE_ID_PARAM,
|
||||
'type_dossier_id': TYPE_DOSSIER_ID_PARAM,
|
||||
'objet_demande_id': OBJET_DEMANDE_ID_PARAM,
|
||||
'tracking_code': TRACKING_CODE_PARAM,
|
||||
'email': {
|
||||
'description': _('Email of requester'),
|
||||
},
|
||||
})
|
||||
def send(self, request, commune_id, type_dossier_id, objet_demande_id, tracking_code, email):
|
||||
dossier = CartaDSDossier(
|
||||
commune_id=commune_id,
|
||||
type_dossier_id=type_dossier_id,
|
||||
objet_demande_id=objet_demande_id,
|
||||
tracking_code=tracking_code,
|
||||
email=email,
|
||||
)
|
||||
dossier.save()
|
||||
signer = Signer(salt='cart@ds_cs/dossier')
|
||||
dossier.notification_url = request.build_absolute_uri(
|
||||
reverse('generic-endpoint', kwargs={
|
||||
'connector': 'cartads-cs',
|
||||
'slug': self.slug,
|
||||
'endpoint': 'notification'})) + '/%s/' % signer.sign(str(dossier.id))
|
||||
dossier.save()
|
||||
self.add_job('pack', dossier_id=dossier.id)
|
||||
return {'err': 0, 'dossier_id': dossier.id}
|
||||
|
||||
def pack(self, dossier_id):
|
||||
dossier = CartaDSDossier.objects.get(id=dossier_id)
|
||||
zip_filename = os.path.join(default_storage.path('cartads_cs'), '%s.zip' % dossier.tracking_code)
|
||||
zip_file = zipfile.ZipFile(zip_filename, mode='w')
|
||||
liste_pdf = self.liste_pdf(None, dossier.type_dossier_id)
|
||||
cerfa_id = liste_pdf['data'][0]['id'].replace('*', '-')
|
||||
pieces = self.pieces(None, dossier.type_dossier_id, dossier.objet_demande_id, dossier.tracking_code)
|
||||
for piece in pieces['data']:
|
||||
cnt = 1
|
||||
for file in piece['files']:
|
||||
if not file.get('id'):
|
||||
continue
|
||||
cartads_file = CartaDSFile.objects.get(id=file['id'])
|
||||
if piece['id'] == 'cerfa-%s-%s' % (dossier.type_dossier_id, dossier.objet_demande_id):
|
||||
zip_file.write(
|
||||
cartads_file.uploaded_file.path,
|
||||
'%s.pdf' % cerfa_id)
|
||||
elif piece['id'].startswith('cerfa-autres-'):
|
||||
zip_file.write(
|
||||
cartads_file.uploaded_file.path,
|
||||
'Fiches_complementaires/Cerfa_autres_demandeurs_%d.pdf' % cnt)
|
||||
else:
|
||||
zip_file.write(
|
||||
cartads_file.uploaded_file.path,
|
||||
'Pieces/%s-%s%s%s' % (
|
||||
piece['id'],
|
||||
piece['codePiece'],
|
||||
cnt,
|
||||
os.path.splitext(cartads_file.uploaded_file.path)[-1]))
|
||||
cnt += 1
|
||||
zip_file.close()
|
||||
dossier.zip_ready = True
|
||||
dossier.save()
|
||||
self.add_job('send_to_cartads', dossier_id=dossier.id)
|
||||
|
||||
def send_to_cartads(self, dossier_id):
|
||||
dossier = CartaDSDossier.objects.get(id=dossier_id)
|
||||
zip_filename = os.path.join(default_storage.path('cartads_cs'), '%s.zip' % dossier.tracking_code)
|
||||
|
||||
ftp = FTP(self.ftp_server)
|
||||
ftp.login(self.ftp_username, self.ftp_password)
|
||||
ftp.cwd(self.ftp_client_name)
|
||||
ftp.storbinary(
|
||||
'STOR %s' % os.path.basename(zip_filename),
|
||||
open(zip_filename))
|
||||
ftp.quit()
|
||||
|
||||
def key_value_of_stringstring(d):
|
||||
return {'KeyValueOfstringstring': [{'Key': x, 'Value': y} for x, y in d.items()]}
|
||||
|
||||
client = self.soap_client()
|
||||
resp = client.service.NotifierDepotDossier(
|
||||
self.get_token(),
|
||||
dossier.commune_id,
|
||||
dossier.type_dossier_id,
|
||||
os.path.basename(zip_filename),
|
||||
dossier.email,
|
||||
key_value_of_stringstring(
|
||||
{
|
||||
'TraitementImmediat': 1,
|
||||
'UrlNotification': dossier.notification_url,
|
||||
}))
|
||||
dossier.zip_sent = True
|
||||
dossier.zip_ack_response = str(resp)
|
||||
dossier.save()
|
||||
|
||||
@endpoint(pattern='^(?P<signed_dossier_id>[\w:_-]+)/$',
|
||||
methods=['post'],
|
||||
description=_('Notification of file processing by Cart@DS CS'),
|
||||
parameters={
|
||||
'signed_dossier_id': {
|
||||
'description': _('Signed identifier of file')
|
||||
},
|
||||
})
|
||||
def notification(self, request, signed_dossier_id):
|
||||
signer = Signer(salt='cart@ds_cs/dossier')
|
||||
dossier_id = signer.unsign(signed_dossier_id)
|
||||
dossier = CartaDSDossier.objects.get(id=dossier_id)
|
||||
dossier.notification_message = request.body
|
||||
notification = etree.fromstring(request.POST['notification'])
|
||||
dossier.cartads_id_dossier = notification.find('InformationsComplementaires/IdDossierCartads').text
|
||||
dossier.cartads_numero_dossier = notification.find('InformationsComplementaires/NumeroDossier').text
|
||||
dossier.save()
|
||||
return HttpResponse('ok', content_type='text/plain')
|
||||
|
||||
@endpoint(perm='can_access',
|
||||
description=_('Get status of file'),
|
||||
parameters={
|
||||
'dossier_id': {
|
||||
'description': _('Identifier of file'),
|
||||
}
|
||||
})
|
||||
def status(self, request, dossier_id):
|
||||
dossier = CartaDSDossier.objects.get(id=dossier_id)
|
||||
extra = None
|
||||
if dossier.cartads_id_dossier:
|
||||
client = self.soap_client(wsdl_url=self.get_wsdl_url('ServiceEtapeDossier'))
|
||||
resp = client.service.GetEtapesDossier(self.get_token(),
|
||||
dossier.cartads_id_dossier, [])
|
||||
steps = []
|
||||
for step in resp:
|
||||
steps.append(step)
|
||||
steps.sort(key=lambda x: x['DateReference'])
|
||||
status_id = 'cartads-%s' % steps[-1]['IdEtape']
|
||||
status_label = steps[-1]['LibelleEtape']
|
||||
extra = {}
|
||||
for key in steps[-1].keys():
|
||||
extra[key] = steps[-1][key]
|
||||
elif dossier.zip_sent:
|
||||
status_id = 'zip-sent'
|
||||
status_label = _('File sent')
|
||||
elif dossier.zip_ready:
|
||||
status_id = _('File ready to be sent')
|
||||
else:
|
||||
status_label = 'pending'
|
||||
return {'status_id': status_id, 'status_label': status_label, 'extra': extra}
|
||||
|
|
@ -122,6 +122,7 @@ INSTALLED_APPS = (
|
|||
'passerelle.apps.arcgis',
|
||||
'passerelle.apps.base_adresse',
|
||||
'passerelle.apps.bdp',
|
||||
'passerelle.apps.cartads_cs',
|
||||
'passerelle.apps.choosit',
|
||||
'passerelle.apps.cityweb',
|
||||
'passerelle.apps.clicrdv',
|
||||
|
|
|
|||
3
setup.py
3
setup.py
|
|
@ -105,7 +105,8 @@ setup(name='passerelle',
|
|||
'Pillow',
|
||||
'python-magic',
|
||||
'jsonschema',
|
||||
'zeep < 3.0'
|
||||
'zeep < 3.0',
|
||||
'pycrypto',
|
||||
],
|
||||
cmdclass={
|
||||
'build': build,
|
||||
|
|
|
|||
|
|
@ -0,0 +1,210 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import datetime
|
||||
|
||||
import mock
|
||||
import pytest
|
||||
|
||||
from django.core.files.storage import default_storage
|
||||
|
||||
from passerelle.apps.cartads_cs.models import CartaDSCS, CartaDSFile, CartaDSDossier
|
||||
from passerelle.base.models import Job
|
||||
|
||||
from . import utils
|
||||
|
||||
@pytest.fixture
|
||||
def connector(db):
|
||||
return utils.make_resource(CartaDSCS,
|
||||
title='Test',
|
||||
slug='test',
|
||||
description='...',
|
||||
wsdl_base_url='http://test.invalid/adscs/webservices/',
|
||||
username='test',
|
||||
password='test',
|
||||
iv='x'*16,
|
||||
secret_key='y'*16,
|
||||
ftp_server='ftp.invalid',
|
||||
ftp_username='test',
|
||||
ftp_password='test',
|
||||
ftp_client_name='test'
|
||||
)
|
||||
|
||||
|
||||
class FakeService():
|
||||
def GetCommunes(self, token):
|
||||
return [{'Key': 2, 'Value': 'AIGREFEUILLE SUR MAINE'}]
|
||||
|
||||
def GetTypesDossier(self, token, commune_id):
|
||||
return [{'Key': 'CU', 'Value': "Certificat d'urbanisme"}]
|
||||
|
||||
def GetObjetsDemande(self, token, type_dossier_id):
|
||||
return [{'Key': 1, 'Value': "CU d'information"}]
|
||||
|
||||
def GetListePdf(self, token, type_dossier_id, options=None):
|
||||
return [{'UrlTelechargement': 'https://invalid/adscs/webservices/ServicePDF.ashx?pdf=13410*04',
|
||||
'Nom': 'Cerfa 13410-04',
|
||||
'Description': "Demande de Certificat d'urbanisme",
|
||||
'Identifiant': '13410*04'}]
|
||||
|
||||
def GetPieces(self, token, type_dossier_id, objet_demande_id):
|
||||
return [{'IdPiece': 1065,
|
||||
'Libelle': 'DECLARATION PREALABLE INCOMPLETE',
|
||||
'CodePiece': 'CU',
|
||||
'Descriptif': 'Complétez la rubrique',
|
||||
'Reglementaire': False,
|
||||
},
|
||||
{'IdPiece': 1,
|
||||
'Libelle': 'Plan de situation du terrain',
|
||||
'CodePiece': 'CU01',
|
||||
'Descriptif': 'Un plan de situation du terrain [Art. R. 410-1 al 1 du code de l\'urbanisme]',
|
||||
'Reglementaire': True,
|
||||
}
|
||||
]
|
||||
|
||||
def NotifierDepotDossier(self, token, commune_id, type_dossier_id, filename, email, infos):
|
||||
return 'True'
|
||||
|
||||
def GetEtapesDossier(self, token, dossier_id, infos):
|
||||
return [{
|
||||
'DateEcheance': datetime.datetime(2019, 3, 1, 0, 0),
|
||||
'DateRealisation': None,
|
||||
'DateReference': datetime.datetime(2019, 2, 14, 0, 0),
|
||||
'IdDossier': 135792,
|
||||
'IdEtape': 1,
|
||||
'IdEtapeDossier': 692232,
|
||||
'LibelleEtape': 'En cours de saisie'
|
||||
}]
|
||||
|
||||
|
||||
def test_communes(connector, app):
|
||||
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
|
||||
client.return_value = mock.Mock(service=FakeService())
|
||||
resp = app.get('/cartads-cs/test/communes')
|
||||
assert resp.json == {'data': [{'text': 'AIGREFEUILLE SUR MAINE', 'id': '2'}], 'err': 0}
|
||||
|
||||
def test_types_dossier(connector, app):
|
||||
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
|
||||
client.return_value = mock.Mock(service=FakeService())
|
||||
resp = app.get('/cartads-cs/test/types_dossier', status=400)
|
||||
resp = app.get('/cartads-cs/test/types_dossier?commune_id=1')
|
||||
assert resp.json == {'data': [{'id': 'CU', 'text': "Certificat d'urbanisme"}], 'err': 0}
|
||||
|
||||
def test_objets_demande(connector, app):
|
||||
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
|
||||
client.return_value = mock.Mock(service=FakeService())
|
||||
resp = app.get('/cartads-cs/test/objets_demande?type_dossier_id=CU')
|
||||
assert resp.json == {'data': [{'id': '1', 'text': "CU d'information"}], 'err': 0}
|
||||
|
||||
def test_liste_pdf(connector, app):
|
||||
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
|
||||
client.return_value = mock.Mock(service=FakeService())
|
||||
resp = app.get('/cartads-cs/test/liste_pdf?type_dossier_id=CU')
|
||||
assert resp.json == {'data': [{'id': '13410*04',
|
||||
'text': "Cerfa 13410-04: Demande de Certificat d'urbanisme",
|
||||
'url': 'https://invalid/adscs/webservices/ServicePDF.ashx?pdf=13410*04'}],
|
||||
'err': 0}
|
||||
|
||||
def test_pieces_management(connector, app):
|
||||
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
|
||||
client.return_value = mock.Mock(service=FakeService())
|
||||
resp = app.get('/cartads-cs/test/pieces?type_dossier_id=CU&objet_demande_id=1&tracking_code=BBBBBBBB')
|
||||
data = resp.json['data']
|
||||
assert len(data) == 4
|
||||
assert data[0]['text'] == 'Cerfa rempli'
|
||||
assert data[0]['max_files'] == 1
|
||||
assert data[1]['text'] == u'Cerfa demandeurs complémentaires'
|
||||
assert data[1]['max_files'] == 6
|
||||
assert data[2]['text'] == 'Plan de situation du terrain'
|
||||
assert data[2]['max_files'] == 6
|
||||
assert data[3]['text'] == 'DECLARATION PREALABLE INCOMPLETE'
|
||||
assert data[3]['max_files'] == 6
|
||||
for piece in data:
|
||||
assert len(piece['files']) == 1
|
||||
assert piece['files'][0].keys() == ['url']
|
||||
|
||||
resp = app.get('/cartads-cs/test/check_pieces?type_dossier_id=CU&objet_demande_id=1&tracking_code=BBBBBBBB')
|
||||
assert resp.json == {'result': False, 'err': 0}
|
||||
|
||||
resp = app.post(data[0]['files'][0]['url'],
|
||||
upload_files=[('files[]', 'test.pdf', '%PDF...')])
|
||||
cerfa_token = resp.json[0]['token']
|
||||
|
||||
resp = app.get('/cartads-cs/test/pieces?type_dossier_id=CU&objet_demande_id=1&tracking_code=BBBBBBBB')
|
||||
data = resp.json['data']
|
||||
assert data[0]['files'][0]['name']
|
||||
|
||||
resp = app.post(data[0]['files'][0]['url'] + '%s/delete/' % cerfa_token)
|
||||
assert resp.json == {'err': 0}
|
||||
|
||||
resp = app.get('/cartads-cs/test/pieces?type_dossier_id=CU&objet_demande_id=1&tracking_code=BBBBBBBB')
|
||||
data = resp.json['data']
|
||||
assert 'name' not in data[0]['files'][0]
|
||||
|
||||
resp = app.post(data[0]['files'][0]['url'],
|
||||
upload_files=[('files[]', 'test.pdf', '%PDF...')])
|
||||
|
||||
resp = app.post(data[1]['files'][0]['url'],
|
||||
upload_files=[('files[]', 'test.pdf', '%PDF...')])
|
||||
resp = app.post(data[1]['files'][0]['url'],
|
||||
upload_files=[('files[]', 'test.pdf', '%PDF...')])
|
||||
resp = app.get('/cartads-cs/test/pieces?type_dossier_id=CU&objet_demande_id=1&tracking_code=BBBBBBBB')
|
||||
data = resp.json['data']
|
||||
assert len(data[1]['files']) == 3
|
||||
|
||||
resp = app.get('/cartads-cs/test/check_pieces?type_dossier_id=CU&objet_demande_id=1&tracking_code=BBBBBBBB')
|
||||
assert resp.json == {'result': False, 'err': 0}
|
||||
|
||||
resp = app.post(data[2]['files'][0]['url'],
|
||||
upload_files=[('files[]', 'test.pdf', '%PDF...')])
|
||||
|
||||
resp = app.get('/cartads-cs/test/check_pieces?type_dossier_id=CU&objet_demande_id=1&tracking_code=BBBBBBBB')
|
||||
assert resp.json == {'result': True, 'err': 0}
|
||||
|
||||
def test_send(connector, app):
|
||||
CartaDSFile.objects.all().delete()
|
||||
Job.objects.all().delete()
|
||||
test_pieces_management(connector, app)
|
||||
resp = app.get('/cartads-cs/test/send?commune_id=1&type_dossier_id=CU&objet_demande_id=1&tracking_code=BBBBBBBB&email=test@invalid')
|
||||
assert CartaDSDossier.objects.all().count() == 1
|
||||
dossier = CartaDSDossier.objects.all().first()
|
||||
assert resp.json['dossier_id'] == dossier.id
|
||||
assert Job.objects.all().count() == 1
|
||||
|
||||
# test_pack
|
||||
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
|
||||
client.return_value = mock.Mock(service=FakeService())
|
||||
with mock.patch('passerelle.apps.cartads_cs.models.FTP') as FTP:
|
||||
connector.jobs()
|
||||
assert Job.objects.filter(method_name='pack', status='completed').count()
|
||||
assert Job.objects.filter(method_name='send_to_cartads', status='completed').count()
|
||||
|
||||
dossier = CartaDSDossier.objects.get(id=dossier.id)
|
||||
assert dossier.zip_ack_response == 'True'
|
||||
|
||||
resp = app.post(dossier.notification_url, params={'notification': '''<?xml version="1.0" encoding="utf-8"?>
|
||||
<Notification xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
|
||||
<NomArchive>SKTJCMPD.zip</NomArchive>
|
||||
<DateDepot>2019-02-14T00:00:00</DateDepot>
|
||||
<EmailDemandeur>test@invalid</EmailDemandeur>
|
||||
<Succes>true</Succes>
|
||||
<InformationsComplementaires>
|
||||
<Etape>PriseEnChargeAutoTerminee</Etape>
|
||||
<MessageErreur />
|
||||
<IdDossierCartads>135792</IdDossierCartads>
|
||||
<NumeroDossier>CU 044 043 19 A0006</NumeroDossier>
|
||||
</InformationsComplementaires>
|
||||
</Notification>'''})
|
||||
|
||||
dossier = CartaDSDossier.objects.get(id=dossier.id)
|
||||
assert dossier.cartads_id_dossier == '135792'
|
||||
assert dossier.cartads_numero_dossier == 'CU 044 043 19 A0006'
|
||||
|
||||
def test_status(connector, app):
|
||||
CartaDSDossier.objects.all().delete()
|
||||
test_send(connector, app)
|
||||
dossier = CartaDSDossier.objects.all()[0]
|
||||
|
||||
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
|
||||
client.return_value = mock.Mock(service=FakeService())
|
||||
resp = app.get('/cartads-cs/test/status?dossier_id=%s' % dossier.id)
|
||||
assert resp.json['status_label'] == 'En cours de saisie'
|
||||
Loading…
Reference in New Issue