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

739 lines
30 KiB
Python
Raw Permalink Blame History

import datetime
import json
import os
from collections import OrderedDict
from unittest import mock
import pytest
from django.test import override_settings
from httmock import HTTMock
import tests.utils
from passerelle.apps.cartads_cs.models import CartaDSCS, CartaDSDossier, CartaDSFile
from passerelle.base.models import Job
@pytest.fixture
def connector(db):
return tests.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, options=None):
return [{'Key': 2, 'Value': 'AIGREFEUILLE SUR MAINE'}]
def GetTypesDossier(self, token, commune_id, options=None):
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,
},
{
'IdPiece': 62,
'Libelle': 'Plan de masse des constructions à démolir',
'Descriptif': 'Un plan de masse des constructions...',
'CodePiece': 'PCA1',
'Reglementaire': False,
},
]
def NotifierDepotDossier(self, token, commune_id, type_dossier_id, filename, email, infos):
return 'True'
def GetEtapesDossier(self, token, dossier_id, infos):
try:
dossier = CartaDSDossier.objects.get(cartads_id_dossier=dossier_id)
except CartaDSDossier.DoesNotExist:
dossier = None
if dossier and dossier.tracking_code == 'DOCXXXX':
return [
{
'LibelleEtape': 'Attente DOC',
'IdEtapeDossier': 2356473,
'DateEcheance': '2022-04-01T00:00:00',
'IdEtape': 12,
'DateRealisation': None,
'DateReference': '2019-03-25T00:00:00',
'IdDossier': 471160,
},
]
elif dossier and dossier.tracking_code == 'DAACTXXXX':
return [
{
'LibelleEtape': 'Attente DAACT',
'IdEtapeDossier': 2356473,
'DateEcheance': '2022-04-01T00:00:00',
'IdEtape': 12,
'DateRealisation': None,
'DateReference': '2019-03-25T00:00:00',
'IdDossier': 471160,
},
]
# default
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 GetPiecesDossierACompleter(self, token, dossier_id):
return [
OrderedDict(
[
('CodePiece', 'PC07'),
('DateDemande', datetime.datetime(2019, 4, 15, 0, 0)),
('DatePresentation', None),
('DateReception', None),
('Descriptif', 'Un document graphique...'),
('IdDosPiece', 133837),
('IdPiece', 58),
('LibellePiece', 'Document graphique permettant...'),
('NbDocuments', 0),
]
),
OrderedDict(
[
('CodePiece', 'PC16-1'),
('DateDemande', datetime.datetime(2019, 4, 15, 0, 0)),
('DatePresentation', None),
('DateReception', None),
('Descriptif', 'Formulaire attestant...'),
('IdDosPiece', 133840),
('IdPiece', 99),
('LibellePiece', 'Formulaire attestant...'),
('NbDocuments', 0),
]
),
]
def GetPiecesDaact(self, token, dossier_id):
return [
OrderedDict(
[
('CodePiece', 'AT1'),
('DateDemande', None),
('DatePresentation', None),
('DateReception', None),
('Descriptif', "L'attestation constatant..."),
('IdDosPiece', 0),
('IdPiece', 191),
('LibellePiece', 'Attestation constat des travaux'),
('NbDocuments', 0),
]
),
OrderedDict(
[
('CodePiece', 'AT2'),
('DateDemande', None),
('DatePresentation', None),
('DateReception', None),
('Descriptif', 'Dans les cas...'),
('IdDosPiece', 0),
('IdPiece', 192),
('LibellePiece', 'Document du...'),
('NbDocuments', 0),
]
),
]
def UploadFile(self, FileByteStream, _soapheaders):
assert FileByteStream
assert _soapheaders
def ActiverServiceSuiviNumerique(self, toke, dossier_number, dossier_password):
return 123
def GetInfosDossier(self, token, id_dossier):
assert id_dossier in (123, '135792')
return OrderedDict(
[
('AdresseTerrain', 'all\xe9e des Fleurs'),
('CoTypeDossier', 'PC'),
('Commune', 'AIGREFEUILLE SUR MAINE'),
('DateDepot', datetime.datetime(2019, 9, 19, 0, 0)),
('IdDossier', 478864),
('NomDossier', 'PC 069 085 19 00010'),
('TypeDossier', 'Permis de construire'),
]
)
def GetMotPasse(self, token, id_dossier):
return 'D8B912CE-2A0C-4504-AE3B-74F2EF6BABA6'
def pdf_mock(url, request):
return {'content': b'%PDF...', 'status_code': 200}
@pytest.fixture
def cached_data(connector, app):
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
client.return_value = mock.Mock(service=FakeService())
with HTTMock(pdf_mock):
connector.daily()
def test_communes(connector, app, cached_data):
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, cached_data):
resp = app.get('/cartads-cs/test/types_dossier', status=400)
resp = app.get('/cartads-cs/test/types_dossier?commune_id=2')
assert resp.json == {'data': [{'id': 'CU', 'text': "Certificat d'urbanisme"}], 'err': 0}
resp = app.get('/cartads-cs/test/types_dossier?commune_id=2&filter=CU')
assert resp.json == {'data': [{'id': 'CU', 'text': "Certificat d'urbanisme"}], 'err': 0}
resp = app.get('/cartads-cs/test/types_dossier?commune_id=2&filter=AT')
assert resp.json == {'data': [], 'err': 0}
def test_objets_demande(connector, app, cached_data):
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, cached_data):
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': 'http://testserver/media/public/cartads_cs/test/documents/cerfa_13410-04.pdf',
}
],
'err': 0,
}
def test_pieces_management(connector, app, cached_data):
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) == 5
assert data[0]['text'] == 'Cerfa rempli'
assert data[0]['max_files'] == 1
assert data[1]['text'] == '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 list(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=[])
assert resp.json == []
resp = app.post(data[0]['files'][0]['url'], upload_files=[('foobar', 'test.pdf', b'%PDF...')])
assert resp.json == []
resp = app.post(data[0]['files'][0]['url'], upload_files=[('files[]', 'test.pdf', b'%PDF...')])
assert resp.json == [{'error': 'The CERFA should be a PDF file.'}]
with open(os.path.join(os.path.dirname(__file__), 'data', 'minimal.pdf'), 'rb') as fd:
pdf_contents = fd.read()
resp = app.post(data[0]['files'][0]['url'], upload_files=[('files[]', 'test.pdf', pdf_contents)])
assert resp.json == [{'error': 'The CERFA should not be a scanned document.'}]
with open(os.path.join(os.path.dirname(__file__), 'data', 'pdf-form.pdf'), 'rb') as fd:
pdf_contents = fd.read()
resp = app.post(data[0]['files'][0]['url'], upload_files=[('files[]', 'test.pdf', pdf_contents)])
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.get('/cartads-cs/test/pieces?type_dossier_id=CU&objet_demande_id=1&tracking_code=BBBBBBBB')
assert [x['id'] for x in resp.json['data']] == ['cerfa-CU-1', 'cerfa-autres-CU-1', '1', '1065', '62']
resp = app.get(
'/cartads-cs/test/pieces?type_dossier_id=CU&objet_demande_id=1&tracking_code=BBBBBBBB&demolitions=true'
)
assert [x['id'] for x in resp.json['data']] == ['cerfa-CU-1', 'cerfa-autres-CU-1', '1', '1065', '62']
resp = app.get(
'/cartads-cs/test/pieces?type_dossier_id=CU&objet_demande_id=1&tracking_code=BBBBBBBB&demolitions=false'
)
assert [x['id'] for x in resp.json['data']] == ['cerfa-CU-1', 'cerfa-autres-CU-1', '1', '1065']
resp = app.post(data[0]['files'][0]['url'], upload_files=[('files[]', 'test.pdf', pdf_contents)])
resp = app.post(data[1]['files'][0]['url'], upload_files=[('files[]', 'test.pdf', pdf_contents)])
resp = app.post(data[1]['files'][0]['url'], upload_files=[('files[]', 'test.pdf', pdf_contents)])
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_contents)])
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}
resp = app.post(data[2]['files'][0]['url'], upload_files=[('files[]', 'test.jpeg', b'...')])
assert resp.json[0]['token']
resp = app.post(data[2]['files'][0]['url'], upload_files=[('files[]', 'test.gif', b'...')])
assert resp.json == [{'error': 'The file should be a PDF document or a JPEG image.'}]
def test_send(connector, app, cached_data):
CartaDSFile.objects.all().delete()
Job.objects.all().delete()
test_pieces_management(connector, app, cached_data)
resp = app.get(
'/cartads-cs/test/send?commune_id=2&type_dossier_id=CU&objet_demande_id=1&tracking_code=BBBBBBBB&email=test@invalid'
)
CartaDSDossier.objects.all().delete()
Job.objects.all().delete()
resp = app.get(
'/cartads-cs/test/send?commune_id=2&type_dossier_id=CU'
'&objet_demande_id=1&tracking_code=BBBBBBBB&email=test@invalid&name_id=1234'
)
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'):
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_send_notification_error(connector, app, cached_data):
CartaDSFile.objects.all().delete()
Job.objects.all().delete()
test_pieces_management(connector, app, cached_data)
resp = app.get(
'/cartads-cs/test/send?commune_id=2&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'):
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>DJWQWLNZ.zip</NomArchive>
<DateDepot>2019-05-09T14:50:16.516718</DateDepot>
<EmailDemandeur>test@localhost</EmailDemandeur>
<Succes>false</Succes>
<InformationsComplementaires>
<Etape>ArchiveExtraite</Etape>
<MessageErreur>Le modèle sélectionné ne correspond à aucun Cerfa géré par l'application.</MessageErreur>
<IdDossierCartads />
<IdDossierExterne />
<NumeroDossier />
</InformationsComplementaires>
</Notification>'''
},
)
dossier = CartaDSDossier.objects.get(id=dossier.id)
assert dossier.cartads_id_dossier is None
assert dossier.notification_message is not None
def test_status(connector, app, cached_data):
CartaDSDossier.objects.all().delete()
test_send(connector, app, cached_data)
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'
assert resp.json['cartads_reference_dossier'] == 'CU 044 043 19 A0006'
# test it got cached
resp = app.get('/cartads-cs/test/status?dossier_id=%s' % dossier.id)
assert resp.json['status_label'] == 'En cours de saisie'
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
client.return_value = mock.Mock(service=FakeService())
dossier = CartaDSDossier.objects.all()[0]
dossier.cartads_steps_cache = {}
dossier.save()
connector.hourly()
assert CartaDSDossier.objects.get(pk=dossier.id).cartads_steps_cache
assert CartaDSDossier.objects.get(pk=dossier.id).cartads_cache_code_acces
assert CartaDSDossier.objects.get(pk=dossier.id).cartads_cache_infos
def test_status_error(connector, app, cached_data):
CartaDSDossier.objects.all().delete()
test_send_notification_error(connector, app, cached_data)
dossier = CartaDSDossier.objects.all()[0]
resp = app.get('/cartads-cs/test/status?dossier_id=%s' % dossier.id)
assert (
resp.json['status_label']
== "File refused (Le modèle sélectionné ne correspond à aucun Cerfa géré par l'application.)"
)
def test_status_zip_not_considered_error(connector, app, cached_data):
CartaDSDossier.objects.all().delete()
test_send_notification_error(connector, app, cached_data)
dossier = CartaDSDossier.objects.all()[0]
dossier.notification_message = None
dossier.zip_ack_response = 'False'
dossier.save()
resp = app.get('/cartads-cs/test/status?dossier_id=%s' % dossier.id)
assert resp.json['status_label'] == 'File not considered'
def test_additional_pieces_management(connector, app, cached_data):
CartaDSDossier.objects.all().delete()
test_send(connector, app, cached_data)
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/additional_pieces?tracking_code=%s' % dossier.tracking_code)
data = resp.json['data']
assert len(data) == 2
assert data[0]['text'] == 'Document graphique permettant...'
assert data[0]['max_files'] == 6
assert data[1]['text'] == 'Formulaire attestant...'
assert data[1]['max_files'] == 6
for piece in data:
assert len(piece['files']) == 1
assert list(piece['files'][0].keys()) == ['url']
resp = app.post(data[0]['files'][0]['url'], upload_files=[('files[]', 'test.pdf', b'%PDF...')])
assert resp.json[0]['token']
assert CartaDSFile.objects.filter(tracking_code=dossier.tracking_code, sent_to_cartads=None).count() == 1
Job.objects.all().delete()
resp = app.get('/cartads-cs/test/send_additional_pieces?tracking_code=%s' % dossier.tracking_code)
assert Job.objects.all().count() == 1
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
client.return_value = mock.Mock(service=FakeService())
connector.jobs()
assert Job.objects.filter(method_name='send_additional_pieces_to_cartads', status='completed').count()
assert CartaDSFile.objects.filter(tracking_code=dossier.tracking_code, sent_to_cartads=None).count() == 0
def test_doc_pieces_management(connector, app, cached_data):
CartaDSDossier.objects.all().delete()
test_send(connector, app, cached_data)
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/doc_pieces?tracking_code=%s' % dossier.tracking_code)
assert resp.json['err']
dossier.tracking_code = 'DOCXXXX' # to get the right status
dossier.save()
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/doc_pieces?tracking_code=%s' % dossier.tracking_code)
data = resp.json['data']
assert len(data) == 1
assert data[0]['text'] == 'CERFA'
assert data[0]['max_files'] == 1
assert len(data[0]['files']) == 1
assert list(data[0]['files'][0].keys()) == ['url']
with open(os.path.join(os.path.dirname(__file__), 'data', 'pdf-form.pdf'), 'rb') as fd:
pdf_contents = fd.read()
resp = app.post(data[0]['files'][0]['url'], upload_files=[('files[]', 'test.pdf', pdf_contents)])
assert resp.json[0]['token']
assert CartaDSFile.objects.filter(tracking_code=dossier.tracking_code, sent_to_cartads=None).count() == 1
Job.objects.all().delete()
resp = app.get('/cartads-cs/test/send_doc_pieces?tracking_code=%s' % dossier.tracking_code)
assert Job.objects.all().count() == 1
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
client.return_value = mock.Mock(service=FakeService())
connector.jobs()
assert Job.objects.filter(method_name='send_doc_pieces_to_cartads', status='completed').count()
assert CartaDSFile.objects.filter(tracking_code=dossier.tracking_code, sent_to_cartads=None).count() == 0
def test_daact_pieces_management(connector, app, cached_data):
CartaDSDossier.objects.all().delete()
test_send(connector, app, cached_data)
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/daact_pieces?tracking_code=%s' % dossier.tracking_code)
assert resp.json['err']
dossier.tracking_code = 'DAACTXXXX' # to get the right status
dossier.save()
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/daact_pieces?tracking_code=%s' % dossier.tracking_code)
data = resp.json['data']
assert len(data) == 3
assert data[0]['text'] == 'CERFA'
assert data[0]['max_files'] == 1
for piece in data:
assert len(piece['files']) == 1
assert list(piece['files'][0].keys()) == ['url']
with open(os.path.join(os.path.dirname(__file__), 'data', 'pdf-form.pdf'), 'rb') as fd:
pdf_contents = fd.read()
resp = app.post(data[0]['files'][0]['url'], upload_files=[('files[]', 'test.pdf', pdf_contents)])
assert resp.json[0]['token']
assert CartaDSFile.objects.filter(tracking_code=dossier.tracking_code, sent_to_cartads=None).count() == 1
with open(os.path.join(os.path.dirname(__file__), 'data', 'pdf-form.pdf'), 'rb') as fd:
pdf_contents = fd.read()
resp = app.post(data[1]['files'][0]['url'], upload_files=[('files[]', 'test.pdf', pdf_contents)])
assert resp.json[0]['token']
assert CartaDSFile.objects.filter(tracking_code=dossier.tracking_code, sent_to_cartads=None).count() == 2
Job.objects.all().delete()
resp = app.get('/cartads-cs/test/send_daact_pieces?tracking_code=%s' % dossier.tracking_code)
assert Job.objects.all().count() == 1
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
client.return_value = mock.Mock(service=FakeService())
connector.jobs()
assert Job.objects.filter(method_name='send_daact_pieces_to_cartads', status='completed').count()
assert CartaDSFile.objects.filter(tracking_code=dossier.tracking_code, sent_to_cartads=None).count() == 0
def test_list_of_files(connector, app, cached_data):
CartaDSDossier.objects.all().delete()
test_send(connector, app, cached_data)
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
client.return_value = mock.Mock(service=FakeService())
# all for user
resp = app.get('/cartads-cs/test/files?name_id=1234')
assert len(resp.json['data']) == 1
assert resp.json['data'][0]['status']['status_label'] == 'En cours de saisie'
assert resp.json['data'][0]['commune_label'] == 'AIGREFEUILLE SUR MAINE'
assert not resp.json['data'][0]['cartads_infos'] # not in cache
# invalid user
resp = app.get('/cartads-cs/test/files?name_id=5678')
assert len(resp.json['data']) == 0
# existing status
resp = app.get('/cartads-cs/test/files?name_id=1234&status=En cours de saisie')
assert len(resp.json['data']) == 1
# missing status
resp = app.get('/cartads-cs/test/files?name_id=1234&status=Attente DOC')
assert len(resp.json['data']) == 0
# hourly job to cache additional data
with mock.patch('passerelle.apps.cartads_cs.models.CartaDSCS.soap_client') as client:
client.return_value = mock.Mock(service=FakeService())
connector.hourly()
resp = app.get('/cartads-cs/test/files?name_id=1234')
assert len(resp.json['data']) == 1
assert resp.json['data'][0]['cartads_infos']
assert resp.json['data'][0]['type_dossier_label'] == "Certificat d'urbanisme"
def test_join(connector, app, cached_data):
CartaDSDossier.objects.all().delete()
# new
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/join?name_id=3456&dossier_number=123&dossier_password=XXX&formdata_url=https://etc.'
)
dossier = CartaDSDossier.objects.get(id=resp.json['dossier_id'])
assert dossier.commune_id == '2'
assert dossier.type_dossier_id == 'PC'
assert dossier.formdata_url == 'https://etc.'
assert dossier.tracking_code == resp.json['tracking_code']
# existing
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/join?name_id=2345&dossier_number=123&dossier_password=XXX&formdata_url=other'
)
dossier = CartaDSDossier.objects.get(id=resp.json['dossier_id'])
assert CartaDSDossier.objects.count() == 1
assert dossier.commune_id == '2'
assert dossier.type_dossier_id == 'PC'
assert dossier.formdata_url == 'https://etc.'
assert [x.name_id for x in dossier.subscribers.all().order_by('name_id')] == ['2345', '3456']
# check /files API afterwards
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/files?name_id=1234')
assert len(resp.json['data']) == 0
resp = app.get('/cartads-cs/test/files?name_id=2345')
assert len(resp.json['data']) == 1
resp = app.get('/cartads-cs/test/files?name_id=3456')
assert len(resp.json['data']) == 1
def test_unsubscribe(connector, app, cached_data):
test_join(connector, app, cached_data)
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/files?name_id=3456')
assert len(resp.json['data']) == 1
resp = app.get('/cartads-cs/test/unsubscribe?name_id=3456&dossier_number=123')
resp = app.get('/cartads-cs/test/files?name_id=3456')
assert len(resp.json['data']) == 0
# error cases
resp = app.get('/cartads-cs/test/unsubscribe?name_id=3456&dossier_number=123')
assert resp.json['err'] == 1
resp = app.get('/cartads-cs/test/unsubscribe?name_id=4567&dossier_number=123')
assert resp.json['err'] == 1
resp = app.get('/cartads-cs/test/unsubscribe?name_id=4567&dossier_number=434')
assert resp.json['err'] == 1
def test_role_sync(connector, app, cached_data):
def idp_mock(url, request):
assert url.netloc == 'idp.example.org'
if url.path == '/api/roles/':
dossier = CartaDSDossier.objects.all().first()
assert json.loads(request.body) == json.loads(
'{"name": "Suivi Cart@DS (%s)", "slug": "_cartads_%s"}' % (dossier.id, dossier.id)
)
return {'content': json.dumps({'uuid': 'role-uuid'}), 'status_code': 200}
elif url.path == '/api/roles/role-uuid/relationships/members/':
body = json.loads(request.body)
idp_mock.subscribed_roles = {x['uuid'] for x in body['data']}
return {'content': json.dumps({'err': 0}), 'status_code': 200}
raise Exception('unhandled http call (%s)' % url)
with HTTMock(idp_mock), override_settings(
KNOWN_SERVICES={
'authentic': {
'idp': {
'url': 'http://idp.example.org/',
'verif_orig': 'abc',
'secret': 'def',
}
}
}
):
test_join(connector, app, cached_data)
assert idp_mock.subscribed_roles == {'2345', '3456'}
View Git Blame Copy Permalink