913 lines
36 KiB
Python
913 lines
36 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
# to run it use the following command in the 'tests' directory:
|
|
# ~> DJANGO_SETTINGS_MODULE=passerelle.settings pytest -W ignore::django.utils.deprecation.RemovedInDjango20Warning test_atreal_openads.py -vv
|
|
#
|
|
# and with 'coverage':
|
|
# ~> DJANGO_SETTINGS_MODULE=passerelle.settings pytest -W ignore::django.utils.deprecation.RemovedInDjango20Warning test_atreal_openads.py -vv --cov=~/src/passerelle/passerelle/apps/atreal_openads
|
|
|
|
import pytest
|
|
import mock
|
|
import json
|
|
import os
|
|
import base64
|
|
import re
|
|
import datetime
|
|
import magic
|
|
|
|
from requests import Response
|
|
|
|
from django.http import Http404
|
|
from django.http.request import HttpRequest, QueryDict
|
|
from django.http.response import JsonResponse
|
|
from django.core.files import File
|
|
from django.core.files.base import ContentFile
|
|
#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,
|
|
Guichet,
|
|
Collectivite,
|
|
AtrealOpenads
|
|
)
|
|
|
|
|
|
CONNECTOR_NAME = 'atreal-openads'
|
|
CONNECTOR_SLUG = 'atreal'
|
|
COLLECTIVITE = 79
|
|
OPENADS_API_LOGIN = 'publik-passerelle'
|
|
OPENADS_API_PASSWORD = base64.urlsafe_b64encode(os.urandom(20))
|
|
OPENADS_API_URL = 'http://openads.api/'
|
|
|
|
FAKE_COOKIE_CRSF = base64.urlsafe_b64encode(os.urandom(20))
|
|
FAKE_NUMERO_DOSSIER = base64.urlsafe_b64encode(os.urandom(10))
|
|
|
|
TESTS_DIR = os.path.dirname(__file__)
|
|
RESOURCES_DIR = os.path.join(TESTS_DIR, 'resources')
|
|
TEST_FILE_CERFA_DIA = os.path.join(RESOURCES_DIR, 'cerfa_10072-02.pdf')
|
|
TEST_FILE_PLAN_CADASTRAL = os.path.join(RESOURCES_DIR, 'plancadastral.pdf')
|
|
|
|
|
|
@pytest.fixture
|
|
def atreal_openads(db):
|
|
return AtrealOpenads.objects.create(
|
|
slug = CONNECTOR_SLUG,
|
|
default_collectivite_openADS_id = COLLECTIVITE,
|
|
openADS_API_url = OPENADS_API_URL,
|
|
basic_auth_username = OPENADS_API_LOGIN,
|
|
basic_auth_password = OPENADS_API_PASSWORD
|
|
)
|
|
|
|
@pytest.fixture
|
|
def collectivite_1(db, atreal_openads):
|
|
return Collectivite.objects.create(
|
|
name = u'Macollectivité',
|
|
connecteur = atreal_openads,
|
|
openADS_id = '3'
|
|
)
|
|
|
|
@pytest.fixture
|
|
def collectivite_1_guichet(db, atreal_openads, collectivite_1):
|
|
return Guichet.objects.create(
|
|
collectivite = collectivite_1,
|
|
ouverture_jour_h = datetime.time(9, 0),
|
|
fermeture_jour_h = datetime.time(17, 0),
|
|
ouverture_sem_d = 1, # Lundi
|
|
fermeture_sem_d = 5, # Vendredi
|
|
ouverture_sem_h = datetime.time(8, 30),
|
|
fermeture_sem_h = datetime.time(12, 15)
|
|
)
|
|
|
|
|
|
def upload2ForwardFile(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 fp:
|
|
fwd_file.file_hash = get_file_digest(fp)
|
|
fwd_file.upload_file = File(open(path, 'r'))
|
|
fwd_file.upload_status = 'pending'
|
|
return fwd_file
|
|
return None
|
|
|
|
|
|
def test_forward_file(atreal_openads):
|
|
ff = ForwardFile(
|
|
numero_demande='45641531',
|
|
numero_dossier=FAKE_NUMERO_DOSSIER,
|
|
type_fichier='CERFA',
|
|
orig_filename='afile',
|
|
file_hash='ffdf456fdsvgb4bgfb6g4f5b',
|
|
upload_status='pending',
|
|
connecteur=atreal_openads,
|
|
collectivite=None
|
|
)
|
|
ff.upload_file.save(ff.orig_filename, ContentFile(get_file_data(TEST_FILE_CERFA_DIA)))
|
|
ff.save()
|
|
|
|
assert repr(ff) == (
|
|
u'ForwardFile(id=%s,connecteur=%s,collectivite=%s'
|
|
',demande=%s,dossier=%s,type=%s,filename=%s,status=%s)' % (
|
|
ff.id, unicode(ff.connecteur), None,
|
|
ff.numero_demande, ff.numero_dossier,
|
|
ff.type_fichier, ff.orig_filename, ff.upload_status
|
|
)
|
|
).encode('utf-8')
|
|
|
|
assert str(ff) == '%s[%s]' % (trunc_str_values(ff.orig_filename, 20), 'Pending')
|
|
assert unicode(ff) == u'%s[%s]' % (trunc_str_values(ff.orig_filename, 20), 'Pending')
|
|
|
|
assert ff.get_status() == 'Pending'
|
|
assert ff.get_status('invalid') == 'invalid'
|
|
|
|
params = ff.get_url_params()
|
|
assert params['connecteur'] == atreal_openads.slug
|
|
|
|
assert ff.upload_file is not None
|
|
assert ff.upload_file.size > 0
|
|
assert ff.size == ff.upload_file.size
|
|
assert ff.file_hash == '811588016518eedeb4507f3e4c41be958a03576b0cd20bdb2cb9c6a186dbd887'
|
|
|
|
ff.content_type = 'application/pdf'
|
|
ff.upload_status = 'success'
|
|
ff.save()
|
|
assert ff.upload_status == 'success'
|
|
assert ff.get_status() == 'Success'
|
|
assert ff.content_type == 'application/pdf'
|
|
|
|
with pytest.raises(ValueError) as e:
|
|
ff.upload_file.size
|
|
assert unicode(e.value) == "The 'upload_file' attribute has no file associated with it."
|
|
assert ff.size > 0
|
|
assert ff.file_hash == '811588016518eedeb4507f3e4c41be958a03576b0cd20bdb2cb9c6a186dbd887'
|
|
|
|
ff.file_hash = ''
|
|
ff.update_file_hash()
|
|
ff.update_content_type()
|
|
ff.save()
|
|
assert ff.file_hash == ''
|
|
assert ff.content_type == ''
|
|
|
|
ff.orig_filename = ''
|
|
with pytest.raises(ValidationError) as e:
|
|
ff.save()
|
|
assert len(e.value.messages) == 1
|
|
assert '__all__' in e.value.message_dict
|
|
assert unicode(e.value.message_dict['__all__'][0]) == u"A %s cannot have all the following fields empty: %s." % (
|
|
ff.get_verbose_name(),
|
|
['file_hash', 'orig_filename', 'upload_file'])
|
|
|
|
ff.delete()
|
|
|
|
|
|
def test_collectivite(collectivite_1, collectivite_1_guichet):
|
|
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
|
|
|
|
|
|
def test_guichet(collectivite_1_guichet):
|
|
g = collectivite_1_guichet
|
|
|
|
assert repr(g) == (
|
|
u'Guichet(id=%s,collectivite=%s,%s)' % (
|
|
1, unicode(g.collectivite), unicode(g)
|
|
)
|
|
).encode('utf-8')
|
|
|
|
assert str(g) == u'Monday 08:30 -> Friday 12:15 [09:00/17:00]'.encode('utf-8')
|
|
|
|
assert unicode(g) == u'Monday 08:30 -> Friday 12:15 [09:00/17:00]'
|
|
|
|
params = g.get_url_params()
|
|
assert params['collectivite'] == g.collectivite.id
|
|
|
|
with pytest.raises(Exception) as e:
|
|
g.get_list_url()
|
|
assert unicode(e.value) == u"Guichet:get_list_url() method should not be called"
|
|
|
|
|
|
def test_guichet_is_open(collectivite_1_guichet):
|
|
g = 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 d in [d_monday, d_tuesday, d_wednesday, d_thursday, d_friday]:
|
|
for t in [(t_open, True), (t_closed_before, False), (t_closed_after, False)]:
|
|
dt = datetime.datetime.strptime(d + ' ' + t[0], dt_fmt)
|
|
assert g.is_open(dt) == t[1]
|
|
|
|
dt = datetime.datetime.strptime(d_friday + ' 16:12', dt_fmt)
|
|
assert g.is_open(dt) == False
|
|
|
|
for d in [d_saturday, d_sunday]:
|
|
for t in [t_open, t_closed_before, t_closed_after]:
|
|
dt = datetime.datetime.strptime(d + ' ' + t, dt_fmt)
|
|
assert g.is_open(dt) == False
|
|
|
|
with pytest.raises(TypeError) as e:
|
|
g.is_open('invalid datetime')
|
|
assert unicode(e.value) == u"is_open() expect a datetime object (not a %s)" % type('')
|
|
|
|
assert g.is_open(None) == False
|
|
|
|
|
|
def test_openads_log_json_payload(atreal_openads):
|
|
# TODO implement
|
|
assert True
|
|
# change the debug file path
|
|
# check that the function log to it
|
|
# check that what was is logged is correct
|
|
|
|
|
|
def test_openads_get_files_from_json_payload(atreal_openads):
|
|
title = 'payload'
|
|
|
|
assert atreal_openads.get_files_from_json_payload({'files':[{'a':'file'}]}) == [{'a':'file'}]
|
|
|
|
with pytest.raises(APIError) as e:
|
|
atreal_openads.get_files_from_json_payload({})
|
|
assert unicode(e.value) == u"Expecting '%s' key in JSON %s" % ('files', title)
|
|
|
|
with pytest.raises(APIError) as e:
|
|
atreal_openads.get_files_from_json_payload({'files': 'invalid'})
|
|
assert unicode(e.value) == u"Expecting '%s' value in JSON %s to be a %s (not a %s)" % (
|
|
'files', title, 'list', type(''))
|
|
|
|
with pytest.raises(APIError) as e:
|
|
atreal_openads.get_files_from_json_payload({'files': {'i':'invalid'}})
|
|
assert unicode(e.value) == u"Expecting '%s' value in JSON %s to be a %s (not a %s)" % (
|
|
'files', title, 'list', type({}))
|
|
|
|
with pytest.raises(APIError) as e:
|
|
atreal_openads.get_files_from_json_payload({'files': []})
|
|
assert unicode(e.value) == u"Expecting non-empty '%s' value in JSON %s" % ('files', title)
|
|
|
|
|
|
def test_check_file_dict(atreal_openads):
|
|
title = 'payload'
|
|
|
|
d = {
|
|
'content': get_file_data(TEST_FILE_CERFA_DIA, b64=False),
|
|
'filename': os.path.basename(TEST_FILE_CERFA_DIA),
|
|
'content_type': 'application/pdf'
|
|
}
|
|
|
|
d64 = {
|
|
'b64_content': get_file_data(TEST_FILE_CERFA_DIA, b64=True),
|
|
'filename': os.path.basename(TEST_FILE_CERFA_DIA),
|
|
'content_type': 'application/pdf'
|
|
}
|
|
|
|
assert atreal_openads.check_file_dict(d, b64=False) is None
|
|
assert atreal_openads.check_file_dict(d64) is None
|
|
|
|
d['filename'] = {'a','filename'}
|
|
with pytest.raises(APIError) as e:
|
|
atreal_openads.check_file_dict(d, b64=False)
|
|
assert unicode(e.value) == u"Expecting '%s' value in JSON %s in file dict to be a %s (not a %s)" % (
|
|
'file.filename', title, 'string', type(d['filename']))
|
|
|
|
d['content'] = {'a','filename'}
|
|
with pytest.raises(APIError) as e:
|
|
atreal_openads.check_file_dict(d, b64=False)
|
|
assert unicode(e.value) == u"Expecting '%s' value in JSON %s in file dict to be a %s (not a %s)" % (
|
|
'file.content', title, 'string', type(d['content']))
|
|
|
|
del(d['content'])
|
|
with pytest.raises(APIError) as e:
|
|
atreal_openads.check_file_dict(d, b64=False)
|
|
assert unicode(e.value) == u"Expecting 'file.%s' key in JSON %s" % ('content', title)
|
|
|
|
del(d64['b64_content'])
|
|
with pytest.raises(APIError) as e:
|
|
atreal_openads.check_file_dict(d, b64=True)
|
|
assert unicode(e.value) == u"Expecting 'file.%s' key in JSON %s" % ('b64_content', title)
|
|
|
|
|
|
def test_get_first_file_from_json_payload(atreal_openads):
|
|
title = 'payload'
|
|
|
|
d = {
|
|
'files': [{
|
|
'content': get_file_data(TEST_FILE_CERFA_DIA, b64=False),
|
|
'filename': os.path.basename(TEST_FILE_CERFA_DIA),
|
|
'content_type': 'application/pdf'
|
|
}]
|
|
}
|
|
|
|
assert atreal_openads.get_first_file_from_json_payload(
|
|
d, title, ensure_content=True, b64=False) == d['files'][0]
|
|
|
|
|
|
def test_openads_check_status(atreal_openads):
|
|
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
|
|
|
|
|
|
def test_openads_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
|
|
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(TEST_FILE_CERFA_DIA),
|
|
"content_type": "invalid/content type",
|
|
"field_id" : "50",
|
|
"filename" : os.path.basename(TEST_FILE_CERFA_DIA)
|
|
},
|
|
"plan_cadastral_1": {
|
|
"content" : get_file_data(TEST_FILE_PLAN_CADASTRAL),
|
|
"content_type": "application/pdf",
|
|
"filename" : os.path.basename(TEST_FILE_PLAN_CADASTRAL)
|
|
},
|
|
"plan_cadastral_2": {
|
|
"content" : get_file_data(TEST_FILE_PLAN_CADASTRAL),
|
|
"content_type": "application/pdf",
|
|
#"filename" : 'plan_cad'
|
|
},
|
|
"pouvoir_mandat": {
|
|
"content" : get_file_data(TEST_FILE_CERFA_DIA),
|
|
"content_type": "application/pdf",
|
|
"filename" : 'mandat'
|
|
}
|
|
}
|
|
}
|
|
req = HttpRequest()
|
|
req._body = json.dumps(fake_req_json)
|
|
req.path = '/test'
|
|
req.method = 'POST'
|
|
req.encoding = 'utf-8'
|
|
req.GET = QueryDict(mutable=True) # required because of encoding setter
|
|
req.POST = QueryDict(mutable=True) # required because of encoding setter
|
|
req.content_type = 'application/json'
|
|
req.content_params = None
|
|
req.COOKIES = {}
|
|
req.META = {}
|
|
req._read_started = False
|
|
|
|
fake_resp_bad = Response()
|
|
fake_resp_bad.status_code = 502
|
|
fake_resp_bad.reason = 'Bad gateway'
|
|
|
|
with pytest.raises(ValueError) as e:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp_bad
|
|
atreal_openads.create_dossier(req, 'DIA', collectivite='not an integer')
|
|
assert unicode(e.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 e:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp_bad
|
|
atreal_openads.create_dossier(req, 'DIA', collectivite_1.openADS_id, now=now_open)
|
|
assert unicode(e.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 e:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp_404
|
|
atreal_openads.create_dossier(req, 'DIA', collectivite=999)
|
|
assert unicode(e.value) == "HTTP error: 404"
|
|
|
|
jresp = atreal_openads.create_dossier(req, 'DIA', 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(req, 'DIA', 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 e:
|
|
jresp = atreal_openads.create_dossier(req, 'DIA', collectivite_1.openADS_id, now=now_invalid)
|
|
assert unicode(e.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)
|
|
|
|
fake_resp_json = {
|
|
'numero_dossier' : FAKE_NUMERO_DOSSIER,
|
|
'files': [{
|
|
'b64_content' : get_file_data(TEST_FILE_CERFA_DIA),
|
|
'content_type' : 'text/plain',
|
|
'filename' : 'recepisse_depot_%s.pdf' % 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
|
|
jresp = atreal_openads.create_dossier(req, 'DIA', collectivite_1.openADS_id, now=now_open)
|
|
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)
|
|
with pytest.raises(APIError) as e:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
atreal_openads.create_dossier(req, 'DIA', collectivite_1.openADS_id, now=now_open)
|
|
assert unicode(e.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)
|
|
with pytest.raises(APIError) as e:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
atreal_openads.create_dossier(req, 'DIA', collectivite_1.openADS_id, now=now_open)
|
|
assert unicode(e.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)
|
|
with pytest.raises(APIError) as e:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
atreal_openads.create_dossier(req, 'DIA', collectivite_1.openADS_id, now=now_open)
|
|
assert unicode(e.value) == u'Failed to decode recepisse content from base 64'
|
|
|
|
fake_resp._content = 'df[{gfd;g#vfd'
|
|
with pytest.raises(APIError) as e:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
atreal_openads.create_dossier(req, 'DIA', collectivite_1.openADS_id, now=now_open)
|
|
assert unicode(e.value) == u'No JSON content returned: %r' % fake_resp._content
|
|
|
|
job = Job.objects.filter(natural_id=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_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']
|
|
|
|
FFs = ForwardFile.objects.filter(id__in=file_ids)
|
|
for FF in FFs:
|
|
assert len(FF.numero_demande) > 0
|
|
assert FF.numero_dossier == FAKE_NUMERO_DOSSIER
|
|
assert len(FF.file_hash) > 0
|
|
assert FF.upload_status == 'pending'
|
|
|
|
fake_resp_json = "You want add some files on %s " % 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)
|
|
assert job.status == 'completed'
|
|
|
|
FFs = ForwardFile.objects.filter(id__in=file_ids)
|
|
for FF in FFs:
|
|
assert FF.upload_status == 'success'
|
|
|
|
|
|
def test_openads_get_dossier(atreal_openads):
|
|
fake_resp_bad = Response()
|
|
fake_resp_bad.status_code = 502
|
|
fake_resp_bad.reason = 'Bad gateway'
|
|
|
|
with pytest.raises(APIError) as e:
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp_bad
|
|
atreal_openads.get_dossier(None, 'DIA', FAKE_NUMERO_DOSSIER)
|
|
assert unicode(e.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)
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp
|
|
jresp = atreal_openads.get_dossier(None, 'DIA', 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'
|
|
with pytest.raises(APIError) as e:
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp
|
|
atreal_openads.get_dossier(None, 'DIA', FAKE_NUMERO_DOSSIER)
|
|
assert unicode(e.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)
|
|
with pytest.raises(APIError) as e:
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp
|
|
atreal_openads.get_dossier(None, 'invalid_type', FAKE_NUMERO_DOSSIER)
|
|
assert unicode(e.value) == u"HTTP error: 404, [path] (Invalid Type) \"invalid_type\" is not one of DIA, PC, DP, AT, PD"
|
|
|
|
|
|
def test_openads_get_fwd_files(atreal_openads):
|
|
with pytest.raises(APIError) as e:
|
|
atreal_openads.get_fwd_files(None, FAKE_NUMERO_DOSSIER, fichier_id='not an integer')
|
|
assert unicode(e.value) == u"fichier_id must be an integer"
|
|
|
|
with pytest.raises(Http404) as e:
|
|
atreal_openads.get_fwd_files(None, FAKE_NUMERO_DOSSIER, fichier_id=18)
|
|
assert unicode(e.value) == u"No file matches 'numero_dossier=%s' and 'id=%s'." % (FAKE_NUMERO_DOSSIER, 18)
|
|
|
|
resp_empty = atreal_openads.get_fwd_files(None, FAKE_NUMERO_DOSSIER, fichier_id=None)
|
|
assert resp_empty is not None
|
|
assert len(resp_empty) == 0
|
|
|
|
FF = upload2ForwardFile(atreal_openads, TEST_FILE_CERFA_DIA, FAKE_NUMERO_DOSSIER, 'cerfa')
|
|
FF.save()
|
|
assert isinstance(FF, ForwardFile)
|
|
|
|
jresp = atreal_openads.get_fwd_files(None, FAKE_NUMERO_DOSSIER, fichier_id=None)
|
|
assert jresp is not None
|
|
assert len(jresp) == 1
|
|
assert jresp[0]['id'] == FF.id
|
|
for k in ['numero_dossier', 'type_fichier', 'file_hash', 'orig_filename', 'content_type', 'upload_status', 'upload_msg']:
|
|
assert jresp[0][k] == getattr(FF, k)
|
|
assert jresp[0]['content_size'] == len(get_file_data(FF.upload_file.path, b64=False))
|
|
assert jresp[0]['last_update_datetime'] == FF.last_update_datetime
|
|
|
|
jresp = atreal_openads.get_fwd_files(None, FAKE_NUMERO_DOSSIER, fichier_id=FF.id)
|
|
assert jresp is not None
|
|
assert len(jresp) == 1
|
|
assert jresp[0]['id'] == FF.id
|
|
for k in ['numero_dossier', 'type_fichier', 'file_hash', 'orig_filename', 'content_type', 'upload_status', 'upload_msg']:
|
|
assert jresp[0][k] == getattr(FF, k)
|
|
assert jresp[0]['content_size'] == len(get_file_data(FF.upload_file.path, b64=False))
|
|
assert jresp[0]['last_update_datetime'] == FF.last_update_datetime
|
|
|
|
|
|
def test_openads_get_fwd_files_status(atreal_openads):
|
|
with pytest.raises(Http404) as e:
|
|
atreal_openads.get_fwd_files_status(None, FAKE_NUMERO_DOSSIER, fichier_id=18)
|
|
assert re.search(r"^No file matches 'numero_dossier=[^']+' and 'id=[^']+'.$", str(e.value))
|
|
|
|
FF = upload2ForwardFile(atreal_openads, TEST_FILE_CERFA_DIA, FAKE_NUMERO_DOSSIER, 'cerfa')
|
|
FF.save()
|
|
assert isinstance(FF, ForwardFile)
|
|
|
|
jresp = atreal_openads.get_fwd_files_status(None, FAKE_NUMERO_DOSSIER, fichier_id=None)
|
|
assert jresp is not None
|
|
assert jresp['all_forwarded'] == False
|
|
status_msg = '[%s] %s => %s' % (FF.id, FF.orig_filename, FF.upload_msg)
|
|
assert len(jresp['pending']) == 1
|
|
assert jresp['pending'][0] == status_msg
|
|
assert len(jresp['uploading']) == 0
|
|
assert len(jresp['success']) == 0
|
|
assert len(jresp['failed']) == 0
|
|
|
|
jresp = atreal_openads.get_fwd_files_status(None, FAKE_NUMERO_DOSSIER, fichier_id=FF.id)
|
|
assert jresp is not None
|
|
assert jresp['all_forwarded'] == False
|
|
status_msg = '[%s] %s => %s' % (FF.id, FF.orig_filename, FF.upload_msg)
|
|
assert len(jresp['pending']) == 1
|
|
assert jresp['pending'][0] == status_msg
|
|
assert len(jresp['uploading']) == 0
|
|
assert len(jresp['success']) == 0
|
|
assert len(jresp['failed']) == 0
|
|
|
|
|
|
def test_openads_get_courrier(atreal_openads):
|
|
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 e:
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp_bad
|
|
atreal_openads.get_courrier(None, 'DIA', FAKE_NUMERO_DOSSIER, lettre_type)
|
|
assert unicode(e.value) == "HTTP error: 502"
|
|
|
|
fake_resp_json = {
|
|
'files': [{
|
|
'filename' : "instruction_4.pdf",
|
|
'content_type' : "text/plain",
|
|
'b64_content' : get_file_data(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)
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp
|
|
jresp = atreal_openads.get_courrier(None, 'DIA', 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)
|
|
with pytest.raises(APIError) as e:
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp
|
|
atreal_openads.get_courrier(None, 'DIA', FAKE_NUMERO_DOSSIER, lettre_type)
|
|
assert unicode(e.value) == u'Failed to decode courrier content from base 64'
|
|
|
|
fake_resp._content = 'df[{gfd;g#vfd'
|
|
with pytest.raises(APIError) as e:
|
|
with mock.patch('passerelle.utils.Request.get') as requests_get:
|
|
requests_get.return_value = fake_resp
|
|
atreal_openads.get_courrier(None, 'DIA', FAKE_NUMERO_DOSSIER, lettre_type)
|
|
assert unicode(e.value) == u'No JSON content returned: %r' % fake_resp._content
|
|
|
|
|
|
def test_get_response_error(atreal_openads):
|
|
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)
|
|
|
|
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}'
|
|
error_msg = atreal_openads.get_response_error(fake_resp)
|
|
assert error_msg == u"HTTP error: %s, %s" % (fake_resp.status_code, fake_resp._content)
|
|
|
|
|
|
def test_openads_upload_user_files(atreal_openads):
|
|
|
|
req = HttpRequest()
|
|
req._body = ''
|
|
req.path = '/upload_user_files'
|
|
req.method = 'GET'
|
|
req.encoding = 'utf-8'
|
|
req.GET = QueryDict(mutable=True) # required because of encoding setter
|
|
req.POST = QueryDict(mutable=True) # required because of encoding setter
|
|
req.content_type = 'application/json'
|
|
req.content_params = None
|
|
req.COOKIES = {}
|
|
req.META = {}
|
|
req._read_started = False
|
|
|
|
with pytest.raises(ForwardFile.DoesNotExist) as e:
|
|
atreal_openads.upload_user_files(req, 'DIA', FAKE_NUMERO_DOSSIER, file_ids=[999])
|
|
assert unicode(e.value) == u"The following ForwardFile IDs were not found: %s." % [999]
|
|
|
|
with pytest.raises(ValueError) as e:
|
|
atreal_openads.upload_user_files(req, 'DIA', FAKE_NUMERO_DOSSIER, file_ids='invalid string')
|
|
assert unicode(e.value) == u"invalid literal for int() with base 10: '%s'" % 'invalid string'
|
|
|
|
with pytest.raises(TypeError) as e:
|
|
atreal_openads.upload_user_files(req, 'DIA', FAKE_NUMERO_DOSSIER, file_ids={'invalid':'type'})
|
|
assert unicode(e.value) == u"Invalid 'file_ids' argument type '%s' (must be string or list)" % type({'invalid':'type'})
|
|
|
|
FF = upload2ForwardFile(atreal_openads, TEST_FILE_CERFA_DIA, FAKE_NUMERO_DOSSIER, 'cerfa')
|
|
FF.save()
|
|
assert isinstance(FF, ForwardFile)
|
|
assert FF.upload_status == 'pending'
|
|
file_id = FF.id
|
|
assert file_id
|
|
|
|
fake_resp_bad = Response()
|
|
fake_resp_bad.status_code = 502
|
|
fake_resp_bad.reason = 'Bad gateway'
|
|
|
|
with pytest.raises(APIError) as e:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp_bad
|
|
atreal_openads.upload_user_files(req, 'DIA', FAKE_NUMERO_DOSSIER, file_ids=str(file_id))
|
|
assert unicode(e.value) == u'HTTP error: 502'
|
|
|
|
FFup = ForwardFile.objects.get(id=file_id)
|
|
assert isinstance(FFup, ForwardFile)
|
|
for k in ['numero_dossier', 'type_fichier', 'file_hash', 'orig_filename', 'content_type']:
|
|
assert getattr(FFup, k) == getattr(FF, 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}'
|
|
with pytest.raises(APIError) as e:
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
atreal_openads.upload_user_files(req, 'DIA', FAKE_NUMERO_DOSSIER, file_ids=[file_id])
|
|
assert unicode(e.value) == u'No JSON content returned: %r' % fake_resp.content
|
|
|
|
FFup = ForwardFile.objects.get(id=file_id)
|
|
assert isinstance(FFup, ForwardFile)
|
|
for k in ['numero_dossier', 'type_fichier', 'file_hash', 'orig_filename', 'content_type']:
|
|
assert getattr(FFup, k) == getattr(FF, k)
|
|
assert FFup.upload_attempt == 2
|
|
assert FFup.upload_status == 'failed'
|
|
assert FFup.upload_msg == u'No JSON content returned: %r' % fake_resp._content
|
|
|
|
jresp = atreal_openads.upload_user_files(req, 'DIA', FAKE_NUMERO_DOSSIER)
|
|
assert jresp == {'message': 'no file to transfer'}
|
|
|
|
FFup = ForwardFile.objects.get(id=file_id)
|
|
FFup.upload_status = 'pending'
|
|
FFup.save()
|
|
|
|
fake_resp_json = "You want add some files on %s " % FAKE_NUMERO_DOSSIER
|
|
fake_resp._content = json.dumps(fake_resp_json)
|
|
with mock.patch('passerelle.utils.Request.post') as requests_post:
|
|
requests_post.return_value = fake_resp
|
|
jresp = atreal_openads.upload_user_files(req, 'DIA', FAKE_NUMERO_DOSSIER)
|
|
assert jresp == {'message': 'all files transfered successfully'}
|
|
|
|
FFup = ForwardFile.objects.get(id=file_id)
|
|
assert isinstance(FFup, ForwardFile)
|
|
for k in ['numero_dossier', 'type_fichier', 'file_hash', 'orig_filename', 'content_type']:
|
|
assert getattr(FFup, k) == getattr(FF, k)
|
|
assert FFup.upload_attempt == 3
|
|
assert FFup.upload_status == 'success'
|
|
assert FFup.upload_msg == 'uploaded successfuly'
|
|
|