1072 lines
46 KiB
Python
1072 lines
46 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
|
|
# This file is part of passerelle-atreal-openads - a Publik connector to openADS
|
|
#
|
|
# Copyright (C) 2019 Atreal
|
|
#
|
|
# 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/>.
|
|
|
|
"""Tests for connector AtrealOpenads."""
|
|
# pylint: disable=too-many-lines
|
|
|
|
import base64
|
|
import datetime
|
|
import json
|
|
import os
|
|
import re
|
|
|
|
import magic
|
|
import mock
|
|
import pytest
|
|
|
|
from requests import Response
|
|
|
|
from django.http import Http404
|
|
from django.http.response import JsonResponse
|
|
from django.core.files import File
|
|
# from django.db.models.query import QuerySet
|
|
from django.core.exceptions import ValidationError
|
|
|
|
from passerelle.utils.jsonresponse import APIError
|
|
from passerelle.base.models import Job
|
|
|
|
from atreal_openads.utils import (
|
|
get_file_data,
|
|
get_file_digest,
|
|
trunc_str_values
|
|
)
|
|
|
|
from atreal_openads.models import (
|
|
ForwardFile,
|
|
Collectivite,
|
|
)
|
|
|
|
|
|
def build_forwardfile_from_path(connecteur, path, numero_dossier, type_fichier):
|
|
"""Convert a file path to a ForwardFile."""
|
|
if path:
|
|
rand_id = base64.urlsafe_b64encode(os.urandom(6))
|
|
fwd_file = ForwardFile()
|
|
fwd_file.connecteur = connecteur
|
|
fwd_file.numero_demande = rand_id
|
|
fwd_file.numero_dossier = numero_dossier
|
|
fwd_file.type_fichier = type_fichier
|
|
fwd_file.orig_filename = os.path.basename(path)
|
|
fwd_file.content_type = magic.from_file(path, mime=True)
|
|
with open(path, 'r') as file_pt:
|
|
fwd_file.file_hash = get_file_digest(file_pt)
|
|
fwd_file.upload_file = File(open(path, 'r'))
|
|
fwd_file.upload_status = 'pending'
|
|
return fwd_file
|
|
return None
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_forward_file(forwardfile_2, atreal_openads):
|
|
"""Test ForwardFile object methods."""
|
|
|
|
forwardfile_2.orig_filename = 'afile'
|
|
forwardfile_2.connecteur = atreal_openads
|
|
|
|
# forwardfile_2.upload_file.save(forwardfile_2.orig_filename, forwardfile_2.upload_file)
|
|
forwardfile_2.save()
|
|
|
|
assert repr(forwardfile_2) == (
|
|
u'ForwardFile(id=%s,connecteur=%s,collectivite=%s'
|
|
',demande=%s,dossier=%s,type=%s,filename=%s,status=%s)' % (
|
|
forwardfile_2.id, unicode(forwardfile_2.connecteur), None, # pylint: disable=no-member
|
|
forwardfile_2.numero_demande, forwardfile_2.numero_dossier,
|
|
forwardfile_2.type_fichier, forwardfile_2.orig_filename, forwardfile_2.upload_status
|
|
)
|
|
).encode('utf-8')
|
|
|
|
assert str(forwardfile_2) == '%s[%s]' % (trunc_str_values(forwardfile_2.orig_filename, 20),
|
|
'Pending')
|
|
assert unicode(forwardfile_2) == u'%s[%s]' % (
|
|
trunc_str_values(forwardfile_2.orig_filename, 20),
|
|
'Pending')
|
|
|
|
assert forwardfile_2.get_status() == 'Pending'
|
|
assert forwardfile_2.get_status('invalid') == 'invalid'
|
|
|
|
params = forwardfile_2.get_url_params()
|
|
assert params['connecteur'] == atreal_openads.slug
|
|
|
|
assert forwardfile_2.upload_file is not None
|
|
assert forwardfile_2.upload_file.size > 0
|
|
assert forwardfile_2.size == forwardfile_2.upload_file.size
|
|
assert forwardfile_2.file_hash == ('cc90a620982760fdee16a5b4fe1b5ac3'
|
|
'b4fe868fd02d2f70b27f1e46d283ea51')
|
|
|
|
forwardfile_2.content_type = 'application/pdf'
|
|
forwardfile_2.upload_status = 'success'
|
|
forwardfile_2.save()
|
|
assert forwardfile_2.upload_status == 'success'
|
|
assert forwardfile_2.get_status() == 'Success'
|
|
assert forwardfile_2.content_type == 'application/pdf'
|
|
|
|
with pytest.raises(ValueError) as exception:
|
|
forwardfile_2.upload_file.size # pylint: disable=pointless-statement
|
|
assert unicode(exception.value) == "The 'upload_file' attribute has no file associated with it."
|
|
assert forwardfile_2.size > 0
|
|
assert forwardfile_2.file_hash == ('cc90a620982760fdee16a5b4fe1b5ac3'
|
|
'b4fe868fd02d2f70b27f1e46d283ea51')
|
|
|
|
forwardfile_2.file_hash = ''
|
|
forwardfile_2.update_file_hash()
|
|
forwardfile_2.update_content_type()
|
|
forwardfile_2.save()
|
|
assert forwardfile_2.file_hash == ''
|
|
assert forwardfile_2.content_type == ''
|
|
|
|
forwardfile_2.orig_filename = ''
|
|
with pytest.raises(ValidationError) as exception:
|
|
forwardfile_2.save()
|
|
assert len(exception.value.messages) == 1
|
|
assert '__all__' in exception.value.message_dict
|
|
assert unicode(exception.value.message_dict['__all__'][0]) == (
|
|
u"A %s cannot have all the following fields empty: %s." % (
|
|
forwardfile_2.get_verbose_name(),
|
|
['file_hash', 'orig_filename', 'upload_file']))
|
|
|
|
forwardfile_2.delete()
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_collectivite(collectivite_1, collectivite_1_guichet):
|
|
"""Test Collectivite object methods."""
|
|
|
|
col = collectivite_1
|
|
|
|
assert repr(col) == (
|
|
u'Collectivite(id=%s,name=%s,connecteur=%s,openADS_id=%s,guichet=%s)' % (
|
|
1, unicode(col.name), unicode(col.connecteur), col.openADS_id,
|
|
unicode(col.guichet) if hasattr(col, 'guichet') else None
|
|
)
|
|
).encode('utf-8')
|
|
|
|
assert str(col) == col.name.encode('utf-8')
|
|
|
|
assert unicode(col) == col.name
|
|
|
|
class_fields = Collectivite.get_fields()
|
|
assert len(class_fields) == 6
|
|
assert class_fields[0].name == 'id'
|
|
assert class_fields[1].name == 'name'
|
|
assert class_fields[2].name == 'connecteur'
|
|
assert class_fields[3].name == 'openADS_id'
|
|
assert class_fields[4].name == 'guichet'
|
|
assert class_fields[5].name == 'forward_file'
|
|
|
|
instance_fields = col.get_fields_kv()
|
|
assert len(instance_fields) == 6
|
|
assert instance_fields[0][0].name == 'id'
|
|
assert instance_fields[1][0].name == 'name'
|
|
assert instance_fields[2][0].name == 'connecteur'
|
|
assert instance_fields[3][0].name == 'openADS_id'
|
|
assert instance_fields[4][0].name == 'guichet'
|
|
assert instance_fields[5][0].name == 'forward_file'
|
|
assert instance_fields[0][1] == col.id
|
|
assert instance_fields[1][1] == col.name
|
|
assert instance_fields[2][1] is col.connecteur
|
|
assert instance_fields[3][1] == col.openADS_id
|
|
assert instance_fields[4][1] is col.guichet
|
|
assert instance_fields[5][1] is None # shouldn't it be QuerySet?
|
|
|
|
params = col.get_url_params()
|
|
assert params['connecteur'] == col.connecteur.slug
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_guichet(collectivite_1_guichet):
|
|
"""Test Guichet object methods."""
|
|
|
|
guichet = collectivite_1_guichet
|
|
|
|
assert repr(guichet) == (
|
|
u'Guichet(id=%s,collectivite=%s,%s)' % (
|
|
1, unicode(guichet.collectivite), unicode(guichet)
|
|
)
|
|
).encode('utf-8')
|
|
|
|
assert str(guichet) == u'Monday 08:30 -> Friday 12:15 [09:00/17:00]'.encode('utf-8')
|
|
|
|
assert unicode(guichet) == u'Monday 08:30 -> Friday 12:15 [09:00/17:00]'
|
|
|
|
params = guichet.get_url_params()
|
|
assert params['collectivite'] == guichet.collectivite.id
|
|
|
|
with pytest.raises(Exception) as exception:
|
|
guichet.get_list_url()
|
|
assert unicode(exception.value) == u"Guichet:get_list_url() method should not be called"
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_guichet_is_open(collectivite_1_guichet): # pylint: disable=too-many-locals
|
|
"""Test the method Guichet.is_open()."""
|
|
|
|
guichet = collectivite_1_guichet
|
|
|
|
dt_fmt = '%Y-%m-%d %H:%M'
|
|
|
|
d_monday = '2019-07-29'
|
|
d_sunday = '2019-07-28'
|
|
d_saturday = '2019-07-27'
|
|
d_friday = '2019-07-26'
|
|
d_thursday = '2019-07-25'
|
|
d_wednesday = '2019-07-24'
|
|
d_tuesday = '2019-07-22'
|
|
t_open = '10:44'
|
|
t_closed_before = '6:33'
|
|
t_closed_after = '20:08'
|
|
|
|
for date_str in [d_monday, d_tuesday, d_wednesday, d_thursday, d_friday]:
|
|
for time_and_cond in [(t_open, True), (t_closed_before, False), (t_closed_after, False)]:
|
|
date_time = datetime.datetime.strptime(date_str + ' ' + time_and_cond[0], dt_fmt)
|
|
assert guichet.is_open(date_time) == time_and_cond[1]
|
|
|
|
date_time = datetime.datetime.strptime(d_friday + ' 16:12', dt_fmt)
|
|
assert not guichet.is_open(date_time)
|
|
|
|
for date_str in [d_saturday, d_sunday]:
|
|
for time_str in [t_open, t_closed_before, t_closed_after]:
|
|
date_time = datetime.datetime.strptime(date_str + ' ' + time_str, dt_fmt)
|
|
assert not guichet.is_open(date_time)
|
|
|
|
with pytest.raises(TypeError) as exception:
|
|
guichet.is_open('invalid datetime')
|
|
assert unicode(exception.value) == u"is_open() expect a datetime object (not a %s)" % type('')
|
|
|
|
assert not guichet.is_open(None)
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_log_json_payload(atreal_openads):
|
|
"""Test the logging of JSON payload."""
|
|
# TODO implement
|
|
assert True
|
|
# change the debug file path
|
|
# check that the function log to it
|
|
# check that what was is logged is correct
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_get_files_from_payload(atreal_openads):
|
|
"""Test the method AtrealOpenads.get_files_from_payload()."""
|
|
|
|
title = 'payload'
|
|
|
|
assert atreal_openads.get_files_from_payload({'files': [{'a': 'file'}]}) == [{'a': 'file'}]
|
|
|
|
with pytest.raises(APIError) as exception:
|
|
atreal_openads.get_files_from_payload({})
|
|
assert unicode(exception.value) == u"Expecting '%s' key in JSON %s" % ('files', title)
|
|
|
|
with pytest.raises(APIError) as exception:
|
|
atreal_openads.get_files_from_payload({'files': 'invalid'})
|
|
assert unicode(exception.value) == u"Expecting '%s' value in JSON %s to be a %s (not a %s)" % (
|
|
'files', title, 'list', type(''))
|
|
|
|
with pytest.raises(APIError) as exception:
|
|
atreal_openads.get_files_from_payload({'files': {'i': 'invalid'}})
|
|
assert unicode(exception.value) == u"Expecting '%s' value in JSON %s to be a %s (not a %s)" % (
|
|
'files', title, 'list', type({}))
|
|
|
|
with pytest.raises(APIError) as exception:
|
|
atreal_openads.get_files_from_payload({'files': []})
|
|
assert unicode(exception.value) == u"Expecting non-empty '%s' value in JSON %s" % (
|
|
'files', title)
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_check_file_dict(fake_conf, atreal_openads):
|
|
"""Test the method AtrealOpenads.check_file_dict()."""
|
|
|
|
title = 'payload'
|
|
|
|
dic = {
|
|
'content': get_file_data(fake_conf['TEST_FILE_CERFA_DIA'], b64=False),
|
|
'filename': os.path.basename(fake_conf['TEST_FILE_CERFA_DIA']),
|
|
'content_type': 'application/pdf'
|
|
}
|
|
|
|
d64 = {
|
|
'b64_content': get_file_data(fake_conf['TEST_FILE_CERFA_DIA'], b64=True),
|
|
'filename': os.path.basename(fake_conf['TEST_FILE_CERFA_DIA']),
|
|
'content_type': 'application/pdf'
|
|
}
|
|
|
|
assert atreal_openads.check_file_dict(dic, b64=False) is None
|
|
assert atreal_openads.check_file_dict(d64) is None
|
|
|
|
dic['filename'] = {'a', 'filename'}
|
|
with pytest.raises(APIError) as exception:
|
|
atreal_openads.check_file_dict(dic, b64=False)
|
|
assert unicode(exception.value) == (
|
|
u"Expecting '%s' value in JSON %s in file dict to be a %s (not a %s)" % (
|
|
'file.filename', title, 'string', type(dic['filename'])))
|
|
|
|
dic['content'] = {'a', 'filename'}
|
|
with pytest.raises(APIError) as exception:
|
|
atreal_openads.check_file_dict(dic, b64=False)
|
|
assert unicode(exception.value) == (
|
|
u"Expecting '%s' value in JSON %s in file dict to be a %s (not a %s)" % (
|
|
'file.content', title, 'string', type(dic['content'])))
|
|
|
|
del dic['content']
|
|
with pytest.raises(APIError) as exception:
|
|
atreal_openads.check_file_dict(dic, b64=False)
|
|
assert unicode(exception.value) == u"Expecting 'file.%s' key in JSON %s" % ('content', title)
|
|
|
|
del d64['b64_content']
|
|
with pytest.raises(APIError) as exception:
|
|
atreal_openads.check_file_dict(dic, b64=True)
|
|
assert unicode(exception.value) == (
|
|
u"Expecting 'file.%s' key in JSON %s" % ('b64_content', title))
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_get_first_file_from_payload(fake_conf, atreal_openads): # pylint: disable=invalid-name
|
|
"""Test the method AtrealOpenads.get_first_file_from_payload()."""
|
|
|
|
title = 'payload'
|
|
|
|
dic = {
|
|
'files': [{
|
|
'content': get_file_data(fake_conf['TEST_FILE_CERFA_DIA'], b64=False),
|
|
'filename': os.path.basename(fake_conf['TEST_FILE_CERFA_DIA']),
|
|
'content_type': 'application/pdf'
|
|
}]
|
|
}
|
|
|
|
assert atreal_openads.get_first_file_from_payload(
|
|
dic, title, ensure_content=True, b64=False) == dic['files'][0]
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_check_status(atreal_openads):
|
|
"""Test the method AtrealOpenads.check_status()."""
|
|
|
|
fake_resp_json = {
|
|
'message': 'Service online'
|
|
}
|
|
fake_resp = JsonResponse(fake_resp_json)
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = mock.Mock(content=fake_resp, status_code=200)
|
|
jresp = atreal_openads.check_status(None)
|
|
assert jresp['response'] == 200
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
# pylint: disable=too-many-statements,too-many-locals
|
|
def test_create_dossier(fake_conf, atreal_openads, collectivite_1, collectivite_1_guichet,
|
|
request_1):
|
|
"""Test the method AtrealOpenads.create_dossier()."""
|
|
|
|
fake_req_json = {
|
|
"fields": {
|
|
|
|
# proprietaire
|
|
"proprietaire": "Non",
|
|
"proprietaire_raw": "Non",
|
|
|
|
# mandataire
|
|
"mandataire_prenom": "John",
|
|
"mandataire_nom": "Man",
|
|
"mandataire_email": "mandataire_email@domain.example",
|
|
|
|
"mandataire_qualite": "Une personne morale",
|
|
"mandataire_qualite_raw": "Une personne morale",
|
|
|
|
"mandataire_denomination": "SELARL",
|
|
"mandataire_raison_sociale": "Super Juriste",
|
|
|
|
"mandataire_numero_voie": "808",
|
|
"mandataire_nom_voie": "Avenue de l'argent",
|
|
"mandataire_lieu_dit": "geoisbour",
|
|
"mandataire_code_postal": "13004",
|
|
"mandataire_localite": "Marseille",
|
|
|
|
# petitionnaire
|
|
"prenom": "Toto",
|
|
"nom": "Loulou",
|
|
"email": "petitionnaire_email@domain.example",
|
|
|
|
"qualite": "Un particulier",
|
|
"qualite_raw": "Un particulier",
|
|
|
|
"numero_voie": "52",
|
|
"nom_voie": "Avenue de la Blaque",
|
|
"lieu_dit": "tierquar",
|
|
"code_postal": "13004",
|
|
"localite": "Marseille",
|
|
|
|
# terrain
|
|
"terrain_numero_voie": "23",
|
|
"terrain_nom_voie": "Boulevard de la République",
|
|
"terrain_lieu_dit": "Leecorne",
|
|
"terrain_code_postal": "13002",
|
|
"terrain_localite": "Marseille",
|
|
|
|
# références cadastrales
|
|
"reference_cadastrale": [["999", "Z", "0010"]],
|
|
"autres_parcelles": True,
|
|
"references_cadastrales": [["123", "D", "9874"]],
|
|
|
|
# user attached files
|
|
"cerfa": {
|
|
"content": get_file_data(fake_conf['TEST_FILE_CERFA_DIA']),
|
|
"content_type": "invalid/content type",
|
|
"field_id": "50",
|
|
"filename": os.path.basename(fake_conf['TEST_FILE_CERFA_DIA'])
|
|
},
|
|
"annexe_1_type_raw": "plan",
|
|
"annexe_1": {
|
|
"content": get_file_data(fake_conf['TEST_FILE_PLAN_CADASTRAL']),
|
|
"content_type": "application/pdf",
|
|
"filename": os.path.basename(fake_conf['TEST_FILE_PLAN_CADASTRAL'])
|
|
},
|
|
"annexe_2_type_raw": "plan",
|
|
"annexe_2": {
|
|
"content": get_file_data(fake_conf['TEST_FILE_PLAN_CADASTRAL']),
|
|
"content_type": "application/pdf",
|
|
# "filename": 'plan_cad'
|
|
},
|
|
"annexe_3_type_raw": "pouvoir",
|
|
"annexe_3": {
|
|
"content": get_file_data(fake_conf['TEST_FILE_CERFA_DIA']),
|
|
"content_type": "application/pdf",
|
|
"filename": 'mandat'
|
|
},
|
|
}
|
|
}
|
|
request_1.method = 'POST'
|
|
request_1._body = json.dumps(fake_req_json) # pylint: disable=protected-access
|
|
|
|
fake_resp_bad = Response()
|
|
fake_resp_bad.status_code = 502
|
|
fake_resp_bad.reason = 'Bad gateway'
|
|
|
|
with pytest.raises(ValueError) as exception:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp_bad
|
|
atreal_openads.create_dossier(request_1, 'DIA', collectivite='not an integer')
|
|
assert unicode(exception.value) == "invalid literal for int() with base 10: 'not an integer'"
|
|
|
|
# guichet is open from Monday/8:30 to Friday/12:15, between 9:00 and 17:00
|
|
now_open = datetime.datetime(2019, 8, 7, 14, 0, 0) # wednesday
|
|
now_closed = datetime.datetime(2019, 8, 10, 16, 0, 0) # saturday
|
|
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp_bad
|
|
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
|
|
now=now_open)
|
|
assert unicode(exception.value) == "HTTP error: 502"
|
|
|
|
# TODO update the code and return message when it will be
|
|
# correctly implemented in the openADS.API side.
|
|
fake_resp_404 = Response()
|
|
fake_resp_404.status_code = 404
|
|
fake_resp_404.reason = 'Page not found'
|
|
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp_404
|
|
atreal_openads.create_dossier(request_1, 'DIA', collectivite=999)
|
|
assert unicode(exception.value) == "HTTP error: 404"
|
|
|
|
jresp = atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
|
|
now=now_closed)
|
|
assert jresp is not None
|
|
assert len(jresp) == 1
|
|
assert 'message' in jresp
|
|
assert jresp['message'] == u"Guichet closed for collectivite '%s'" % collectivite_1
|
|
|
|
now_str_fmt = '%Y-%m-%d %H:%M:%S'
|
|
now_closed_str = now_closed.strftime(now_str_fmt)
|
|
jresp = atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
|
|
now=now_closed_str)
|
|
assert jresp is not None
|
|
assert len(jresp) == 1
|
|
assert 'message' in jresp
|
|
assert jresp['message'] == u"Guichet closed for collectivite '%s'" % collectivite_1
|
|
|
|
now_invalid = {'invalid': 'type'}
|
|
with pytest.raises(APIError) as exception:
|
|
jresp = atreal_openads.create_dossier(request_1, 'DIA',
|
|
collectivite=collectivite_1.openADS_id,
|
|
now=now_invalid)
|
|
assert unicode(exception.value) == (
|
|
u"Invalid value of type '%s' for now argument of endpoint '%s' (must be: %s)" % (
|
|
type(now_invalid),
|
|
'create_dossier',
|
|
"datetime or string formatted to '%s'" % now_str_fmt))
|
|
|
|
annexe_3_type_bak = fake_req_json['fields']['annexe_3_type_raw']
|
|
fake_req_json['fields']['annexe_3_type_raw'] = 'toooooooooooo_looonnnngggggg'
|
|
request_1._body = json.dumps(fake_req_json) # pylint: disable=protected-access
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp_bad
|
|
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
|
|
now=now_open)
|
|
assert unicode(exception.value) == (
|
|
u"Type '%s' for file '%s' is too long (%d chars, but max is %d)" % (
|
|
fake_req_json['fields']['annexe_3_type_raw'],
|
|
'annexe_3',
|
|
len(fake_req_json['fields']['annexe_3_type_raw']),
|
|
10))
|
|
|
|
del fake_req_json['fields']['annexe_3_type_raw']
|
|
request_1._body = json.dumps(fake_req_json) # pylint: disable=protected-access
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp_bad
|
|
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
|
|
now=now_open)
|
|
assert unicode(exception.value) == u"No type field/value for file '%s'" % 'annexe_3'
|
|
|
|
fake_req_json['fields']['annexe_3_type_raw'] = annexe_3_type_bak
|
|
request_1._body = json.dumps(fake_req_json) # pylint: disable=protected-access
|
|
|
|
fake_resp_json = {
|
|
'numero_dossier': fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
'files': [{
|
|
'b64_content': get_file_data(fake_conf['TEST_FILE_CERFA_DIA']),
|
|
'content_type': 'text/plain',
|
|
'filename': 'recepisse_depot_%s.pdf' % fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
}]
|
|
}
|
|
fake_resp = Response()
|
|
fake_resp.status_code = 200
|
|
fake_resp.headers = {'Content-Type': 'application/json'}
|
|
fake_resp.encoding = 'utf-8'
|
|
fake_resp.reason = 'OK'
|
|
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
|
|
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
jresp = atreal_openads.create_dossier(request_1, 'DIA',
|
|
collectivite=collectivite_1.openADS_id,
|
|
now=now_open, type_dossier_detaille='DIA')
|
|
assert jresp['numero_dossier'] == fake_resp_json['numero_dossier']
|
|
assert jresp['recepisse']['b64_content'] == fake_resp_json['files'][0]['b64_content']
|
|
assert jresp['recepisse']['content_type'] == 'application/pdf'
|
|
assert jresp['recepisse']['filename'] == fake_resp_json['files'][0]['filename']
|
|
|
|
fake_resp_json['numero_dossier'] = {'a': 'invalid type'}
|
|
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
|
|
now=now_open)
|
|
assert unicode(exception.value) == (
|
|
u"Expecting '%s' value in JSON response to be a %s (not a %s)" % (
|
|
'numero_dossier', 'string', type({})))
|
|
|
|
del fake_resp_json['numero_dossier']
|
|
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
|
|
now=now_open)
|
|
assert unicode(exception.value) == u"Expecting 'numero_dossier' key in JSON response"
|
|
|
|
fake_resp_json['files'][0]['b64_content'] = 'invalid_;{[content}'
|
|
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
|
|
now=now_open)
|
|
assert unicode(exception.value) == u'Failed to decode recepisse content from base 64'
|
|
|
|
fake_resp._content = 'df[{gfd;g#vfd' # pylint: disable=protected-access
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
|
|
now=now_open)
|
|
# pylint: disable=protected-access
|
|
assert unicode(exception.value) == u'No JSON content returned: %r' % fake_resp._content
|
|
|
|
# pylint: disable=no-member
|
|
job = Job.objects.filter(natural_id=fake_conf['FAKE_NUMERO_DOSSIER']).last()
|
|
assert job
|
|
job_id = job.id
|
|
assert job.status == 'registered'
|
|
assert job.method_name == 'upload_user_files'
|
|
assert job.natural_id == fake_conf['FAKE_NUMERO_DOSSIER']
|
|
assert job.parameters is not None
|
|
assert len(job.parameters) == 4
|
|
assert 'file_ids' in job.parameters
|
|
assert len(job.parameters['file_ids']) == 4
|
|
file_ids = job.parameters['file_ids']
|
|
|
|
forwardfiles = ForwardFile.objects.filter(id__in=file_ids) # pylint: disable=no-member
|
|
for forwardfile in forwardfiles:
|
|
assert forwardfile.numero_demande
|
|
assert forwardfile.numero_dossier == fake_conf['FAKE_NUMERO_DOSSIER']
|
|
assert forwardfile.file_hash
|
|
assert forwardfile.upload_status == 'pending'
|
|
|
|
fake_resp_json = "You want add some files on %s " % fake_conf['FAKE_NUMERO_DOSSIER']
|
|
fake_resp = Response()
|
|
fake_resp.status_code = 200
|
|
fake_resp.headers = {'Content-Type': 'application/json'}
|
|
fake_resp.encoding = 'utf-8'
|
|
fake_resp.reason = 'OK'
|
|
fake_resp._content = json.dumps(fake_resp_json)
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
atreal_openads.jobs()
|
|
|
|
job = Job.objects.get(id=job_id) # pylint: disable=no-member
|
|
assert job.status == 'completed'
|
|
|
|
forwardfiles = ForwardFile.objects.filter(id__in=file_ids) # pylint: disable=no-member
|
|
for forwardfile in forwardfiles:
|
|
assert forwardfile.upload_status == 'success'
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_get_dossier(fake_conf, atreal_openads):
|
|
"""Test the method AtrealOpenads.get_dossier()."""
|
|
|
|
fake_resp_bad = Response()
|
|
fake_resp_bad.status_code = 502
|
|
fake_resp_bad.reason = 'Bad gateway'
|
|
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp_bad
|
|
atreal_openads.get_dossier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'])
|
|
assert unicode(exception.value) == "HTTP error: 502"
|
|
|
|
fake_resp_json = {
|
|
'etat': u"Non préemption en cours",
|
|
'date_depot': "24/04/2019",
|
|
'date_decision': "",
|
|
'decision': "",
|
|
'date_limite_instruction': "24/06/2019"
|
|
}
|
|
fake_resp = Response()
|
|
fake_resp.status_code = 200
|
|
fake_resp.headers = {'Content-Type': 'application/json'}
|
|
fake_resp.encoding = 'utf-8'
|
|
fake_resp.reason = 'OK'
|
|
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp
|
|
jresp = atreal_openads.get_dossier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'])
|
|
assert jresp['etat'] == fake_resp_json['etat']
|
|
assert jresp['date_depot'] == fake_resp_json['date_depot']
|
|
assert jresp['date_decision'] == fake_resp_json['date_decision']
|
|
assert jresp['decision'] == fake_resp_json['decision']
|
|
assert jresp['date_limite_instruction'] == fake_resp_json['date_limite_instruction']
|
|
|
|
fake_resp._content = 'df[{gfd;g#vfd' # pylint: disable=protected-access
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp
|
|
atreal_openads.get_dossier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'])
|
|
# pylint: disable=protected-access
|
|
assert unicode(exception.value) == u'No JSON content returned: %r' % fake_resp._content
|
|
|
|
fake_resp_json = {
|
|
'errors': [{
|
|
'location': 'path',
|
|
'name': 'Invalid Type',
|
|
'description': '"invalid_type" is not one of DIA, PC, DP, AT, PD'
|
|
}]
|
|
}
|
|
fake_resp.status_code = 404
|
|
fake_resp.reason = 'Resource not found'
|
|
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp
|
|
atreal_openads.get_dossier(None, 'invalid_type', fake_conf['FAKE_NUMERO_DOSSIER'])
|
|
assert unicode(exception.value) == (u"HTTP error: 404, [path] (Invalid Type) "
|
|
"\"invalid_type\" is not one of DIA, PC, DP, AT, PD")
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_get_fwd_files(fake_conf, atreal_openads):
|
|
"""Test the method AtrealOpenads.get_fwd_files()."""
|
|
|
|
with pytest.raises(APIError) as exception:
|
|
atreal_openads.get_fwd_files(None, fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
fichier_id='not an integer')
|
|
assert unicode(exception.value) == u"fichier_id must be an integer"
|
|
|
|
with pytest.raises(Http404) as exception:
|
|
atreal_openads.get_fwd_files(None, fake_conf['FAKE_NUMERO_DOSSIER'], fichier_id=18)
|
|
assert unicode(exception.value) == (
|
|
u"No file matches 'numero_dossier=%s' and 'id=%s'." % (fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
18))
|
|
|
|
resp_empty = atreal_openads.get_fwd_files(None, fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
fichier_id=None)
|
|
assert resp_empty is not None
|
|
assert not resp_empty
|
|
|
|
forwardfile = build_forwardfile_from_path(atreal_openads,
|
|
fake_conf['TEST_FILE_CERFA_DIA'],
|
|
fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
'cerfa')
|
|
forwardfile.save()
|
|
assert isinstance(forwardfile, ForwardFile)
|
|
|
|
jresp = atreal_openads.get_fwd_files(None, fake_conf['FAKE_NUMERO_DOSSIER'], fichier_id=None)
|
|
assert jresp is not None
|
|
assert len(jresp) == 1
|
|
assert jresp[0]['id'] == forwardfile.id # pylint: disable=no-member
|
|
for attr in ['numero_dossier', 'type_fichier', 'file_hash', 'orig_filename', 'content_type',
|
|
'upload_status', 'upload_msg']:
|
|
assert jresp[0][attr] == getattr(forwardfile, attr)
|
|
# pylint: disable=no-member
|
|
assert jresp[0]['content_size'] == len(get_file_data(forwardfile.upload_file.path, b64=False))
|
|
assert jresp[0]['last_update_datetime'] == forwardfile.last_update_datetime
|
|
|
|
# pylint: disable=no-member
|
|
jresp = atreal_openads.get_fwd_files(None, fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
fichier_id=forwardfile.id)
|
|
assert jresp is not None
|
|
assert len(jresp) == 1
|
|
assert jresp[0]['id'] == forwardfile.id # pylint: disable=no-member
|
|
for attr in ['numero_dossier', 'type_fichier', 'file_hash', 'orig_filename', 'content_type',
|
|
'upload_status', 'upload_msg']:
|
|
assert jresp[0][attr] == getattr(forwardfile, attr)
|
|
assert jresp[0]['content_size'] == len(get_file_data(forwardfile.upload_file.path, b64=False))
|
|
assert jresp[0]['last_update_datetime'] == forwardfile.last_update_datetime
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_get_fwd_files_status(fake_conf, atreal_openads):
|
|
"""Test the method AtrealOpenads.get_fwd_files_status()."""
|
|
|
|
with pytest.raises(Http404) as exception:
|
|
atreal_openads.get_fwd_files_status(None, fake_conf['FAKE_NUMERO_DOSSIER'], fichier_id=18)
|
|
assert re.search(r"^No file matches 'numero_dossier=[^']+' and 'id=[^']+'.$",
|
|
str(exception.value))
|
|
|
|
forwardfile = build_forwardfile_from_path(atreal_openads,
|
|
fake_conf['TEST_FILE_CERFA_DIA'],
|
|
fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
'cerfa')
|
|
forwardfile.save()
|
|
assert isinstance(forwardfile, ForwardFile)
|
|
|
|
jresp = atreal_openads.get_fwd_files_status(None, fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
fichier_id=None)
|
|
assert jresp is not None
|
|
assert not jresp['all_forwarded']
|
|
# pylint: disable=no-member
|
|
status_msg = '[%s] %s => %s' % (forwardfile.id, forwardfile.orig_filename,
|
|
forwardfile.upload_msg)
|
|
assert len(jresp['pending']) == 1
|
|
assert jresp['pending'][0] == status_msg
|
|
assert not jresp['uploading']
|
|
assert not jresp['success']
|
|
assert not jresp['failed']
|
|
|
|
# pylint: disable=no-member
|
|
jresp = atreal_openads.get_fwd_files_status(None, fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
fichier_id=forwardfile.id)
|
|
assert jresp is not None
|
|
assert not jresp['all_forwarded']
|
|
# pylint: disable=no-member
|
|
status_msg = '[%s] %s => %s' % (forwardfile.id, forwardfile.orig_filename,
|
|
forwardfile.upload_msg)
|
|
assert len(jresp['pending']) == 1
|
|
assert jresp['pending'][0] == status_msg
|
|
assert not jresp['uploading']
|
|
assert not jresp['success']
|
|
assert not jresp['failed']
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_get_courrier(fake_conf, atreal_openads):
|
|
"""Test the method AtrealOpenads.get_courrier()."""
|
|
|
|
lettre_type = 'dia_renonciation_preempter'
|
|
|
|
fake_resp_bad = Response()
|
|
fake_resp_bad.status_code = 502
|
|
fake_resp_bad.reason = 'Bad gateway'
|
|
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp_bad
|
|
atreal_openads.get_courrier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'], lettre_type)
|
|
assert unicode(exception.value) == "HTTP error: 502"
|
|
|
|
fake_resp_json = {
|
|
'files': [{
|
|
'filename': "instruction_4.pdf",
|
|
'content_type': "text/plain",
|
|
'b64_content': get_file_data(fake_conf['TEST_FILE_CERFA_DIA'])
|
|
}]
|
|
}
|
|
fake_resp = Response()
|
|
fake_resp.status_code = 200
|
|
fake_resp.headers = {'Content-Type': 'application/json'}
|
|
fake_resp.encoding = 'utf-8'
|
|
fake_resp.reason = 'OK'
|
|
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp
|
|
jresp = atreal_openads.get_courrier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
lettre_type)
|
|
assert jresp['courrier']['filename'] == fake_resp_json['files'][0]['filename']
|
|
assert jresp['courrier']['content_type'] == fake_resp_json['files'][0]['content_type']
|
|
assert jresp['courrier']['b64_content'] == fake_resp_json['files'][0]['b64_content']
|
|
|
|
fake_resp_json['files'][0]['b64_content'] = 'invalid_;{[content}'
|
|
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp
|
|
atreal_openads.get_courrier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'], lettre_type)
|
|
assert unicode(exception.value) == u'Failed to decode courrier content from base 64'
|
|
|
|
fake_resp._content = 'df[{gfd;g#vfd' # pylint: disable=protected-access
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp
|
|
atreal_openads.get_courrier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'], lettre_type)
|
|
# pylint: disable=protected-access
|
|
assert unicode(exception.value) == u'No JSON content returned: %r' % fake_resp._content
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
def test_get_response_error(atreal_openads):
|
|
"""Test the method AtrealOpenads.get_response_error()."""
|
|
|
|
fake_resp_json = {
|
|
'errors': [
|
|
{
|
|
'location': 'entity.name',
|
|
'name': 'constraint',
|
|
'description': 'Must start with an uppercase letter'
|
|
}
|
|
]
|
|
}
|
|
fake_resp = Response()
|
|
fake_resp.status_code = 404
|
|
fake_resp.headers = {'Content-Type': 'application/json'}
|
|
fake_resp.encoding = 'utf-8'
|
|
fake_resp.reason = 'Not Found'
|
|
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
|
|
|
|
error_msg = atreal_openads.get_response_error(fake_resp)
|
|
expected_msg = u'[%s] (%s) %s' % (
|
|
fake_resp_json['errors'][0]['location'],
|
|
fake_resp_json['errors'][0]['name'],
|
|
fake_resp_json['errors'][0]['description']
|
|
)
|
|
assert error_msg == u"HTTP error: %s, %s" % (fake_resp.status_code, ','.join([expected_msg]))
|
|
|
|
fake_resp._content = 'invalid_;{[content}' # pylint: disable=protected-access
|
|
error_msg = atreal_openads.get_response_error(fake_resp)
|
|
# pylint: disable=protected-access
|
|
assert error_msg == u"HTTP error: %s, %s" % (fake_resp.status_code, fake_resp._content)
|
|
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
# pylint: disable=too-many-statements
|
|
def test_upload_user_files(fake_conf, atreal_openads, request_1):
|
|
"""Test the method AtrealOpenads.upload_user_files()."""
|
|
|
|
request_1.path = '/upload_user_files'
|
|
|
|
with pytest.raises(ForwardFile.DoesNotExist) as exception: # pylint: disable=no-member
|
|
atreal_openads.upload_user_files(request_1, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
file_ids=[999])
|
|
assert unicode(exception.value) == u"The following ForwardFile IDs were not found: %s." % [999]
|
|
|
|
with pytest.raises(ValueError) as exception:
|
|
atreal_openads.upload_user_files(request_1, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
file_ids='invalid string')
|
|
assert unicode(exception.value) == (
|
|
u"invalid literal for int() with base 10: '%s'" % 'invalid string')
|
|
|
|
with pytest.raises(TypeError) as exception:
|
|
atreal_openads.upload_user_files(request_1, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
file_ids={'invalid': 'type'})
|
|
assert unicode(exception.value) == (
|
|
u"Invalid 'file_ids' argument type '%s' (must be string or list)" % (
|
|
type({'invalid': 'type'})))
|
|
|
|
forwardfile = build_forwardfile_from_path(atreal_openads,
|
|
fake_conf['TEST_FILE_CERFA_DIA'],
|
|
fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
'cerfa')
|
|
forwardfile.save()
|
|
assert isinstance(forwardfile, ForwardFile)
|
|
assert forwardfile.upload_status == 'pending'
|
|
file_id = forwardfile.id # pylint: disable=no-member
|
|
assert file_id
|
|
|
|
fake_resp_bad = Response()
|
|
fake_resp_bad.status_code = 502
|
|
fake_resp_bad.reason = 'Bad gateway'
|
|
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp_bad
|
|
atreal_openads.upload_user_files(request_1, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
file_ids=str(file_id))
|
|
assert unicode(exception.value) == u'HTTP error: 502'
|
|
|
|
ffup = ForwardFile.objects.get(id=file_id) # pylint: disable=no-member
|
|
assert isinstance(ffup, ForwardFile)
|
|
for k in ['numero_dossier', 'type_fichier', 'file_hash', 'orig_filename', 'content_type']:
|
|
assert getattr(ffup, k) == getattr(forwardfile, k)
|
|
assert ffup.upload_attempt == 1
|
|
assert ffup.upload_status == 'failed'
|
|
assert ffup.upload_msg == "HTTP error: 502"
|
|
|
|
ffup.upload_status = 'pending'
|
|
ffup.save()
|
|
|
|
fake_resp = Response()
|
|
fake_resp.status_code = 200
|
|
fake_resp.headers = {'Content-Type': 'application/json'}
|
|
fake_resp.encoding = 'utf-8'
|
|
fake_resp.reason = 'OK'
|
|
|
|
fake_resp._content = 'invalid_;{[content}' # pylint: disable=protected-access
|
|
with pytest.raises(APIError) as exception:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
atreal_openads.upload_user_files(request_1,
|
|
'DIA',
|
|
fake_conf['FAKE_NUMERO_DOSSIER'],
|
|
file_ids=[file_id])
|
|
assert unicode(exception.value) == u'No JSON content returned: %r' % fake_resp.content
|
|
|
|
ffup = ForwardFile.objects.get(id=file_id) # pylint: disable=no-member
|
|
assert isinstance(ffup, ForwardFile)
|
|
for k in ['numero_dossier', 'type_fichier', 'file_hash', 'orig_filename', 'content_type']:
|
|
assert getattr(ffup, k) == getattr(forwardfile, k)
|
|
assert ffup.upload_attempt == 2
|
|
assert ffup.upload_status == 'failed'
|
|
# pylint: disable=protected-access
|
|
assert ffup.upload_msg == u'No JSON content returned: %r' % fake_resp._content
|
|
|
|
jresp = atreal_openads.upload_user_files(request_1, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'])
|
|
assert jresp == {'message': 'no file to transfer'}
|
|
|
|
ffup = ForwardFile.objects.get(id=file_id) # pylint: disable=no-member
|
|
ffup.upload_status = 'pending'
|
|
ffup.save()
|
|
|
|
fake_resp_json = "You want add some files on %s " % fake_conf['FAKE_NUMERO_DOSSIER']
|
|
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
jresp = atreal_openads.upload_user_files(request_1, 'DIA',
|
|
fake_conf['FAKE_NUMERO_DOSSIER'])
|
|
assert jresp == {'message': 'all files transfered successfully'}
|
|
|
|
ffup = ForwardFile.objects.get(id=file_id) # pylint: disable=no-member
|
|
assert isinstance(ffup, ForwardFile)
|
|
for k in ['numero_dossier', 'type_fichier', 'file_hash', 'orig_filename', 'content_type']:
|
|
assert getattr(ffup, k) == getattr(forwardfile, k)
|
|
assert ffup.upload_attempt == 3
|
|
assert ffup.upload_status == 'success'
|
|
assert ffup.upload_msg == 'uploaded successfuly'
|
|
|
|
# pylint: disable=unused-argument,redefined-outer-name
|
|
# pylint: disable=too-many-statements
|
|
def test_get_courrier_type(fake_conf, atreal_openads):
|
|
"""Test the method AtrealOpenads.get_courrier_type()."""
|
|
|
|
type_mapping = {
|
|
'DIA': {
|
|
'default': 'dia_renonciation_preempter',
|
|
'detailled': [],
|
|
'specifics': [
|
|
('delegation', 'dia_delegation'),
|
|
('irrecevabilite', 'dia_irrecevabilite'),
|
|
('preemption', 'dia_souhait_preempter')]},
|
|
'CU': {
|
|
'default': 'CUa',
|
|
'detailled': ['CUa', 'CUb'],
|
|
'CUb': {
|
|
'default': 'CUb - ACCORD',
|
|
'specifics': [
|
|
('refus', 'CUb - REFUS')]}},
|
|
'DP': {
|
|
'default': 'decision_non_opposition_DP',
|
|
'detailled': [],
|
|
'specifics': [
|
|
('annulation', 'arrete_annulation_DP'),
|
|
('prolongation', 'arrete_prorogation_DP'),
|
|
('prolongation_refus', 'arrete_refus_prorogation_DP'),
|
|
('transfert_refus', 'arrete_refus_transfert_DP'),
|
|
('transfert', 'arrete_transfert_DP'),
|
|
('attestation_tacite', 'attestation_non opposition_tacite_DP'),
|
|
('opposition', 'decision_opposition_DP'),
|
|
('certificat_tacite', 'certificat_opposition_tacite_DP')]},
|
|
'PC': {
|
|
'default': 'arrete_ss_reserves',
|
|
'detailled': [],
|
|
'specifics': [
|
|
('refus', 'arrete_refus')]},
|
|
'PI': {
|
|
'default': 'arrete_ss_reserves',
|
|
'detailled': [],
|
|
'specifics': [
|
|
('refus', 'arrete_refus')]}
|
|
}
|
|
|
|
# pylint: disable=too-many-nested-blocks
|
|
for type_dossier, infos in type_mapping.items():
|
|
courrier_type = infos['default']
|
|
types_detailles = infos['detailled'] if infos['detailled'] else [None]
|
|
for type_detaille in types_detailles:
|
|
if type_detaille:
|
|
if type_detaille in infos:
|
|
|
|
if 'default' in infos[type_detaille]:
|
|
courrier_type = infos[type_detaille]['default']
|
|
res = atreal_openads.get_courrier_type(None, type_dossier,
|
|
type_dossier_detaille=type_detaille)
|
|
assert res == {'courrier_type': courrier_type}
|
|
|
|
if 'specifics' in infos[type_detaille]:
|
|
for specific_t in infos[type_detaille]['specifics']:
|
|
res = atreal_openads.get_courrier_type(
|
|
None, type_dossier, type_dossier_detaille=type_detaille,
|
|
specific=specific_t[0])
|
|
assert res == {'courrier_type': specific_t[1]}
|
|
|
|
elif 'specifics' in infos:
|
|
for specific_t in infos['specifics']:
|
|
res = atreal_openads.get_courrier_type(None, type_dossier,
|
|
type_dossier_detaille=type_detaille,
|
|
specific=specific_t[0])
|
|
assert res == {'courrier_type': specific_t[1]}
|
|
|
|
else:
|
|
res = atreal_openads.get_courrier_type(None, type_dossier)
|
|
assert res == {'courrier_type': courrier_type}
|