adapt tests to python3 (#39533)

This commit is contained in:
Emmanuel Cazenave 2020-02-04 17:28:28 +01:00
parent 60d2884f8c
commit 351a1a6242
3 changed files with 103 additions and 92 deletions

View File

@ -33,6 +33,7 @@ import django_webtest
from django.core.cache import cache
from django.core.files import File
from django.http.request import HttpRequest, QueryDict
from django.utils.encoding import force_text
from atreal_openads.models import (
ForwardFile,
@ -115,11 +116,11 @@ def fake_conf():
'CONNECTOR_SLUG': 'atreal',
'COLLECTIVITE': 79,
'OPENADS_API_LOGIN': 'publik-passerelle',
'OPENADS_API_PASSWORD': base64.urlsafe_b64encode(os.urandom(20)),
'OPENADS_API_PASSWORD': force_text(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)),
'FAKE_COOKIE_CRSF': force_text(base64.urlsafe_b64encode(os.urandom(20))),
'FAKE_NUMERO_DOSSIER': force_text(base64.urlsafe_b64encode(os.urandom(10))),
'TESTS_DIR': os.path.dirname(__file__)
}
@ -181,7 +182,7 @@ def forwardfile_1(fake_conf, db, atreal_openads, collectivite_1):
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_file=File(open(fake_conf['TEST_FILE_CERFA_DIA'], 'rb')),
upload_status='pending'
)
@ -199,7 +200,7 @@ def forwardfile_2(fake_conf, connecteur=None, collectivite=None):
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_file=File(open(fake_conf['TEST_FILE_CERFA_DIA'], 'rb')),
upload_status='pending'
)

View File

@ -26,6 +26,7 @@ import datetime
import json
import os
import re
import sys
import magic
import mock
@ -38,6 +39,7 @@ from django.http.response import JsonResponse
from django.core.files import File
# from django.db.models.query import QuerySet
from django.core.exceptions import ValidationError
from django.utils.encoding import force_bytes, force_text
from passerelle.utils.jsonresponse import APIError
from passerelle.base.models import Job
@ -54,6 +56,13 @@ from atreal_openads.models import (
)
def force_encoded_string_output(text, default_enc='utf-8'):
"""Decorator function that return the result converted to str type."""
if sys.version_info.major < 3:
return text.encode(sys.stdout.encoding or default_enc)
return text
def build_forwardfile_from_path(connecteur, path, numero_dossier, type_fichier):
"""Convert a file path to a ForwardFile."""
if path:
@ -65,9 +74,9 @@ def build_forwardfile_from_path(connecteur, path, numero_dossier, type_fichier):
fwd_file.type_fichier = type_fichier
fwd_file.orig_filename = os.path.basename(path)
fwd_file.content_type = magic.from_file(path, mime=True)
with open(path, 'r') as file_pt:
with open(path, 'rb') as file_pt:
fwd_file.file_hash = get_file_digest(file_pt)
fwd_file.upload_file = File(open(path, 'r'))
fwd_file.upload_file = File(open(path, 'rb'))
fwd_file.upload_status = 'pending'
return fwd_file
return None
@ -86,15 +95,15 @@ def test_forward_file(forwardfile_2, atreal_openads):
assert repr(forwardfile_2) == (
u'ForwardFile(id=%s,connecteur=%s,collectivite=%s'
',demande=%s,dossier=%s,type=%s,filename=%s,status=%s)' % (
forwardfile_2.id, unicode(forwardfile_2.connecteur), None, # pylint: disable=no-member
forwardfile_2.id, force_text(forwardfile_2.connecteur), None, # pylint: disable=no-member
forwardfile_2.numero_demande, forwardfile_2.numero_dossier,
forwardfile_2.type_fichier, forwardfile_2.orig_filename, forwardfile_2.upload_status
)
).encode('utf-8')
)
assert str(forwardfile_2) == '%s[%s]' % (trunc_str_values(forwardfile_2.orig_filename, 20),
'Pending')
assert unicode(forwardfile_2) == u'%s[%s]' % (
assert force_text(forwardfile_2) == u'%s[%s]' % (
trunc_str_values(forwardfile_2.orig_filename, 20),
'Pending')
@ -119,7 +128,7 @@ def test_forward_file(forwardfile_2, atreal_openads):
with pytest.raises(ValueError) as exception:
forwardfile_2.upload_file.size # pylint: disable=pointless-statement
assert unicode(exception.value) == "The 'upload_file' attribute has no file associated with it."
assert force_text(exception.value) == "The 'upload_file' attribute has no file associated with it."
assert forwardfile_2.size > 0
assert forwardfile_2.file_hash == ('cc90a620982760fdee16a5b4fe1b5ac3'
'b4fe868fd02d2f70b27f1e46d283ea51')
@ -136,7 +145,7 @@ def test_forward_file(forwardfile_2, atreal_openads):
forwardfile_2.save()
assert len(exception.value.messages) == 1
assert '__all__' in exception.value.message_dict
assert unicode(exception.value.message_dict['__all__'][0]) == (
assert force_text(exception.value.message_dict['__all__'][0]) == (
u"A %s cannot have all the following fields empty: %s." % (
forwardfile_2.get_verbose_name(),
['file_hash', 'orig_filename', 'upload_file']))
@ -150,16 +159,16 @@ def test_collectivite(collectivite_1, collectivite_1_guichet):
col = collectivite_1
assert repr(col) == (
assert repr(col) == force_encoded_string_output(
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
1, force_text(col.name), force_text(col.connecteur), col.openADS_id,
force_text(col.guichet) if hasattr(col, 'guichet') else None
)
).encode('utf-8')
)
assert str(col) == col.name.encode('utf-8')
assert str(col) == force_encoded_string_output(col.name)
assert unicode(col) == col.name
assert force_text(col) == col.name
class_fields = Collectivite.get_fields()
assert len(class_fields) == 6
@ -195,22 +204,22 @@ def test_guichet(collectivite_1_guichet):
guichet = collectivite_1_guichet
assert repr(guichet) == (
assert repr(guichet) == force_encoded_string_output(
u'Guichet(id=%s,collectivite=%s,%s)' % (
1, unicode(guichet.collectivite), unicode(guichet)
1, force_text(guichet.collectivite), force_text(guichet)
)
).encode('utf-8')
)
assert str(guichet) == u'Monday 08:30 -> Friday 12:15 [09:00/17:00]'.encode('utf-8')
assert str(guichet) == force_encoded_string_output(u'Monday 08:30 -> Friday 12:15 [09:00/17:00]')
assert unicode(guichet) == u'Monday 08:30 -> Friday 12:15 [09:00/17:00]'
assert force_text(guichet) == u'Monday 08:30 -> Friday 12:15 [09:00/17:00]'
params = guichet.get_url_params()
assert params['collectivite'] == guichet.collectivite.id
with pytest.raises(Exception) as exception:
guichet.get_list_url()
assert unicode(exception.value) == u"Guichet:get_list_url() method should not be called"
assert force_text(exception.value) == u"Guichet:get_list_url() method should not be called"
# pylint: disable=unused-argument,redefined-outer-name
@ -247,7 +256,7 @@ def test_guichet_is_open(collectivite_1_guichet): # pylint: disable=too-many-lo
with pytest.raises(TypeError) as exception:
guichet.is_open('invalid datetime')
assert unicode(exception.value) == u"is_open() expect a datetime object (not a %s)" % type('')
assert force_text(exception.value) == u"is_open() expect a datetime object (not a %s)" % type('')
assert not guichet.is_open(None)
@ -272,21 +281,21 @@ def test_get_files_from_payload(atreal_openads):
with pytest.raises(APIError) as exception:
atreal_openads.get_files_from_payload({})
assert unicode(exception.value) == u"Expecting '%s' key in JSON %s" % ('files', title)
assert force_text(exception.value) == u"Expecting '%s' key in JSON %s" % ('files', title)
with pytest.raises(APIError) as exception:
atreal_openads.get_files_from_payload({'files': 'invalid'})
assert unicode(exception.value) == u"Expecting '%s' value in JSON %s to be a %s (not a %s)" % (
assert force_text(exception.value) == u"Expecting '%s' value in JSON %s to be a %s (not a %s)" % (
'files', title, 'list', type(''))
with pytest.raises(APIError) as exception:
atreal_openads.get_files_from_payload({'files': {'i': 'invalid'}})
assert unicode(exception.value) == u"Expecting '%s' value in JSON %s to be a %s (not a %s)" % (
assert force_text(exception.value) == u"Expecting '%s' value in JSON %s to be a %s (not a %s)" % (
'files', title, 'list', type({}))
with pytest.raises(APIError) as exception:
atreal_openads.get_files_from_payload({'files': []})
assert unicode(exception.value) == u"Expecting non-empty '%s' value in JSON %s" % (
assert force_text(exception.value) == u"Expecting non-empty '%s' value in JSON %s" % (
'files', title)
@ -303,7 +312,7 @@ def test_check_file_dict(fake_conf, atreal_openads):
}
d64 = {
'b64_content': get_file_data(fake_conf['TEST_FILE_CERFA_DIA'], b64=True),
'b64_content': force_text(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'
}
@ -314,26 +323,26 @@ def test_check_file_dict(fake_conf, atreal_openads):
dic['filename'] = {'a', 'filename'}
with pytest.raises(APIError) as exception:
atreal_openads.check_file_dict(dic, b64=False)
assert unicode(exception.value) == (
assert force_text(exception.value) == (
u"Expecting '%s' value in JSON %s in file dict to be a %s (not a %s)" % (
'file.filename', title, 'string', type(dic['filename'])))
dic['content'] = {'a', 'filename'}
with pytest.raises(APIError) as exception:
atreal_openads.check_file_dict(dic, b64=False)
assert unicode(exception.value) == (
assert force_text(exception.value) == (
u"Expecting '%s' value in JSON %s in file dict to be a %s (not a %s)" % (
'file.content', title, 'string', type(dic['content'])))
del dic['content']
with pytest.raises(APIError) as exception:
atreal_openads.check_file_dict(dic, b64=False)
assert unicode(exception.value) == u"Expecting 'file.%s' key in JSON %s" % ('content', title)
assert force_text(exception.value) == u"Expecting 'file.%s' key in JSON %s" % ('content', title)
del d64['b64_content']
with pytest.raises(APIError) as exception:
atreal_openads.check_file_dict(dic, b64=True)
assert unicode(exception.value) == (
assert force_text(exception.value) == (
u"Expecting 'file.%s' key in JSON %s" % ('b64_content', title))
@ -427,26 +436,26 @@ def test_create_dossier(fake_conf, atreal_openads, collectivite_1, collectivite_
# user attached files
"cerfa": {
"content": get_file_data(fake_conf['TEST_FILE_CERFA_DIA']),
"content": force_text(get_file_data(fake_conf['TEST_FILE_CERFA_DIA'])),
"content_type": "invalid/content type",
"field_id": "50",
"filename": os.path.basename(fake_conf['TEST_FILE_CERFA_DIA'])
},
"annexe_1_type_raw": "plan",
"annexe_1": {
"content": get_file_data(fake_conf['TEST_FILE_PLAN_CADASTRAL']),
"content": force_text(get_file_data(fake_conf['TEST_FILE_PLAN_CADASTRAL'])),
"content_type": "application/pdf",
"filename": os.path.basename(fake_conf['TEST_FILE_PLAN_CADASTRAL'])
},
"annexe_2_type_raw": "plan",
"annexe_2": {
"content": get_file_data(fake_conf['TEST_FILE_PLAN_CADASTRAL']),
"content": force_text(get_file_data(fake_conf['TEST_FILE_PLAN_CADASTRAL'])),
"content_type": "application/pdf",
# "filename": 'plan_cad'
},
"annexe_3_type_raw": "pouvoir",
"annexe_3": {
"content": get_file_data(fake_conf['TEST_FILE_CERFA_DIA']),
"content": force_text(get_file_data(fake_conf['TEST_FILE_CERFA_DIA'])),
"content_type": "application/pdf",
"filename": 'mandat'
},
@ -463,7 +472,7 @@ def test_create_dossier(fake_conf, atreal_openads, collectivite_1, collectivite_
with mock.patch('passerelle.utils.Request.post') as requests_post:
requests_post.return_value = fake_resp_bad
atreal_openads.create_dossier(request_1, 'DIA', collectivite='not an integer')
assert unicode(exception.value) == "invalid literal for int() with base 10: 'not an integer'"
assert force_text(exception.value) == "invalid literal for int() with base 10: 'not an integer'"
# guichet is open from Monday/8:30 to Friday/12:15, between 9:00 and 17:00
now_open = datetime.datetime(2019, 8, 7, 14, 0, 0) # wednesday
@ -474,7 +483,7 @@ def test_create_dossier(fake_conf, atreal_openads, collectivite_1, collectivite_
requests_post.return_value = fake_resp_bad
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
now=now_open)
assert unicode(exception.value) == "HTTP error: 502"
assert force_text(exception.value) == "HTTP error: 502"
# TODO update the code and return message when it will be
# correctly implemented in the openADS.API side.
@ -486,7 +495,7 @@ def test_create_dossier(fake_conf, atreal_openads, collectivite_1, collectivite_
with mock.patch('passerelle.utils.Request.post') as requests_post:
requests_post.return_value = fake_resp_404
atreal_openads.create_dossier(request_1, 'DIA', collectivite=999)
assert unicode(exception.value) == "HTTP error: 404"
assert force_text(exception.value) == "HTTP error: 404"
jresp = atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
now=now_closed)
@ -509,7 +518,7 @@ def test_create_dossier(fake_conf, atreal_openads, collectivite_1, collectivite_
jresp = atreal_openads.create_dossier(request_1, 'DIA',
collectivite=collectivite_1.openADS_id,
now=now_invalid)
assert unicode(exception.value) == (
assert force_text(exception.value) == (
u"Invalid value of type '%s' for now argument of endpoint '%s' (must be: %s)" % (
type(now_invalid),
'create_dossier',
@ -523,7 +532,7 @@ def test_create_dossier(fake_conf, atreal_openads, collectivite_1, collectivite_
requests_post.return_value = fake_resp_bad
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
now=now_open)
assert unicode(exception.value) == (
assert force_text(exception.value) == (
u"Type '%s' for file '%s' is too long (%d chars, but max is %d)" % (
fake_req_json['fields']['annexe_3_type_raw'],
'annexe_3',
@ -537,7 +546,7 @@ def test_create_dossier(fake_conf, atreal_openads, collectivite_1, collectivite_
requests_post.return_value = fake_resp_bad
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
now=now_open)
assert unicode(exception.value) == u"No type field/value for file '%s'" % 'annexe_3'
assert force_text(exception.value) == u"No type field/value for file '%s'" % 'annexe_3'
fake_req_json['fields']['annexe_3_type_raw'] = annexe_3_type_bak
request_1._body = json.dumps(fake_req_json) # pylint: disable=protected-access
@ -545,7 +554,7 @@ def test_create_dossier(fake_conf, atreal_openads, collectivite_1, collectivite_
fake_resp_json = {
'numero_dossier': fake_conf['FAKE_NUMERO_DOSSIER'],
'files': [{
'b64_content': get_file_data(fake_conf['TEST_FILE_CERFA_DIA']),
'b64_content': force_text(get_file_data(fake_conf['TEST_FILE_CERFA_DIA'])),
'content_type': 'text/plain',
'filename': 'recepisse_depot_%s.pdf' % fake_conf['FAKE_NUMERO_DOSSIER'],
}]
@ -555,7 +564,7 @@ def test_create_dossier(fake_conf, atreal_openads, collectivite_1, collectivite_
fake_resp.headers = {'Content-Type': 'application/json'}
fake_resp.encoding = 'utf-8'
fake_resp.reason = 'OK'
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
fake_resp._content = force_bytes(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
@ -568,42 +577,42 @@ def test_create_dossier(fake_conf, atreal_openads, collectivite_1, collectivite_
assert jresp['recepisse']['filename'] == fake_resp_json['files'][0]['filename']
fake_resp_json['numero_dossier'] = {'a': 'invalid type'}
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
fake_resp._content = force_bytes(json.dumps(fake_resp_json)) # pylint: disable=protected-access
with pytest.raises(APIError) as exception:
with mock.patch('passerelle.utils.Request.post') as requests_post:
requests_post.return_value = fake_resp
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
now=now_open)
assert unicode(exception.value) == (
assert force_text(exception.value) == (
u"Expecting '%s' value in JSON response to be a %s (not a %s)" % (
'numero_dossier', 'string', type({})))
del fake_resp_json['numero_dossier']
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
fake_resp._content = force_bytes(json.dumps(fake_resp_json)) # pylint: disable=protected-access
with pytest.raises(APIError) as exception:
with mock.patch('passerelle.utils.Request.post') as requests_post:
requests_post.return_value = fake_resp
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
now=now_open)
assert unicode(exception.value) == u"Expecting 'numero_dossier' key in JSON response"
assert force_text(exception.value) == u"Expecting 'numero_dossier' key in JSON response"
fake_resp_json['files'][0]['b64_content'] = 'invalid_;{[content}'
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
fake_resp._content = force_bytes(json.dumps(fake_resp_json)) # pylint: disable=protected-access
with pytest.raises(APIError) as exception:
with mock.patch('passerelle.utils.Request.post') as requests_post:
requests_post.return_value = fake_resp
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
now=now_open)
assert unicode(exception.value) == u'Failed to decode recepisse content from base 64'
assert force_text(exception.value) == u'Failed to decode recepisse content from base 64'
fake_resp._content = 'df[{gfd;g#vfd' # pylint: disable=protected-access
fake_resp._content = b'df[{gfd;g#vfd' # pylint: disable=protected-access
with pytest.raises(APIError) as exception:
with mock.patch('passerelle.utils.Request.post') as requests_post:
requests_post.return_value = fake_resp
atreal_openads.create_dossier(request_1, 'DIA', collectivite=collectivite_1.openADS_id,
now=now_open)
# pylint: disable=protected-access
assert unicode(exception.value) == u'No JSON content returned: %r' % fake_resp._content
assert force_text(exception.value) == u'No JSON content returned: %r' % fake_resp._content
# pylint: disable=no-member
job = Job.objects.filter(natural_id=fake_conf['FAKE_NUMERO_DOSSIER']).last()
@ -631,7 +640,7 @@ def test_create_dossier(fake_conf, atreal_openads, collectivite_1, collectivite_
fake_resp.headers = {'Content-Type': 'application/json'}
fake_resp.encoding = 'utf-8'
fake_resp.reason = 'OK'
fake_resp._content = json.dumps(fake_resp_json)
fake_resp._content = force_bytes(json.dumps(fake_resp_json))
with mock.patch('passerelle.utils.Request.post') as requests_post:
requests_post.return_value = fake_resp
atreal_openads.jobs()
@ -656,7 +665,7 @@ def test_get_dossier(fake_conf, atreal_openads):
with mock.patch('passerelle.utils.Request.get') as requests_get:
requests_get.return_value = fake_resp_bad
atreal_openads.get_dossier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'])
assert unicode(exception.value) == "HTTP error: 502"
assert force_text(exception.value) == "HTTP error: 502"
fake_resp_json = {
'etat': u"Non préemption en cours",
@ -670,7 +679,7 @@ def test_get_dossier(fake_conf, atreal_openads):
fake_resp.headers = {'Content-Type': 'application/json'}
fake_resp.encoding = 'utf-8'
fake_resp.reason = 'OK'
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
fake_resp._content = force_bytes(json.dumps(fake_resp_json)) # pylint: disable=protected-access
with mock.patch('passerelle.utils.Request.get') as requests_get:
requests_get.return_value = fake_resp
jresp = atreal_openads.get_dossier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'])
@ -680,13 +689,13 @@ def test_get_dossier(fake_conf, atreal_openads):
assert jresp['decision'] == fake_resp_json['decision']
assert jresp['date_limite_instruction'] == fake_resp_json['date_limite_instruction']
fake_resp._content = 'df[{gfd;g#vfd' # pylint: disable=protected-access
fake_resp._content = b'df[{gfd;g#vfd' # pylint: disable=protected-access
with pytest.raises(APIError) as exception:
with mock.patch('passerelle.utils.Request.get') as requests_get:
requests_get.return_value = fake_resp
atreal_openads.get_dossier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'])
# pylint: disable=protected-access
assert unicode(exception.value) == u'No JSON content returned: %r' % fake_resp._content
assert force_text(exception.value) == u'No JSON content returned: %r' % fake_resp._content
fake_resp_json = {
'errors': [{
@ -697,12 +706,12 @@ def test_get_dossier(fake_conf, atreal_openads):
}
fake_resp.status_code = 404
fake_resp.reason = 'Resource not found'
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
fake_resp._content = force_bytes(json.dumps(fake_resp_json)) # pylint: disable=protected-access
with pytest.raises(APIError) as exception:
with mock.patch('passerelle.utils.Request.get') as requests_get:
requests_get.return_value = fake_resp
atreal_openads.get_dossier(None, 'invalid_type', fake_conf['FAKE_NUMERO_DOSSIER'])
assert unicode(exception.value) == (u"HTTP error: 404, [path] (Invalid Type) "
assert force_text(exception.value) == (u"HTTP error: 404, [path] (Invalid Type) "
"\"invalid_type\" is not one of DIA, PC, DP, AT, PD")
@ -713,11 +722,11 @@ def test_get_fwd_files(fake_conf, atreal_openads):
with pytest.raises(APIError) as exception:
atreal_openads.get_fwd_files(None, fake_conf['FAKE_NUMERO_DOSSIER'],
fichier_id='not an integer')
assert unicode(exception.value) == u"fichier_id must be an integer"
assert force_text(exception.value) == u"fichier_id must be an integer"
with pytest.raises(Http404) as exception:
atreal_openads.get_fwd_files(None, fake_conf['FAKE_NUMERO_DOSSIER'], fichier_id=18)
assert unicode(exception.value) == (
assert force_text(exception.value) == (
u"No file matches 'numero_dossier=%s' and 'id=%s'." % (fake_conf['FAKE_NUMERO_DOSSIER'],
18))
@ -815,13 +824,13 @@ def test_get_courrier(fake_conf, atreal_openads):
with mock.patch('passerelle.utils.Request.get') as requests_get:
requests_get.return_value = fake_resp_bad
atreal_openads.get_courrier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'], lettre_type)
assert unicode(exception.value) == "HTTP error: 502"
assert force_text(exception.value) == "HTTP error: 502"
fake_resp_json = {
'files': [{
'filename': "instruction_4.pdf",
'content_type': "text/plain",
'b64_content': get_file_data(fake_conf['TEST_FILE_CERFA_DIA'])
'b64_content': force_text(get_file_data(fake_conf['TEST_FILE_CERFA_DIA'], b64=True))
}]
}
fake_resp = Response()
@ -829,7 +838,7 @@ def test_get_courrier(fake_conf, atreal_openads):
fake_resp.headers = {'Content-Type': 'application/json'}
fake_resp.encoding = 'utf-8'
fake_resp.reason = 'OK'
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
fake_resp._content = force_bytes(json.dumps(fake_resp_json)) # pylint: disable=protected-access
with mock.patch('passerelle.utils.Request.get') as requests_get:
requests_get.return_value = fake_resp
jresp = atreal_openads.get_courrier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'],
@ -839,20 +848,20 @@ def test_get_courrier(fake_conf, atreal_openads):
assert jresp['courrier']['b64_content'] == fake_resp_json['files'][0]['b64_content']
fake_resp_json['files'][0]['b64_content'] = 'invalid_;{[content}'
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
fake_resp._content = force_bytes(json.dumps(fake_resp_json)) # pylint: disable=protected-access
with pytest.raises(APIError) as exception:
with mock.patch('passerelle.utils.Request.get') as requests_get:
requests_get.return_value = fake_resp
atreal_openads.get_courrier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'], lettre_type)
assert unicode(exception.value) == u'Failed to decode courrier content from base 64'
assert force_text(exception.value) == u'Failed to decode courrier content from base 64'
fake_resp._content = 'df[{gfd;g#vfd' # pylint: disable=protected-access
fake_resp._content = b'df[{gfd;g#vfd' # pylint: disable=protected-access
with pytest.raises(APIError) as exception:
with mock.patch('passerelle.utils.Request.get') as requests_get:
requests_get.return_value = fake_resp
atreal_openads.get_courrier(None, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'], lettre_type)
# pylint: disable=protected-access
assert unicode(exception.value) == u'No JSON content returned: %r' % fake_resp._content
assert force_text(exception.value) == u'No JSON content returned: %r' % fake_resp._content
# pylint: disable=unused-argument,redefined-outer-name
@ -873,7 +882,7 @@ def test_get_response_error(atreal_openads):
fake_resp.headers = {'Content-Type': 'application/json'}
fake_resp.encoding = 'utf-8'
fake_resp.reason = 'Not Found'
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
fake_resp._content = force_bytes(json.dumps(fake_resp_json)) # pylint: disable=protected-access
error_msg = atreal_openads.get_response_error(fake_resp)
expected_msg = u'[%s] (%s) %s' % (
@ -883,10 +892,10 @@ def test_get_response_error(atreal_openads):
)
assert error_msg == u"HTTP error: %s, %s" % (fake_resp.status_code, ','.join([expected_msg]))
fake_resp._content = 'invalid_;{[content}' # pylint: disable=protected-access
fake_resp._content = b'invalid_;{[content}' # pylint: disable=protected-access
error_msg = atreal_openads.get_response_error(fake_resp)
# pylint: disable=protected-access
assert error_msg == u"HTTP error: %s, %s" % (fake_resp.status_code, fake_resp._content)
assert error_msg == u"HTTP error: %s, %s" % (fake_resp.status_code, fake_resp.text)
# pylint: disable=unused-argument,redefined-outer-name
@ -899,18 +908,18 @@ def test_upload_user_files(fake_conf, atreal_openads, request_1):
with pytest.raises(ForwardFile.DoesNotExist) as exception: # pylint: disable=no-member
atreal_openads.upload_user_files(request_1, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'],
file_ids=[999])
assert unicode(exception.value) == u"The following ForwardFile IDs were not found: %s." % [999]
assert force_text(exception.value) == u"The following ForwardFile IDs were not found: %s." % [999]
with pytest.raises(ValueError) as exception:
atreal_openads.upload_user_files(request_1, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'],
file_ids='invalid string')
assert unicode(exception.value) == (
assert force_text(exception.value) == (
u"invalid literal for int() with base 10: '%s'" % 'invalid string')
with pytest.raises(TypeError) as exception:
atreal_openads.upload_user_files(request_1, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'],
file_ids={'invalid': 'type'})
assert unicode(exception.value) == (
assert force_text(exception.value) == (
u"Invalid 'file_ids' argument type '%s' (must be string or list)" % (
type({'invalid': 'type'})))
@ -933,7 +942,7 @@ def test_upload_user_files(fake_conf, atreal_openads, request_1):
requests_post.return_value = fake_resp_bad
atreal_openads.upload_user_files(request_1, 'DIA', fake_conf['FAKE_NUMERO_DOSSIER'],
file_ids=str(file_id))
assert unicode(exception.value) == u'HTTP error: 502'
assert force_text(exception.value) == u'HTTP error: 502'
ffup = ForwardFile.objects.get(id=file_id) # pylint: disable=no-member
assert isinstance(ffup, ForwardFile)
@ -952,7 +961,7 @@ def test_upload_user_files(fake_conf, atreal_openads, request_1):
fake_resp.encoding = 'utf-8'
fake_resp.reason = 'OK'
fake_resp._content = 'invalid_;{[content}' # pylint: disable=protected-access
fake_resp._content = b'invalid_;{[content}' # pylint: disable=protected-access
with pytest.raises(APIError) as exception:
with mock.patch('passerelle.utils.Request.post') as requests_post:
requests_post.return_value = fake_resp
@ -960,7 +969,7 @@ def test_upload_user_files(fake_conf, atreal_openads, 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
assert force_text(exception.value) == u'No JSON content returned: %r' % fake_resp.content
ffup = ForwardFile.objects.get(id=file_id) # pylint: disable=no-member
assert isinstance(ffup, ForwardFile)
@ -979,7 +988,7 @@ def test_upload_user_files(fake_conf, atreal_openads, request_1):
ffup.save()
fake_resp_json = "You want add some files on %s " % fake_conf['FAKE_NUMERO_DOSSIER']
fake_resp._content = json.dumps(fake_resp_json) # pylint: disable=protected-access
fake_resp._content = force_bytes(json.dumps(fake_resp_json)) # pylint: disable=protected-access
with mock.patch('passerelle.utils.Request.post') as requests_post:
requests_post.return_value = fake_resp
jresp = atreal_openads.upload_user_files(request_1, 'DIA',

View File

@ -23,6 +23,7 @@
import base64
import re
from django.utils.encoding import force_text
import pytest
from atreal_openads.utils import (
@ -114,15 +115,15 @@ def test_get_file_data(fake_conf):
"""Test for function 'get_file_data()'."""
assert get_file_data(fake_conf['TEST_FILE_CERFA_DIA']) == base64.b64encode(
open(fake_conf['TEST_FILE_CERFA_DIA']).read())
open(fake_conf['TEST_FILE_CERFA_DIA'], 'rb').read())
assert get_file_data(fake_conf['TEST_FILE_CERFA_DIA'], b64=False) == open(
fake_conf['TEST_FILE_CERFA_DIA']).read()
fake_conf['TEST_FILE_CERFA_DIA'], 'rb').read()
def test_get_file_digest(fake_conf):
"""Test for function 'get_file_digest()'."""
with open(fake_conf['TEST_FILE_CERFA_DIA']) as file_pt:
with open(fake_conf['TEST_FILE_CERFA_DIA'], 'rb') as file_pt:
assert get_file_digest(file_pt) == ('cc90a620982760fdee16a5b4fe1b5ac3'
'b4fe868fd02d2f70b27f1e46d283ea51')
@ -190,26 +191,26 @@ def test_dict_dumper():
dumped = DictDumper(dic, use_json_dumps=False)
assert repr(dumped) == (u'DictDumper(dic=%r,max_str_len=%r,use_json_dumps=%r)' % (
dic, dumped.max_str_len, dumped.use_json_dumps)).encode('utf-8')
dic, dumped.max_str_len, dumped.use_json_dumps))
assert str(dumped) == '{}'
assert unicode(dumped) == u'{}'
assert force_text(dumped) == u'{}'
assert dic == dumped.dic
assert unicode(dic) == unicode(dumped)
assert force_text(dic) == force_text(dumped)
dumped = DictDumper(dic, 0, use_json_dumps=False)
assert dic == dumped.dic
assert unicode(dic) == unicode(dumped)
assert force_text(dic) == force_text(dumped)
dic = {'a': '123456789'}
dumped = DictDumper(dic, 10, use_json_dumps=False)
assert dic == dumped.dic
assert unicode(dic) == unicode(dumped)
assert force_text(dic) == force_text(dumped)
dumped = DictDumper(dic, 5, use_json_dumps=False)
assert dic == dumped.dic
assert unicode(dumped) == unicode({'a': u'12345…'})
assert force_text(dumped) == force_text({'a': u'12345…'})
dumped = DictDumper(dic, 5, use_json_dumps=True)
assert dic == dumped.dic
assert unicode(dumped) == u'{"a": "12345\\u2026"}'
assert force_text(dumped) == u'{"a": "12345\\u2026"}'
# pylint: disable=unused-argument,redefined-outer-name
@ -246,10 +247,10 @@ def test_base_model(fake_conf, atreal_openads, collectivite_1, collectivite_1_gu
with pytest.raises(Exception) as exception:
atreal_openads.get_list_url()
assert unicode(exception.value) == u"AtrealOpenads:get_list_url() method should not be called"
assert force_text(exception.value) == u"AtrealOpenads:get_list_url() method should not be called"
# TODO add more collectivite test cases
with pytest.raises(Exception) as exception:
collectivite_1_guichet.get_list_url()
assert unicode(exception.value) == u"Guichet:get_list_url() method should not be called"
assert force_text(exception.value) == u"Guichet:get_list_url() method should not be called"