Tests: Refactoring: moving fixtures to conftest.py, added some fixture for ForwardFile and HttpRequest

This commit is contained in:
Michael Bideau 2019-08-26 14:32:32 +00:00
parent 38061d4593
commit 31ee9a8ea6
5 changed files with 364 additions and 614 deletions

View File

@ -20,6 +20,10 @@
"""Configuration and fixtures for tests files."""
import os
import base64
import datetime
import pytest
from httmock import urlmatch, HTTMock, response
@ -27,6 +31,15 @@ from httmock import urlmatch, HTTMock, response
import django_webtest
from django.core.cache import cache
from django.core.files import File
from django.http.request import HttpRequest, QueryDict
from atreal_openads.models import (
ForwardFile,
Guichet,
Collectivite,
AtrealOpenads
)
@pytest.fixture(autouse=True)
@ -71,7 +84,6 @@ def mock_500():
def relax_openssl(tmpdir):
'''OpenSSL default configuration has been really strict for some years,
this fixture set a temporary really permisive ciphers list.'''
import os
openssl_cnf_path = tmpdir / 'openssl.cnf'
with openssl_cnf_path.open('w') as file_pt:
@ -93,3 +105,118 @@ CipherString = ALL''')
del os.environ['OPENSSL_CONF']
else:
os.environ['OPENSSL_CONF'] = old_value
@pytest.fixture
def fake_conf():
"""Return a dictionnary containing configuration items."""
conf = {
'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__)
}
conf['RESOURCES_DIR'] = os.path.join(conf['TESTS_DIR'], 'resources')
conf['TEST_FILE_CERFA_DIA'] = os.path.join(conf['RESOURCES_DIR'], 'cerfa_10072-02.pdf')
conf['TEST_FILE_PLAN_CADASTRAL'] = os.path.join(conf['RESOURCES_DIR'], 'plancadastral.pdf')
return conf
@pytest.fixture
# pylint: disable=unused-argument,redefined-outer-name,invalid-name
def atreal_openads(fake_conf, db):
"""Return an instance of a connector AtrealOpenads."""
return AtrealOpenads.objects.create( # pylint: disable=no-member
slug=fake_conf['CONNECTOR_SLUG'],
default_collectivite_openADS_id=fake_conf['COLLECTIVITE'],
openADS_API_url=fake_conf['OPENADS_API_URL'],
basic_auth_username=fake_conf['OPENADS_API_LOGIN'],
basic_auth_password=fake_conf['OPENADS_API_PASSWORD']
)
@pytest.fixture
# pylint: disable=unused-argument,redefined-outer-name,invalid-name
def collectivite_1(db, atreal_openads):
"""Return an instance of a 'Collectivite'."""
return Collectivite.objects.create( # pylint: disable=no-member
name=u'Macollectivité',
connecteur=atreal_openads,
openADS_id='3'
)
@pytest.fixture
# pylint: disable=unused-argument,redefined-outer-name,invalid-name
def collectivite_1_guichet(db, atreal_openads, collectivite_1):
"""Return an instance of a 'Guichet'."""
return Guichet.objects.create( # pylint: disable=no-member
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)
)
@pytest.fixture
# pylint: disable=unused-argument,redefined-outer-name,invalid-name
def forwardfile_1(fake_conf, db, atreal_openads, collectivite_1):
"""Return an instance of a 'ForwardFile'."""
return ForwardFile.objects.create( # pylint: disable=no-member
connecteur=atreal_openads,
collectivite=collectivite_1,
numero_demande='45641531',
numero_dossier=fake_conf['FAKE_NUMERO_DOSSIER'],
type_fichier='CERFA',
orig_filename=os.path.basename(fake_conf['TEST_FILE_CERFA_DIA']),
content_type='application/pdf',
file_hash='ffdf456fdsvgb4bgfb6g4f5b',
upload_file=File(open(fake_conf['TEST_FILE_CERFA_DIA'], 'r')),
upload_status='pending'
)
@pytest.fixture
# pylint: disable=redefined-outer-name,invalid-name
def forwardfile_2(fake_conf, connecteur=None, collectivite=None):
"""Return a forward file object with some values but not saved in db."""
return ForwardFile(
connecteur=connecteur,
collectivite=collectivite,
numero_demande='45641531',
numero_dossier=fake_conf['FAKE_NUMERO_DOSSIER'],
type_fichier='CERFA',
orig_filename=os.path.basename(fake_conf['TEST_FILE_CERFA_DIA']),
content_type='application/pdf',
file_hash='ffdf456fdsvgb4bgfb6g4f5b',
upload_file=File(open(fake_conf['TEST_FILE_CERFA_DIA'], 'r')),
upload_status='pending'
)
@pytest.fixture
def request_1():
"""Return an HttpRequest object with some default values."""
req = HttpRequest()
req._body = '' # pylint: disable=protected-access
req.path = '/'
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 # pylint: disable=protected-access
return req

View File

@ -34,10 +34,8 @@ import pytest
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
@ -52,67 +50,10 @@ from atreal_openads.utils import (
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
# pylint: disable=unused-argument,invalid-name
def atreal_openads(db):
"""Return an instance of a connector AtrealOpenads."""
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
# pylint: disable=unused-argument,redefined-outer-name,invalid-name
def collectivite_1(db, atreal_openads):
"""Return an instance of a 'Collectivite'."""
return Collectivite.objects.create( # pylint: disable=no-member
name=u'Macollectivité',
connecteur=atreal_openads,
openADS_id='3'
)
@pytest.fixture
# pylint: disable=unused-argument,redefined-outer-name,invalid-name
def collectivite_1_guichet(db, atreal_openads, collectivite_1):
"""Return an instance of a 'Guichet'."""
return Guichet.objects.create( # pylint: disable=no-member
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 build_forwardfile_from_path(connecteur, path, numero_dossier, type_fichier):
"""Convert a file path to a ForwardFile."""
if path:
@ -133,82 +74,74 @@ def build_forwardfile_from_path(connecteur, path, numero_dossier, type_fichier):
# pylint: disable=unused-argument,redefined-outer-name
def test_forward_file(atreal_openads):
def test_forward_file(forwardfile_2, atreal_openads):
"""Test ForwardFile object methods."""
forwardfile = ForwardFile(
numero_demande='45641531',
numero_dossier=FAKE_NUMERO_DOSSIER,
type_fichier='CERFA',
orig_filename='afile',
file_hash='forwardfiledf456fdsvgb4bgfb6g4f5b',
upload_status='pending',
connecteur=atreal_openads,
collectivite=None
)
# pylint: disable=no-member
forwardfile.upload_file.save(forwardfile.orig_filename,
ContentFile(get_file_data(TEST_FILE_CERFA_DIA)))
forwardfile.save()
forwardfile_2.orig_filename = 'afile'
forwardfile_2.connecteur = atreal_openads
assert repr(forwardfile) == (
# 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.id, unicode(forwardfile.connecteur), None, # pylint: disable=no-member
forwardfile.numero_demande, forwardfile.numero_dossier,
forwardfile.type_fichier, forwardfile.orig_filename, forwardfile.upload_status
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) == '%s[%s]' % (trunc_str_values(forwardfile.orig_filename, 20),
'Pending')
assert unicode(forwardfile) == u'%s[%s]' % (trunc_str_values(forwardfile.orig_filename, 20),
'Pending')
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.get_status() == 'Pending'
assert forwardfile.get_status('invalid') == 'invalid'
assert forwardfile_2.get_status() == 'Pending'
assert forwardfile_2.get_status('invalid') == 'invalid'
params = forwardfile.get_url_params()
params = forwardfile_2.get_url_params()
assert params['connecteur'] == atreal_openads.slug
assert forwardfile.upload_file is not None
assert forwardfile.upload_file.size > 0
assert forwardfile.size == forwardfile.upload_file.size
assert forwardfile.file_hash == ('811588016518eedeb4507f3e4c41be95'
'8a03576b0cd20bdb2cb9c6a186dbd887')
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.content_type = 'application/pdf'
forwardfile.upload_status = 'success'
forwardfile.save()
assert forwardfile.upload_status == 'success'
assert forwardfile.get_status() == 'Success'
assert forwardfile.content_type == 'application/pdf'
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.upload_file.size # pylint: disable=pointless-statement
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.size > 0
assert forwardfile.file_hash == ('811588016518eedeb4507f3e4c41be95'
'8a03576b0cd20bdb2cb9c6a186dbd887')
assert forwardfile_2.size > 0
assert forwardfile_2.file_hash == ('cc90a620982760fdee16a5b4fe1b5ac3'
'b4fe868fd02d2f70b27f1e46d283ea51')
forwardfile.file_hash = ''
forwardfile.update_file_hash()
forwardfile.update_content_type()
forwardfile.save()
assert forwardfile.file_hash == ''
assert forwardfile.content_type == ''
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.orig_filename = ''
forwardfile_2.orig_filename = ''
with pytest.raises(ValidationError) as exception:
forwardfile.save()
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.get_verbose_name(),
forwardfile_2.get_verbose_name(),
['file_hash', 'orig_filename', 'upload_file']))
forwardfile.delete()
forwardfile_2.delete()
# pylint: disable=unused-argument,redefined-outer-name
@ -358,20 +291,20 @@ def test_get_files_from_payload(atreal_openads):
# pylint: disable=unused-argument,redefined-outer-name
def test_check_file_dict(atreal_openads):
def test_check_file_dict(fake_conf, atreal_openads):
"""Test the method AtrealOpenads.check_file_dict()."""
title = 'payload'
dic = {
'content': get_file_data(TEST_FILE_CERFA_DIA, b64=False),
'filename': os.path.basename(TEST_FILE_CERFA_DIA),
'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(TEST_FILE_CERFA_DIA, b64=True),
'filename': os.path.basename(TEST_FILE_CERFA_DIA),
'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'
}
@ -405,15 +338,15 @@ def test_check_file_dict(atreal_openads):
# pylint: disable=unused-argument,redefined-outer-name
def test_get_first_file_from_payload(atreal_openads): # pylint: disable=invalid-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(TEST_FILE_CERFA_DIA, b64=False),
'filename': os.path.basename(TEST_FILE_CERFA_DIA),
'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'
}]
}
@ -438,7 +371,8 @@ def test_check_status(atreal_openads):
# pylint: disable=unused-argument,redefined-outer-name
# pylint: disable=too-many-statements,too-many-locals
def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
def test_create_dossier(fake_conf, atreal_openads, collectivite_1, collectivite_1_guichet,
request_1):
"""Test the method AtrealOpenads.create_dossier()."""
fake_req_json = {
@ -493,40 +427,30 @@ def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
# user attached files
"cerfa": {
"content": get_file_data(TEST_FILE_CERFA_DIA),
"content": get_file_data(fake_conf['TEST_FILE_CERFA_DIA']),
"content_type": "invalid/content type",
"field_id": "50",
"filename": os.path.basename(TEST_FILE_CERFA_DIA)
"filename": os.path.basename(fake_conf['TEST_FILE_CERFA_DIA'])
},
"plan_cadastral_1": {
"content": get_file_data(TEST_FILE_PLAN_CADASTRAL),
"content": get_file_data(fake_conf['TEST_FILE_PLAN_CADASTRAL']),
"content_type": "application/pdf",
"filename": os.path.basename(TEST_FILE_PLAN_CADASTRAL)
"filename": os.path.basename(fake_conf['TEST_FILE_PLAN_CADASTRAL'])
},
"plan_cadastral_2": {
"content": get_file_data(TEST_FILE_PLAN_CADASTRAL),
"content": get_file_data(fake_conf['TEST_FILE_PLAN_CADASTRAL']),
"content_type": "application/pdf",
# "filename": 'plan_cad'
},
"pouvoir_mandat": {
"content": get_file_data(TEST_FILE_CERFA_DIA),
"content": get_file_data(fake_conf['TEST_FILE_CERFA_DIA']),
"content_type": "application/pdf",
"filename": 'mandat'
}
}
}
req = HttpRequest()
req._body = json.dumps(fake_req_json) # pylint: disable=protected-access
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 # pylint: disable=protected-access
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
@ -535,7 +459,7 @@ def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
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(req, 'DIA', collectivite='not an integer')
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
@ -545,7 +469,7 @@ def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
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(req, 'DIA', collectivite_1.openADS_id, now=now_open)
atreal_openads.create_dossier(request_1, 'DIA', 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
@ -557,10 +481,10 @@ def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
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(req, 'DIA', collectivite=999)
atreal_openads.create_dossier(request_1, 'DIA', collectivite=999)
assert unicode(exception.value) == "HTTP error: 404"
jresp = atreal_openads.create_dossier(req, 'DIA', collectivite_1.openADS_id, now_closed)
jresp = atreal_openads.create_dossier(request_1, 'DIA', collectivite_1.openADS_id, now_closed)
assert jresp is not None
assert len(jresp) == 1
assert 'message' in jresp
@ -568,7 +492,8 @@ def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
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_closed_str)
jresp = atreal_openads.create_dossier(request_1, 'DIA', collectivite_1.openADS_id,
now_closed_str)
assert jresp is not None
assert len(jresp) == 1
assert 'message' in jresp
@ -576,7 +501,8 @@ def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
now_invalid = {'invalid': 'type'}
with pytest.raises(APIError) as exception:
jresp = atreal_openads.create_dossier(req, 'DIA', collectivite_1.openADS_id, now_invalid)
jresp = atreal_openads.create_dossier(request_1, 'DIA', collectivite_1.openADS_id,
now_invalid)
assert unicode(exception.value) == (
u"Invalid value of type '%s' for now argument of endpoint '%s' (must be: %s)" % (
type(now_invalid),
@ -584,11 +510,11 @@ def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
"datetime or string formatted to '%s'" % now_str_fmt))
fake_resp_json = {
'numero_dossier': FAKE_NUMERO_DOSSIER,
'numero_dossier': fake_conf['FAKE_NUMERO_DOSSIER'],
'files': [{
'b64_content': get_file_data(TEST_FILE_CERFA_DIA),
'b64_content': get_file_data(fake_conf['TEST_FILE_CERFA_DIA']),
'content_type': 'text/plain',
'filename': 'recepisse_depot_%s.pdf' % FAKE_NUMERO_DOSSIER,
'filename': 'recepisse_depot_%s.pdf' % fake_conf['FAKE_NUMERO_DOSSIER'],
}]
}
fake_resp = Response()
@ -600,7 +526,8 @@ def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
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)
jresp = atreal_openads.create_dossier(request_1, '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'
@ -611,7 +538,7 @@ def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
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(req, 'DIA', collectivite_1.openADS_id, now=now_open)
atreal_openads.create_dossier(request_1, 'DIA', 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({})))
@ -621,7 +548,7 @@ def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
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(req, 'DIA', collectivite_1.openADS_id, now=now_open)
atreal_openads.create_dossier(request_1, 'DIA', 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}'
@ -629,23 +556,24 @@ def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
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(req, 'DIA', collectivite_1.openADS_id, now=now_open)
atreal_openads.create_dossier(request_1, 'DIA', 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(req, 'DIA', collectivite_1.openADS_id, now=now_open)
atreal_openads.create_dossier(request_1, 'DIA', collectivite_1.openADS_id, now=now_open)
# pylint: disable=protected-access
assert unicode(exception.value) == u'No JSON content returned: %r' % fake_resp._content
job = Job.objects.filter(natural_id=FAKE_NUMERO_DOSSIER).last() # pylint: disable=no-member
# 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_NUMERO_DOSSIER
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
@ -655,11 +583,11 @@ def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
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_NUMERO_DOSSIER
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_NUMERO_DOSSIER
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'}
@ -679,7 +607,7 @@ def test_create_dossier(atreal_openads, collectivite_1, collectivite_1_guichet):
# pylint: disable=unused-argument,redefined-outer-name
def test_get_dossier(atreal_openads):
def test_get_dossier(fake_conf, atreal_openads):
"""Test the method AtrealOpenads.get_dossier()."""
fake_resp_bad = Response()
@ -689,7 +617,7 @@ def test_get_dossier(atreal_openads):
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_NUMERO_DOSSIER)
atreal_openads.get_dossier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'])
assert unicode(exception.value) == "HTTP error: 502"
fake_resp_json = {
@ -707,7 +635,7 @@ def test_get_dossier(atreal_openads):
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_NUMERO_DOSSIER)
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']
@ -718,7 +646,7 @@ def test_get_dossier(atreal_openads):
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_NUMERO_DOSSIER)
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
@ -735,36 +663,39 @@ def test_get_dossier(atreal_openads):
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_NUMERO_DOSSIER)
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(atreal_openads):
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_NUMERO_DOSSIER, fichier_id='not an integer')
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_NUMERO_DOSSIER, fichier_id=18)
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_NUMERO_DOSSIER, 18))
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_NUMERO_DOSSIER, fichier_id=None)
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,
TEST_FILE_CERFA_DIA,
FAKE_NUMERO_DOSSIER,
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_NUMERO_DOSSIER, fichier_id=None)
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
@ -776,7 +707,8 @@ def test_get_fwd_files(atreal_openads):
assert jresp[0]['last_update_datetime'] == forwardfile.last_update_datetime
# pylint: disable=no-member
jresp = atreal_openads.get_fwd_files(None, FAKE_NUMERO_DOSSIER, fichier_id=forwardfile.id)
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
@ -788,22 +720,23 @@ def test_get_fwd_files(atreal_openads):
# pylint: disable=unused-argument,redefined-outer-name
def test_get_fwd_files_status(atreal_openads):
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_NUMERO_DOSSIER, fichier_id=18)
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,
TEST_FILE_CERFA_DIA,
FAKE_NUMERO_DOSSIER,
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_NUMERO_DOSSIER, fichier_id=None)
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
@ -816,7 +749,7 @@ def test_get_fwd_files_status(atreal_openads):
assert not jresp['failed']
# pylint: disable=no-member
jresp = atreal_openads.get_fwd_files_status(None, FAKE_NUMERO_DOSSIER,
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']
@ -831,7 +764,7 @@ def test_get_fwd_files_status(atreal_openads):
# pylint: disable=unused-argument,redefined-outer-name
def test_get_courrier(atreal_openads):
def test_get_courrier(fake_conf, atreal_openads):
"""Test the method AtrealOpenads.get_courrier()."""
lettre_type = 'dia_renonciation_preempter'
@ -843,14 +776,14 @@ def test_get_courrier(atreal_openads):
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_NUMERO_DOSSIER, lettre_type)
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(TEST_FILE_CERFA_DIA)
'b64_content': get_file_data(fake_conf['TEST_FILE_CERFA_DIA'])
}]
}
fake_resp = Response()
@ -861,7 +794,8 @@ def test_get_courrier(atreal_openads):
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_NUMERO_DOSSIER, lettre_type)
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']
@ -871,14 +805,14 @@ def test_get_courrier(atreal_openads):
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_NUMERO_DOSSIER, lettre_type)
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_NUMERO_DOSSIER, lettre_type)
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
@ -919,42 +853,32 @@ def test_get_response_error(atreal_openads):
# pylint: disable=unused-argument,redefined-outer-name
# pylint: disable=too-many-statements
def test_upload_user_files(atreal_openads):
def test_upload_user_files(fake_conf, atreal_openads, request_1):
"""Test the method AtrealOpenads.upload_user_files()."""
req = HttpRequest()
req._body = '' # pylint: disable=protected-access
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 # pylint: disable=protected-access
request_1.path = '/upload_user_files'
with pytest.raises(ForwardFile.DoesNotExist) as exception: # pylint: disable=no-member
atreal_openads.upload_user_files(req, 'DIA', FAKE_NUMERO_DOSSIER, file_ids=[999])
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(req, 'DIA', FAKE_NUMERO_DOSSIER,
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(req, 'DIA', FAKE_NUMERO_DOSSIER,
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,
TEST_FILE_CERFA_DIA,
FAKE_NUMERO_DOSSIER,
fake_conf['TEST_FILE_CERFA_DIA'],
fake_conf['FAKE_NUMERO_DOSSIER'],
'cerfa')
forwardfile.save()
assert isinstance(forwardfile, ForwardFile)
@ -969,7 +893,7 @@ def test_upload_user_files(atreal_openads):
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(req, 'DIA', FAKE_NUMERO_DOSSIER,
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'
@ -994,7 +918,10 @@ def test_upload_user_files(atreal_openads):
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(req, 'DIA', FAKE_NUMERO_DOSSIER, file_ids=[file_id])
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
@ -1006,18 +933,19 @@ def test_upload_user_files(atreal_openads):
# pylint: disable=protected-access
assert ffup.upload_msg == u'No JSON content returned: %r' % fake_resp._content
jresp = atreal_openads.upload_user_files(req, 'DIA', FAKE_NUMERO_DOSSIER)
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_NUMERO_DOSSIER
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(req, 'DIA', FAKE_NUMERO_DOSSIER)
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

View File

@ -20,14 +20,8 @@
"""Testing forms."""
import os
import base64
import datetime
import pytest
from django.core.files import File
from atreal_openads.forms import (
ForwardFileForm,
CollectiviteForm,
@ -35,94 +29,24 @@ from atreal_openads.forms import (
)
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
# pylint: disable=unused-argument,invalid-name
def atreal_openads(db):
"""Return an instance of a connector AtrealOpenads."""
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
# pylint: disable=unused-argument,redefined-outer-name,invalid-name
def collectivite_1(db, atreal_openads):
"""Return an instance of a 'Collectivite'."""
return Collectivite.objects.create( # pylint: disable=no-member
name=u'Macollectivité',
connecteur=atreal_openads,
openADS_id='3'
)
@pytest.fixture
# pylint: disable=unused-argument,redefined-outer-name,invalid-name
def collectivite_1_guichet(db, atreal_openads, collectivite_1):
"""Return an instance of a 'Guichet'."""
return Guichet.objects.create( # pylint: disable=no-member
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)
)
# pylint: disable=unused-argument,redefined-outer-name
def test_forwardfile_form(atreal_openads, collectivite_1):
def test_forwardfile_form(forwardfile_2, atreal_openads, collectivite_1):
"""Test for ForwardFileForm."""
form = ForwardFileForm()
assert form.instance is not None
forwardfile = ForwardFile(
connecteur=None,
collectivite=None,
numero_demande='45641531',
numero_dossier=FAKE_NUMERO_DOSSIER,
type_fichier='CERFA',
orig_filename=os.path.basename(TEST_FILE_CERFA_DIA),
content_type='application/pdf',
file_hash='ffdf456fdsvgb4bgfb6g4f5b',
upload_file=File(open(TEST_FILE_CERFA_DIA, 'r')),
upload_status='pending'
)
form_with_instance = ForwardFileForm(instance=forwardfile, collectivite=collectivite_1)
assert form_with_instance.instance is forwardfile
form_with_instance = ForwardFileForm(instance=forwardfile_2, collectivite=collectivite_1)
assert form_with_instance.instance is forwardfile_2
assert form_with_instance.instance.collectivite is collectivite_1
form_with_instance = ForwardFileForm(instance=forwardfile, connecteur=atreal_openads)
assert form_with_instance.instance is forwardfile
form_with_instance = ForwardFileForm(instance=forwardfile_2, connecteur=atreal_openads)
assert form_with_instance.instance is forwardfile_2
assert form_with_instance.instance.connecteur is atreal_openads
# TODO check the queryset of the collectivite

View File

@ -20,16 +20,11 @@
"""Testing utilities functions."""
import os
import base64
import re
import datetime
import pytest
from django.core.files import File
from django.core.files.base import ContentFile
from atreal_openads.utils import (
to_dash_case,
force_encoded_string_output,
@ -44,68 +39,6 @@ from atreal_openads.utils import (
DictDumper
)
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
# pylint: disable=unused-argument,invalid-name
def atreal_openads(db):
"""Return an instance of a connector AtrealOpenads."""
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
# pylint: disable=unused-argument,redefined-outer-name,invalid-name
def collectivite_1(db, atreal_openads):
"""Return an instance of a 'Collectivite'."""
return Collectivite.objects.create( # pylint: disable=no-member
name=u'Macollectivité',
connecteur=atreal_openads,
openADS_id='3'
)
@pytest.fixture
# pylint: disable=unused-argument,redefined-outer-name,invalid-name
def collectivite_1_guichet(db, atreal_openads, collectivite_1):
"""Return an instance of a 'Guichet'."""
return Guichet.objects.create( # pylint: disable=no-member
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 test_to_dash_case():
"""Test for function 'to_dash_case()'."""
@ -177,39 +110,29 @@ def test_normalize():
assert normalize(astring) == 'a a b b c c d d'
def test_get_file_data():
def test_get_file_data(fake_conf):
"""Test for function 'get_file_data()'."""
assert get_file_data(TEST_FILE_CERFA_DIA) == base64.b64encode(open(TEST_FILE_CERFA_DIA).read())
assert get_file_data(TEST_FILE_CERFA_DIA, b64=False) == open(TEST_FILE_CERFA_DIA).read()
assert get_file_data(fake_conf['TEST_FILE_CERFA_DIA']) == base64.b64encode(
open(fake_conf['TEST_FILE_CERFA_DIA']).read())
assert get_file_data(fake_conf['TEST_FILE_CERFA_DIA'], b64=False) == open(
fake_conf['TEST_FILE_CERFA_DIA']).read()
def test_get_file_digest():
def test_get_file_digest(fake_conf):
"""Test for function 'get_file_digest()'."""
with open(TEST_FILE_CERFA_DIA) as file_pt:
with open(fake_conf['TEST_FILE_CERFA_DIA']) as file_pt:
assert get_file_digest(file_pt) == ('cc90a620982760fdee16a5b4fe1b5ac3'
'b4fe868fd02d2f70b27f1e46d283ea51')
def test_get_upload_path():
def test_get_upload_path(forwardfile_2):
"""Test for function 'get_upload_path()'."""
forwardfile = ForwardFile(
numero_demande='45641531',
numero_dossier=FAKE_NUMERO_DOSSIER,
type_fichier='CERFA',
orig_filename=os.path.basename(TEST_FILE_CERFA_DIA),
content_type='application/pdf',
file_hash='ffdf456fdsvgb4bgfb6g4f5b',
upload_file=File(open(TEST_FILE_CERFA_DIA, 'r')),
upload_status='pending',
connecteur=None,
collectivite=None
)
regex = r"^to_openADS__%s__%s\.pdf$" % (
'[0-9]{4}-[0-9]{2}-[0-9]{2}_[0-9]{2}h[0-9]{2}m[0-9]{2}s[0-9]+', 'ffdf')
assert re.search(regex, get_upload_path(forwardfile))
assert re.search(regex, get_upload_path(forwardfile_2))
def test_get_file_extension():
@ -290,40 +213,28 @@ def test_dict_dumper():
# pylint: disable=unused-argument,redefined-outer-name
def test_base_model(atreal_openads, collectivite_1, collectivite_1_guichet):
def test_base_model(fake_conf, atreal_openads, collectivite_1, collectivite_1_guichet,
forwardfile_1):
"""Test for methods of class 'BaseModel' through instance of a ForwardFile."""
forwardfile = ForwardFile(
numero_demande='45641531',
numero_dossier=FAKE_NUMERO_DOSSIER,
type_fichier='CERFA',
orig_filename=os.path.basename(TEST_FILE_CERFA_DIA),
content_type='application/pdf',
file_hash='ffdf456fdsvgb4bgfb6g4f5b',
upload_file=ContentFile('toto'),
upload_status='pending',
connecteur=atreal_openads,
collectivite=None
)
assert forwardfile_1.get_verbose_name() == 'Forward File'
assert forwardfile_1.get_verbose_name_plural() == 'Forward Files'
assert forwardfile.get_verbose_name() == 'Forward File'
assert forwardfile.get_verbose_name_plural() == 'Forward Files'
assert forwardfile_1.get_class_name() == 'ForwardFile'
assert forwardfile_1.get_class_name_plural() == 'ForwardFiles'
assert forwardfile.get_class_name() == 'ForwardFile'
assert forwardfile.get_class_name_plural() == 'ForwardFiles'
assert forwardfile_1.get_class_name_dash_case() == 'forward-file'
assert forwardfile_1.get_class_name_plural_dash_case() == 'forward-files'
assert forwardfile.get_class_name_dash_case() == 'forward-file'
assert forwardfile.get_class_name_plural_dash_case() == 'forward-files'
assert forwardfile_1.get_class_name_title() == 'Forward File'
assert forwardfile_1.get_class_name_plural_title() == 'Forward Files'
assert forwardfile.get_class_name_title() == 'Forward File'
assert forwardfile.get_class_name_plural_title() == 'Forward Files'
assert forwardfile_1.get_url_name('list', plural=True) == 'list-forward-files'
assert forwardfile.get_url_name('list', plural=True) == 'list-forward-files'
assert forwardfile.get_absolute_url() == '/manage/atreal-openads/atreal/forward-file/None'
assert forwardfile.get_edit_url() == '/manage/atreal-openads/atreal/edit-forward-file/None'
assert forwardfile.get_delete_url() == '/manage/atreal-openads/atreal/delete-forward-file/None'
assert forwardfile.get_list_url() == '/manage/atreal-openads/atreal/forward-files'
assert forwardfile_1.get_absolute_url() == '/manage/atreal-openads/atreal/forward-file/1'
assert forwardfile_1.get_edit_url() == '/manage/atreal-openads/atreal/edit-forward-file/1'
assert forwardfile_1.get_delete_url() == '/manage/atreal-openads/atreal/delete-forward-file/1'
assert forwardfile_1.get_list_url() == '/manage/atreal-openads/atreal/forward-files'
assert atreal_openads.get_class_name_plural() == 'AtrealOpenads'

View File

@ -20,15 +20,7 @@
"""Testing views."""
import os
import base64
import datetime
import pytest
from django.http.request import HttpRequest, QueryDict
from django.urls.base import resolve
from django.core.files import File
from atreal_openads.views import (
get_connecteur_from_request,
@ -49,106 +41,19 @@ from atreal_openads.views import (
GuichetDeleteView
)
from atreal_openads.models import (
ForwardFile,
Guichet,
Collectivite,
AtrealOpenads
)
from atreal_openads.models import Collectivite
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
# pylint: disable=unused-argument,invalid-name
def atreal_openads(db):
"""Return an instance of a connector AtrealOpenads."""
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
# pylint: disable=unused-argument,redefined-outer-name,invalid-name
def collectivite_1(db, atreal_openads):
"""Return an instance of a 'Collectivite'."""
return Collectivite.objects.create( # pylint: disable=no-member
name=u'Macollectivité',
connecteur=atreal_openads,
openADS_id='3'
)
@pytest.fixture
# pylint: disable=unused-argument,redefined-outer-name,invalid-name
def collectivite_1_guichet(db, atreal_openads, collectivite_1):
"""Return an instance of a 'Guichet'."""
return Guichet.objects.create( # pylint: disable=no-member
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)
)
@pytest.fixture
# pylint: disable=unused-argument,redefined-outer-name,invalid-name
def forwardfile_1(db, atreal_openads, collectivite_1):
"""Return an instance of a 'ForwardFile'."""
return ForwardFile.objects.create( # pylint: disable=no-member
connecteur=atreal_openads,
collectivite=collectivite_1,
numero_demande='45641531',
numero_dossier=FAKE_NUMERO_DOSSIER,
type_fichier='CERFA',
orig_filename=os.path.basename(TEST_FILE_CERFA_DIA),
content_type='application/pdf',
file_hash='ffdf456fdsvgb4bgfb6g4f5b',
upload_file=File(open(TEST_FILE_CERFA_DIA, 'r')),
upload_status='pending'
)
# pylint: disable=unused-argument,redefined-outer-name
def test_get_connecteur_from_request(atreal_openads, forwardfile_1): # pylint: disable=invalid-name
def test_get_connecteur_from_request(atreal_openads, forwardfile_1, request_1):
"""Test for function 'get_connecteur_from_request()'."""
req = HttpRequest()
req.path = '/manage/atreal-openads/%s/forward-file/%s' % (
request_1.path = '/manage/atreal-openads/%s/forward-file/%s' % (
atreal_openads.slug, forwardfile_1.id)
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_params = None
req.COOKIES = {}
req.META = {}
req._read_started = False # pylint: disable=protected-access
req.resolver_match = resolve(req.path)
request_1.resolver_match = resolve(request_1.path)
view = ForwardFileView()
view.request = req
view.request = request_1
connecteur = get_connecteur_from_request(view)
assert connecteur is not None
@ -156,24 +61,15 @@ def test_get_connecteur_from_request(atreal_openads, forwardfile_1): # pylint:
# pylint: disable=unused-argument,redefined-outer-name
def test_get_collectivite_from_request(atreal_openads, collectivite_1): # noqa: E501, pylint: disable=invalid-name
def test_get_collectivite_from_request(atreal_openads, collectivite_1, request_1): # noqa: E501, pylint: disable=invalid-name
"""Test for function 'get_collectivite_from_request()'."""
req = HttpRequest()
req.path = '/manage/atreal-openads/%s/collectivite/%s/forward-files' % (
request_1.path = '/manage/atreal-openads/%s/collectivite/%s/forward-files' % (
atreal_openads.slug, collectivite_1.id)
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_params = None
req.COOKIES = {}
req.META = {}
req._read_started = False # pylint: disable=protected-access
req.resolver_match = resolve(req.path)
request_1.resolver_match = resolve(request_1.path)
view = ForwardFileListView()
view.request = req
view.request = request_1
collectivite = get_collectivite_from_request(view)
assert collectivite is not None
@ -181,70 +77,61 @@ def test_get_collectivite_from_request(atreal_openads, collectivite_1): # noqa:
# pylint: disable=too-many-statements
def test_forwardfile_view(atreal_openads, collectivite_1, forwardfile_1):
def test_forwardfile_view(atreal_openads, collectivite_1, forwardfile_1, request_1):
"""Test for views 'ForwardFile*View'."""
req = HttpRequest()
req.path = '/manage/atreal-openads/%s/forward-file/%s' % (
request_1.path = '/manage/atreal-openads/%s/forward-file/%s' % (
atreal_openads.slug, forwardfile_1.id)
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_params = None
req.COOKIES = {}
req.META = {}
req._read_started = False # pylint: disable=protected-access
req.resolver_match = resolve(req.path)
request_1.resolver_match = resolve(request_1.path)
view = ForwardFileView()
view.request = req
view.request = request_1
view.object = None
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
view = ForwardFileUpdateView()
view.request = req
view.request = request_1
view.object = None
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
url = view.get_success_url()
assert url == u'/manage/atreal-openads/%s/forward-file/%s' % (
atreal_openads.slug, forwardfile_1.id)
req.GET['back-to'] = 'list-forward-files'
request_1.GET['back-to'] = 'list-forward-files'
url = view.get_success_url()
assert url == u'/manage/atreal-openads/%s/forward-files' % atreal_openads.slug
req.GET['back-to'] = 'col-list-forward-files'
request_1.GET['back-to'] = 'col-list-forward-files'
url = view.get_success_url()
assert url == u'/manage/atreal-openads/%s/collectivite/%s/forward-files' % (
atreal_openads.slug, collectivite_1.id)
view = ForwardFileDeleteView()
view.request = req
view.request = request_1
view.object = None
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
del req.GET['back-to']
del request_1.GET['back-to']
url = view.get_success_url()
assert url == u'/atreal-openads/%s/' % atreal_openads.slug
req.GET['back-to'] = 'list-forward-files'
request_1.GET['back-to'] = 'list-forward-files'
url = view.get_success_url()
assert url == u'/manage/atreal-openads/%s/forward-files' % atreal_openads.slug
req.GET['back-to'] = 'col-list-forward-files'
request_1.GET['back-to'] = 'col-list-forward-files'
url = view.get_success_url()
assert url == u'/manage/atreal-openads/%s/collectivite/%s/forward-files' % (
atreal_openads.slug, collectivite_1.id)
req.path = '/manage/atreal-openads/%s/collectivite/%s/forward-files' % (
request_1.path = '/manage/atreal-openads/%s/collectivite/%s/forward-files' % (
atreal_openads.slug, collectivite_1.id)
req.resolver_match = resolve(req.path)
request_1.resolver_match = resolve(request_1.path)
view = ForwardFileListView()
view.request = req
view.request = request_1
view.object_list = []
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
assert context['collectivite'].id == collectivite_1.id
@ -256,20 +143,20 @@ def test_forwardfile_view(atreal_openads, collectivite_1, forwardfile_1):
assert queryset.query.get_meta().ordering == ['-last_update_datetime']
assert queryset.ordered
req.GET['order-by'] = '-id'
request_1.GET['order-by'] = '-id'
queryset = view.get_queryset()
assert queryset.query is not None
assert queryset.query.order_by == ['-id']
assert queryset.query.default_ordering
req.path = '/manage/atreal-openads/%s/forward-files' % atreal_openads.slug
req.resolver_match = resolve(req.path)
del req.GET['back-to']
del req.GET['order-by']
request_1.path = '/manage/atreal-openads/%s/forward-files' % atreal_openads.slug
request_1.resolver_match = resolve(request_1.path)
del request_1.GET['back-to']
del request_1.GET['order-by']
view = ForwardFileListView()
view.request = req
view.request = request_1
view.object_list = []
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
@ -282,26 +169,17 @@ def test_forwardfile_view(atreal_openads, collectivite_1, forwardfile_1):
# pylint: disable=too-many-statements
def test_collectivite_view(atreal_openads, collectivite_1, forwardfile_1):
def test_collectivite_view(atreal_openads, collectivite_1, forwardfile_1, request_1):
"""Test for views 'Collectivite*View'."""
req = HttpRequest()
req.path = '/manage/atreal-openads/%s/collectivite/%s' % (
request_1.path = '/manage/atreal-openads/%s/collectivite/%s' % (
atreal_openads.slug, collectivite_1.id)
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_params = None
req.COOKIES = {}
req.META = {}
req._read_started = False # pylint: disable=protected-access
req.resolver_match = resolve(req.path)
request_1.resolver_match = resolve(request_1.path)
view = CollectiviteView()
view.request = req
view.request = request_1
view.object = None
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
assert context['guichet_add_url'] == (
@ -312,54 +190,54 @@ def test_collectivite_view(atreal_openads, collectivite_1, forwardfile_1):
atreal_openads.slug, collectivite_1.id))
view = CollectiviteUpdateView()
view.request = req
view.request = request_1
view.object = None
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
url = view.get_success_url()
assert url == u'/manage/atreal-openads/%s/collectivite/%s' % (
atreal_openads.slug, collectivite_1.id)
req.GET['back-to'] = 'list-collectivites'
request_1.GET['back-to'] = 'list-collectivites'
url = view.get_success_url()
assert url == u'/manage/atreal-openads/%s/collectivites' % atreal_openads.slug
view = CollectiviteDeleteView()
view.request = req
view.request = request_1
view.object = None
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
del req.GET['back-to']
del request_1.GET['back-to']
url = view.get_success_url()
assert url == u'/atreal-openads/%s/' % atreal_openads.slug
req.GET['back-to'] = 'list-collectivites'
request_1.GET['back-to'] = 'list-collectivites'
url = view.get_success_url()
assert url == u'/manage/atreal-openads/%s/collectivites' % atreal_openads.slug
view = CollectiviteCreateView()
req.path = '/manage/atreal-openads/%s/create-collectivite' % atreal_openads.slug
req.resolver_match = resolve(req.path)
view.request = req
request_1.path = '/manage/atreal-openads/%s/create-collectivite' % atreal_openads.slug
request_1.resolver_match = resolve(request_1.path)
view.request = request_1
view.object = None
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
kwargs = view.get_form_kwargs()
assert kwargs['connecteur'].slug == atreal_openads.slug
del req.GET['back-to']
del request_1.GET['back-to']
url = view.get_success_url()
assert url == u'/atreal-openads/%s/' % atreal_openads.slug
req.GET['back-to'] = 'list-collectivites'
request_1.GET['back-to'] = 'list-collectivites'
url = view.get_success_url()
assert url == u'/manage/atreal-openads/%s/collectivites' % atreal_openads.slug
req.path = '/manage/atreal-openads/%s/collectivites' % atreal_openads.slug
req.resolver_match = resolve(req.path)
request_1.path = '/manage/atreal-openads/%s/collectivites' % atreal_openads.slug
request_1.resolver_match = resolve(request_1.path)
view = CollectiviteListView()
view.request = req
view.request = request_1
view.object_list = []
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
assert context['collectivite_add_url'] == (
@ -372,49 +250,40 @@ def test_collectivite_view(atreal_openads, collectivite_1, forwardfile_1):
assert queryset.query.get_meta().ordering == ['name']
assert queryset.ordered
req.GET['order-by'] = '-id'
request_1.GET['order-by'] = '-id'
queryset = view.get_queryset()
assert queryset.query is not None
assert queryset.query.order_by == ['-id']
assert queryset.query.default_ordering
def test_guichet_view(atreal_openads, collectivite_1, collectivite_1_guichet):
def test_guichet_view(atreal_openads, collectivite_1, collectivite_1_guichet, request_1):
"""Test for views 'Guichet*View'."""
req = HttpRequest()
req.path = '/manage/atreal-openads/%s/collectivite/%s/guichet/%s' % (
request_1.path = '/manage/atreal-openads/%s/collectivite/%s/guichet/%s' % (
atreal_openads.slug, collectivite_1.id, collectivite_1_guichet.id)
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_params = None
req.COOKIES = {}
req.META = {}
req._read_started = False # pylint: disable=protected-access
req.resolver_match = resolve(req.path)
request_1.resolver_match = resolve(request_1.path)
view = GuichetView()
view.request = req
view.request = request_1
view.object = None
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
assert context['collectivite'].id == collectivite_1.id
view = GuichetUpdateView()
view.request = req
view.request = request_1
view.object = None
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
assert context['collectivite'].id == collectivite_1.id
view = GuichetDeleteView()
view.request = req
view.request = request_1
view.object = None
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
assert context['collectivite'].id == collectivite_1.id
@ -423,12 +292,12 @@ def test_guichet_view(atreal_openads, collectivite_1, collectivite_1_guichet):
atreal_openads.slug, collectivite_1.id)
view = GuichetCreateView()
req.path = '/manage/atreal-openads/%s/collectivite/%s/create-guichet' % (
request_1.path = '/manage/atreal-openads/%s/collectivite/%s/create-guichet' % (
atreal_openads.slug, collectivite_1.id)
req.resolver_match = resolve(req.path)
view.request = req
request_1.resolver_match = resolve(request_1.path)
view.request = request_1
view.object = None
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['connecteur'].slug == atreal_openads.slug
assert context['collectivite'].id == collectivite_1.id
@ -439,25 +308,16 @@ def test_guichet_view(atreal_openads, collectivite_1, collectivite_1_guichet):
atreal_openads.slug, collectivite_1.id)
def test_connecteur_view(atreal_openads):
def test_connecteur_view(atreal_openads, request_1):
"""Test for views 'AtrealOpenadsView'."""
req = HttpRequest()
req.path = '/atreal-openads/%s/' % atreal_openads.slug
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_params = None
req.COOKIES = {}
req.META = {}
req._read_started = False # pylint: disable=protected-access
req.resolver_match = resolve(req.path)
request_1.path = '/atreal-openads/%s/' % atreal_openads.slug
request_1.resolver_match = resolve(request_1.path)
view = AtrealOpenadsView()
view.request = req
view.request = request_1
view.object = atreal_openads
view.kwargs = req.resolver_match.kwargs
view.kwargs = request_1.resolver_match.kwargs
context = view.get_context_data()
assert context['collectivite_fields'] == Collectivite.get_fields()
assert context['collectivite_add_url'] == (