From 351a1a6242f9ed8d474d74c111f723a40d14585e Mon Sep 17 00:00:00 2001 From: Emmanuel Cazenave Date: Tue, 4 Feb 2020 17:28:28 +0100 Subject: [PATCH] adapt tests to python3 (#39533) --- tests/conftest.py | 11 +-- tests/test_atreal_openads.py | 159 ++++++++++++++++++----------------- tests/test_utils.py | 25 +++--- 3 files changed, 103 insertions(+), 92 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index a6d3e18..4b030d3 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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' ) diff --git a/tests/test_atreal_openads.py b/tests/test_atreal_openads.py index ebee38f..f350eed 100644 --- a/tests/test_atreal_openads.py +++ b/tests/test_atreal_openads.py @@ -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', diff --git a/tests/test_utils.py b/tests/test_utils.py index dbf2064..de03a74 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -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"