Commit 9cacf50a authored by Jérome Perrin's avatar Jérome Perrin

bt5/test (not reviewed)

parent 2c0c2049
......@@ -595,7 +595,7 @@ class TestERP5Catalog(ERP5TypeTestCase, LogInterceptor):
uid_dict = {}
for _ in range(UID_BUFFER_SIZE * 3):
uid = portal_catalog.newUid()
self.assertIsInstance(uid, long)
self.assertIsInstance(uid, long) # pylint:disable=possibly-used-before-assignment
self.assertNotIn(uid, uid_dict)
uid_dict[uid] = None
......
......@@ -297,3 +297,26 @@ class TestIdToolUpgrade(ERP5TypeTestCase):
id_generator.clearGenerator() # clear stored data
self._checkDataStructureMigration(id_generator)
def test_portal_ids_table_id_group_column_binary(self):
"""portal_ids.id_group is now created as VARCHAR,
but it use to be binary. There is no data migration, the
SQL method has been adjusted to cast during select.
This checks that id generator works well when the column
is VARBINARY, like it's the case for old instances.
"""
self.assertEqual(
self.sql_generator.generateNewId(id_group=self.id()),
0)
exported = self.sql_generator.exportGeneratorIdDict()
self.tic()
self.portal.portal_ids.IdTool_zCommit()
self.portal.erp5_sql_connection.manage_test(
'ALTER TABLE portal_ids MODIFY COLUMN id_group VARBINARY(255)'
)
self.tic()
self.sql_generator.importGeneratorIdDict(exported, clear=True)
self.tic()
self.assertEqual(
self.sql_generator.generateNewId(id_group=self.id()),
1)
......@@ -12,16 +12,17 @@ from ZPublisher.HTTPResponse import HTTPResponse
import base64
import DateTime
import StringIO
from six.moves import cStringIO as StringIO
import json
import re
from six.moves.urllib.parse import quote, quote_plus
import six
import mock
from zope.globalrequest import setRequest # pylint: disable=no-name-in-module, import-error
from Acquisition import aq_base
from Products.ERP5Form.Selection import Selection, DomainSelection
from Products.ERP5Type.Utils import str2unicode, unicode2str
from Products.ERP5Type.Utils import bytes2str, str2unicode, unicode2str
def changeSkin(skin_name):
......@@ -118,7 +119,7 @@ def do_fake_request(request_method, headers=None, data=()):
env['GATEWAY_INTERFACE']='CGI/1.1 '
env['SCRIPT_NAME']='Main'
env.update(headers)
body_stream = StringIO.StringIO()
body_stream = StringIO()
# for some mysterious reason QUERY_STRING does not get parsed into data fields
if data and request_method.upper() == 'GET':
......@@ -1324,7 +1325,7 @@ class TestERP5Document_getHateoas_mode_traverse(ERP5HALJSONStyleSkinsMixin):
def test_getHateoasDocument_property_corrupted_encoding(self):
document = self._makeDocument()
# this sequence of bytes does not encode to UTF-8
document.setTitle('\xe9\xcf\xf3\xaf')
document.setTitle(b'\xe9\xcf\xf3\xaf')
fake_request = do_fake_request("GET")
result = self.portal.web_site_module.hateoas.ERP5Document_getHateoas(REQUEST=fake_request, mode="traverse", relative_url=document.getRelativeUrl(), view="view")
self.assertEqual(fake_request.RESPONSE.status, 200)
......@@ -1332,9 +1333,10 @@ class TestERP5Document_getHateoas_mode_traverse(ERP5HALJSONStyleSkinsMixin):
"application/hal+json"
)
result_dict = json.loads(result)
self.assertEqual(result_dict['_embedded']['_view']['my_title']['default'], u'\ufffd\ufffd\ufffd')
self.assertEqual(result_dict['title'], u'\ufffd\ufffd\ufffd')
self.assertEqual(result_dict['_embedded']['_view']['_links']['traversed_document']['title'], u'\ufffd\ufffd\ufffd')
expected = u'\ufffd\ufffd\ufffd' if six.PY2 else u'\udce9\udccf\udcf3\udcaf'
self.assertEqual(result_dict['_embedded']['_view']['my_title']['default'], expected)
self.assertEqual(result_dict['title'], expected)
self.assertEqual(result_dict['_embedded']['_view']['_links']['traversed_document']['title'], expected)
class TestERP5Document_getHateoas_mode_search(ERP5HALJSONStyleSkinsMixin):
......@@ -1682,10 +1684,12 @@ class TestERP5Document_getHateoas_mode_search(ERP5HALJSONStyleSkinsMixin):
def test_getHateoas_default_param_json_param(self):
fake_request = do_fake_request("GET")
unknown_columns_re = re.escape("Unknown columns ['ê']")
if six.PY2:
unknown_columns_re = "Unknown columns.*\\\\xc3\\\\xaa.*"
self.assertRaisesRegex(
TypeError,
# "Unknown columns.*'\\xc3\\xaa'.",
"Unknown columns.*\\\\xc3\\\\xaa.*",
unknown_columns_re,
self.portal.web_site_module.hateoas.ERP5Document_getHateoas,
REQUEST=fake_request,
mode="search",
......@@ -1725,7 +1729,7 @@ return context.getPortalObject().foo_module.contentValues()
form_relative_url='portal_skins/erp5_ui_test/FooModule_viewFooList/listbox'
)
result_dict = json.loads(result)
#editalble creation date is defined at proxy form
# editable creation date is defined at proxy form
# Test the listbox_uid parameter
self.assertEqual(result_dict['_embedded']['contents'][0]['listbox_uid:list']['key'], 'listbox_uid:list')
self.assertEqual(result_dict['_embedded']['contents'][0]['id']['field_gadget_param']['type'], 'StringField')
......@@ -2422,7 +2426,7 @@ return context.getPortalObject().portal_catalog(portal_type='Foo', sort_on=[('id
@changeSkin('Hal')
def test_getHateoas_property_corrupted_encoding(self, document):
# this sequence of bytes does not encode to UTF-8
document.setTitle('\xe9\xcf\xf3\xaf')
document.setTitle(b'\xe9\xcf\xf3\xaf')
# self.tic()
fake_request = do_fake_request("GET")
......@@ -2439,7 +2443,9 @@ return context.getPortalObject().portal_catalog(portal_type='Foo', sort_on=[('id
result_dict = json.loads(result)
self.assertEqual(len(result_dict['_embedded']['contents']), 1)
self.assertEqual(result_dict['_embedded']['contents'][0]["title"], u'\ufffd\ufffd\ufffd')
self.assertEqual(
result_dict['_embedded']['contents'][0]["title"],
u'\ufffd\ufffd\ufffd' if six.PY2 else u'\udce9\udccf\udcf3\udcaf')
class TestERP5Person_getHateoas_mode_search(ERP5HALJSONStyleSkinsMixin):
......@@ -3183,11 +3189,11 @@ class TestERP5ODS(ERP5HALJSONStyleSkinsMixin):
))
fake_portal = replace_request(fake_request, self.portal)
result = fake_portal.web_site_module.hateoas.foo_module.Base_callDialogMethod(
result = bytes2str(fake_portal.web_site_module.hateoas.foo_module.Base_callDialogMethod(
dialog_method='Base_viewAsODS',
dialog_id='Base_viewAsODSDialog',
form_id='FooModule_viewFooList',
)
))
self.assertEqual(fake_request.get('portal_skin'), 'ODS')
self.assertEqual(fake_request.RESPONSE.status, 200)
if IS_ZOPE2:
......@@ -3247,21 +3253,21 @@ class TestERP5ODS(ERP5HALJSONStyleSkinsMixin):
))
fake_portal = replace_request(fake_request, self.portal)
result = fake_portal.web_site_module.hateoas.foo_module.Base_callDialogMethod(
result = bytes2str(fake_portal.web_site_module.hateoas.foo_module.Base_callDialogMethod(
dialog_method='Base_viewAsODS',
dialog_id='Base_viewAsODSDialog',
form_id='FooModule_viewFooList',
)
))
self.assertEqual(fake_request.get('portal_skin'), 'ODS')
self.assertEqual(fake_request.RESPONSE.status, 200)
if IS_ZOPE2:
self.assertEqual(fake_request.RESPONSE.getHeader('Content-Type'), 'text/csv')
else:
self.assertEqual(fake_request.RESPONSE.getHeader('Content-Type'), 'text/csv; charset=utf-8')
self.assertTrue('foook1' in result, result)
self.assertTrue('foook2' in result, result)
self.assertTrue('foonotok' not in result, result)
self.assertIn('foook1', result)
self.assertIn('foook2', result)
self.assertNotIn('foonotok', result)
# Check one of the field name
self.assertTrue('Read-Only Quantity' in result, result)
self.assertIn('Read-Only Quantity', result)
# Ensure it is not the list mode rendering
self.assertTrue(len(result.split('\n')) > 50, result)
......@@ -2057,7 +2057,7 @@ class TestImmobilisationMixin(ERP5TypeTestCase):
e_movement = e_simulation_movement_list[e_cursor]
wrong_movement = 0
key_cursor = 0
key_list = e_movement.keys()
key_list = list(e_movement.keys())
key_list.remove('id')
while key_cursor < len(key_list) and not wrong_movement:
key = key_list[key_cursor]
......@@ -2895,7 +2895,7 @@ class TestImmobilisationMixin(ERP5TypeTestCase):
e_transaction = e_transaction_list[e_cursor]
wrong_transaction = 0
key_cursor = 0
key_list = e_transaction.keys()
key_list = list(e_transaction.keys())
if 'line_list' in key_list:
key_list.remove('line_list')
if 'id' in key_list:
......@@ -2947,7 +2947,7 @@ class TestImmobilisationMixin(ERP5TypeTestCase):
e_line = e_line_list[e_line_cursor]
wrong_line = 0
key_cursor = 0
key_list = e_line.keys()
key_list = list(e_line.keys())
while key_cursor < len(key_list) and not wrong_line:
key = key_list[key_cursor]
e_value = e_line[key]
......
......@@ -33,6 +33,7 @@ from Products.ERP5Type.tests.utils import createZODBPythonScript, removeZODBPyth
from Products.CMFCore.utils import getToolByName
import random
import string
from six.moves import range
# test files' home
FILENAME_REGULAR_EXPRESSION = "(?P<reference>[A-Z&é@{]{3,7})-(?P<language>[a-z]{2})-(?P<version>[0-9]{3})"
......@@ -93,7 +94,7 @@ class TestIngestion(ERP5TypeLiveTestCase):
portal = self.portal
contribution_tool = getToolByName(portal, 'portal_contributions')
# seed parameter is here to ensure entropy for document id generation
seed = ''.join([random.choice(string.ascii_letters) for _ in xrange(20)])
seed = ''.join([random.choice(string.ascii_letters) for _ in range(20)])
url = portal.absolute_url()
url += '/%s?seed=%s' % (script_id, seed)
if filename:
......
......@@ -27,12 +27,56 @@
import email
import mock
import time
import six
# pylint:disable=no-name-in-module
if six.PY3:
from email import message_from_bytes
else:
from email import message_from_string as message_from_bytes
# pylint:enable=no-name-in-module
from email.generator import Generator
from Products.ERP5Type.tests.ERP5TypeLiveTestCase import ERP5TypeTestCase
from Products.ERP5Type.tests.Sequence import SequenceList
from Products.ERP5Type.Utils import bytes2str, str2bytes
from Products.ZSQLCatalog.SQLCatalog import SimpleQuery
from DateTime import DateTime
from six import StringIO
import re
def normalize_email_bytes(email_bytes):
# type: (bytes) -> str
"""
Normalizes the representation of email text, so that it can be compared.
The fields of the message are written in a predefined order, with
the `unixfrom` field removed and no line wrapping.
The code is intended to be compatible with both Python 2 and Python 3.
Args:
email_bytes: Content of the email, including headers.
Returns:
Normalized string representation of the e-mail contents.
"""
# Unfolding removes newlines followed by whitespace, as per RFC5322.
# This SHOULD be done by Python itself, but seemingly no-one cared
# enough.
email_bytes_unfolded = re.sub(
br"\r?\n(?P<space>\s)",
br"\g<space>",
email_bytes,
)
msg = message_from_bytes(email_bytes_unfolded)
fp = StringIO()
g = Generator(fp, mangle_from_=False, maxheaderlen=0)
g.flatten(msg)
return fp.getvalue()
class TestInterfacePost(ERP5TypeTestCase):
"""
......@@ -231,7 +275,7 @@ class TestInterfacePost(ERP5TypeTestCase):
for internet_message_post in internet_message_post_list:
self.assertEqual(internet_message_post.getSimulationState(), 'exported')
mail_object = email.message_from_string(internet_message_post.getData())
mail_object = email.message_from_string(bytes2str(internet_message_post.getData()))
self.assertEqual(
internet_message_post.getReference(), mail_object['message-id'].strip('<>')
)
......@@ -247,7 +291,10 @@ class TestInterfacePost(ERP5TypeTestCase):
last_message, = self.portal.MailHost._message_list
self.assertNotEqual((), last_message)
_, _, message_text = last_message
self.assertIn(message_text, sequence['internet_message_post'].getData())
self.assertEqual(
normalize_email_bytes(message_text),
normalize_email_bytes(sequence['internet_message_post'].getData()),
)
def _getMailHostMessageForRecipient(self, recipient_email_address):
message_list = self.portal.MailHost._message_list
......@@ -263,12 +310,15 @@ class TestInterfacePost(ERP5TypeTestCase):
message_list = self.portal.MailHost._message_list
self.assertEqual(len(message_list), len(self.recipient_list))
for post in sequence['internet_message_post_list']:
post_recipient = email.message_from_string(post.getData())['to']
post_recipient = email.message_from_string(bytes2str(post.getData()))['to']
message_list = self._getMailHostMessageForRecipient(post_recipient)
self.assertEqual(len(message_list), 1)
message = message_list[0]
_, _, message_text = message
self.assertIn(message_text, post.getData())
self.assertEqual(
normalize_email_bytes(message_text),
normalize_email_bytes(post.getData()),
)
def stepCheckMailMessagePreviewDisplaysLatestInternetMessagePostData(self, sequence=None, sequence_list=None):
mail_message = sequence['mail_message']
......@@ -282,7 +332,7 @@ class TestInterfacePost(ERP5TypeTestCase):
post = sequence['internet_message_post']
# Create a response mail object
mail_object = email.message_from_string(post.getData())
mail_object = email.message_from_string(bytes2str(post.getData()))
sender = mail_object['from']
recipient = mail_object['to']
......@@ -297,7 +347,7 @@ class TestInterfacePost(ERP5TypeTestCase):
# Ingest it
response_post = self.portal.internet_message_post_module.newContent(
portal_type='Internet Message Post',
data=mail_object.as_string(),
data=str2bytes(mail_object.as_string()),
)
response_post.prepareImport()
sequence['internet_message_post_response'] = response_post
......
......@@ -40,7 +40,7 @@ class Test(ERP5TypeTestCase):
self.tic()
active_process = self.portal.portal_activities.unrestrictedTraverse(path)
result = active_process.getResultList()
self.assertAlmostEqual(0.98444444444444446, result[0].result)
self.assertAlmostEqual(0.9, result[0].result, 0)
def test_UnderRootOfSquaresFunction(self):
path = self.portal.Base_driverScriptSquareRoot()
......
......@@ -27,6 +27,7 @@
import json
from DateTime import DateTime
import six
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from Products.ERP5Type.tests.utils import createZODBPythonScript
......@@ -149,7 +150,9 @@ return json.dumps({
self.fixJSONForm(method, schema, after_method)
self.tic()
self.assertRaises(ValueError, getattr(self.portal, method), json_data, list_error=True)
error = {"my-schema.json": [[u"Validation Error", u"'2' is not of type u'integer'"], [u"Validation Error", u"2 is not of type u'string'"]]}
error = {"my-schema.json": [["Validation Error", "'2' is not of type 'integer'"], ["Validation Error", "2 is not of type 'string'"]]}
if six.PY2:
error = {"my-schema.json": [[u"Validation Error", u"'2' is not of type u'integer'"], [u"Validation Error", u"2 is not of type u'string'"]]}
try:
getattr(self.portal, method)(json_data, list_error=True)
raise ValueError("No error raised during processing")
......@@ -205,9 +208,15 @@ return json.dumps({
self.assertRaises(ValueError, getattr(self.portal, method), json_data, list_error=True)
error = {
"my-schema.json": [[
"Validation Error", u"'2018-11-13T20:20:67' is not a u'date-time'"
"Validation Error", u"'2018-11-13T20:20:67' is not a 'date-time'"
]]
}
if six.PY2:
error = {
"my-schema.json": [[
"Validation Error", u"'2018-11-13T20:20:67' is not a u'date-time'"
]]
}
try:
getattr(self.portal, method)(json_data, list_error=True)
raise ValueError("No error raised during processing")
......
......@@ -121,7 +121,7 @@ def submitRequestWithFailure(**kw):
class ServiceWithException:
def submit(self, **kw):
raise Exception('exception')
raise RuntimeError('exception')
class ClientWithException:
def __init__(self):
......@@ -416,11 +416,10 @@ class testMailevaSOAPConnector(ERP5TypeTestCase):
def test_maileva_request_validation(self):
xml = self.maileva_connector.generateRequestXML(self.recipient, self.sender, self.document, 'test_track_id', 'maileva_connection_for_test')
# lxml doesn't support https in schemaLocation, download locally
src = open(os.path.join(os.path.dirname(Products.ERP5.tests.__file__), 'test_data', "MailevaPJSSchema.xsd"))
xsd = etree.parse(src)
with open(os.path.join(os.path.dirname(Products.ERP5.tests.__file__), 'test_data', "MailevaPJSSchema.xsd")) as f:
xsd = etree.parse(f)
schema_validator = etree.XMLSchema(xsd)
schema_validator.assertValid(etree.fromstring(xml.encode("UTF-8")))
schema_validator.assertValid(etree.fromstring(xml))
def test_send_state_workflow(self):
pdf = self.portal.document_module.newContent(portal_type='PDF')
......
......@@ -28,7 +28,10 @@
import unittest
from DateTime import DateTime
from Products.ERP5Type.Utils import OrderableKey
from erp5.component.test.testBPMCore import TestBPMMixin
from six.moves import range
import six
class TestMRPMixin(TestBPMMixin):
......@@ -114,11 +117,11 @@ class TestMRPMixin(TestBPMMixin):
self.createCategoriesInCategory(category_tool.quantity_unit.weight, ['kg'])
self.createCategoriesInCategory(category_tool.trade_phase, ['mrp',])
self.createCategoriesInCategory(category_tool.trade_phase.mrp,
('manufacturing_step_' + str(i) for i in xrange(2)))
('manufacturing_step_' + str(i) for i in range(2)))
self.createCategoriesInCategory(category_tool.trade_phase.mrp,
('sourcing_' + str(i) for i in xrange(1)))
('sourcing_' + str(i) for i in range(1)))
self.createCategoriesInCategory(category_tool.trade_state,
('step_' + str(i) for i in xrange(6)))
('step_' + str(i) for i in range(6)))
def createDefaultOrder(self, business_process, transformation=None):
if transformation is None:
......@@ -278,7 +281,7 @@ class TestMRPMixin(TestBPMMixin):
group_by_variation=1):
self.assertEqual(expected_dict.pop((r.node_uid, r.variation_text), 0),
r.inventory)
self.assertFalse(any(expected_dict.itervalues()), expected_dict)
self.assertFalse(any(six.itervalues(expected_dict)), expected_dict)
class TestMRPImplementation(TestMRPMixin):
"""the test for implementation"""
......@@ -339,18 +342,18 @@ class TestMRPImplementation(TestMRPMixin):
reference = None
movement_list.append((sm.getTradePhase(), sm.getQuantity(),
reference, sm.getIndustrialPhaseList()))
movement_list.sort()
self.assertEqual(movement_list, sorted((
('mrp/manufacturing_step_0', -10.0, None, []),
movement_list.sort(key=lambda x: [OrderableKey(e) for e in x])
self.assertEqual(movement_list, [
('mrp/manufacturing_step_0', -30.0, None, []),
('mrp/manufacturing_step_0', -10.0, None, []),
('mrp/manufacturing_step_0', 10.0,
'pr/mrp/manufacturing_step_0', ['trade_phase/mrp/manufacturing_step_0']),
('mrp/manufacturing_step_1', -40.0, None, []),
('mrp/manufacturing_step_1', -10.0, None, []),
('mrp/manufacturing_step_1', -10.0,
'cr/mrp/manufacturing_step_1', ['trade_phase/mrp/manufacturing_step_0']),
('mrp/manufacturing_step_1', -10.0, None, []),
('mrp/manufacturing_step_1', -40.0, None, []),
('mrp/manufacturing_step_1', 10.0, 'pr', []),
)))
])
order.confirm()
# Build Manufacturing Order
......
......@@ -884,6 +884,7 @@ class TestProductionOrderMixin(TestOrderMixin):
'operation/operation1')
component_resource = sequence.get('component1')
# for consumed_movement in (applied_rule.cr_1, applied_rule.cr_2):
operation_movement = component_movement = None
for consumed_movement in simulation_movement_list:
if consumed_movement.getResourceValue() == operation_resource:
operation_movement = consumed_movement
......@@ -1096,6 +1097,7 @@ class TestProductionOrderMixin(TestOrderMixin):
'operation/operation1')
component_resource = sequence.get('component1')
# for consumed_movement in (applied_rule.cr_1, applied_rule.cr_2):
operation_movement = component_movement = None
for consumed_movement in simulation_movement_list:
if consumed_movement.getResourceValue() == operation_resource:
operation_movement = consumed_movement
......@@ -1216,6 +1218,7 @@ class TestProductionOrderMixin(TestOrderMixin):
operation_resource = resource.portal_categories.resolveCategory(
'operation/operation2')
component_resource = sequence.get('component2')
operation_movement = component_movement = None
for consumed_movement in simulation_movement_list:
if consumed_movement.getResourceValue() == operation_resource:
operation_movement = consumed_movement
......
......@@ -24,16 +24,16 @@
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
##############################################################################
import io
import json
from StringIO import StringIO
import urlparse
import httplib
import six.moves.urllib.parse
import six.moves.http_client
import feedparser
from DateTime import DateTime
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
class FileUpload(StringIO):
class FileUpload(io.BytesIO):
filename = 'attached_file.txt'
......@@ -183,7 +183,7 @@ class TestSupportRequestCreateNewSupportRequest(SupportRequestTestCase):
post, = web_message.getAggregateValueList(
portal_type='HTML Post'
)
self.assertEqual('<b>Help !!!</b>', str(post.getData()))
self.assertEqual(b'<b>Help !!!</b>', bytes(post.getData()))
# post have been archived once ingested
self.assertEqual('archived', post.getValidationState())
......@@ -201,7 +201,7 @@ class TestSupportRequestCreateNewSupportRequest(SupportRequestTestCase):
def test_submit_support_request_with_attachment(self):
self.getWebSite().SupportRequestModule_createSupportRequest(
description='<b>Look at this file !</b>',
file=FileUpload("the text content"),
file=FileUpload(b"the text content"),
resource=self.portal.service_module.erp5_officejs_support_request_ui_test_service_001.getRelativeUrl(),
title=self.id(),
project='erp5_officejs_support_request_ui_test_project_001',
......@@ -234,16 +234,16 @@ class TestSupportRequestCreateNewSupportRequest(SupportRequestTestCase):
post, = web_message.getAggregateValueList(
portal_type='HTML Post'
)
self.assertEqual('<b>Look at this file !</b>', str(post.getData()))
self.assertEqual(b'<b>Look at this file !</b>', bytes(post.getData()))
file_post, = post.getSuccessorValueList()
self.assertEqual('the text content', str(file_post.getData()))
self.assertEqual(b'the text content', bytes(file_post.getData()))
# a text was ingested from the file post
file_document, = web_message.getAggregateValueList(
portal_type='Text'
)
self.assertEqual('attached_file.txt', file_document.getFilename())
self.assertEqual('the text content', str(file_document.getData()))
self.assertEqual(b'the text content', bytes(file_document.getData()))
self.assertEqual(
[dict(
......@@ -262,7 +262,7 @@ class TestSupportRequestCommentOnExistingSupportRequest(SupportRequestTestCase):
self.portal.PostModule_createHTMLPostForSupportRequest(
follow_up=support_request.getRelativeUrl(),
predecessor=None,
data="<p>Hello !</p>",
data=b"<p>Hello !</p>",
file=None,
source_reference="xxx-message-id",
web_site_relative_url=self.getWebSite().getRelativeUrl(),
......@@ -300,8 +300,8 @@ class TestSupportRequestCommentOnExistingSupportRequest(SupportRequestTestCase):
self.portal.PostModule_createHTMLPostForSupportRequest(
follow_up=support_request.getRelativeUrl(),
predecessor=None,
data="<p>Please look at the <b>attached file</b></p>",
file=FileUpload("the text content"),
data=b"<p>Please look at the <b>attached file</b></p>",
file=FileUpload(b"the text content"),
source_reference="xxx-message-id",
web_site_relative_url=self.getWebSite().getRelativeUrl(),
)
......@@ -327,7 +327,7 @@ class TestSupportRequestCommentOnExistingSupportRequest(SupportRequestTestCase):
portal_type='Text'
)
self.assertEqual('attached_file.txt', file_document.getFilename())
self.assertEqual('the text content', str(file_document.getData()))
self.assertEqual(b'the text content', bytes(file_document.getData()))
self.assertEqual('shared', file_document.getValidationState())
# this document is also attached to the context of the support request
# and is visible from the document tab.
......@@ -352,7 +352,7 @@ class TestSupportRequestCommentOnExistingSupportRequest(SupportRequestTestCase):
follow_up=support_request.getRelativeUrl(),
predecessor=None,
data="<p>look <script>alert('haha')</script></p>",
file=FileUpload("the text content"),
file=FileUpload(b"the text content"),
source_reference="xxx-message-id",
web_site_relative_url=self.getWebSite().getRelativeUrl(),
)
......@@ -367,8 +367,8 @@ class TestSupportRequestCommentOnExistingSupportRequest(SupportRequestTestCase):
# but the post follow the "store what user entered as-is" rule.
# (so looking at posts can be dangerous)
self.assertEqual(
"<p>look <script>alert('haha')</script></p>",
str(post.getData()))
b"<p>look <script>alert('haha')</script></p>",
bytes(post.getData()))
def test_support_request_comment_include_other_event_type(self):
support_request = self.portal.support_request_module.erp5_officejs_support_request_ui_test_support_reuqest_001
......@@ -461,7 +461,7 @@ class TestSupportRequestRSSSOneEvent(SupportRequestRSSTestCase, DefaultTestRSSMi
"""Tests for simple cases of RSS with only one event.
"""
def _checkRSS(self, response):
self.assertEqual(httplib.OK, response.getStatus())
self.assertEqual(six.moves.http_client.OK, response.getStatus())
rss = feedparser.parse(response.getBody())
self.assertEqual(rss['feed']['title'], "Support Requests")
item, = rss.entries
......@@ -500,7 +500,7 @@ class TestSupportRequestRSSSOneEvent(SupportRequestRSSTestCase, DefaultTestRSSMi
# get rss link url
self.getWebSite().support_request_module.SupportRequestModule_generateRSSLinkUrl()
restricted_access_url = self.portal.REQUEST.form["your_rss_url"]
parsed_url = urlparse.urlparse(restricted_access_url)
parsed_url = six.moves.urllib.parse.urlparse(restricted_access_url)
restricted_access_url = restricted_access_url.replace(
'%s://%s' % (parsed_url.scheme, parsed_url.netloc), '', 1)
# and check it
......@@ -538,7 +538,7 @@ class TestSupportRequestRSSSMultipleEvents(SupportRequestRSSTestCase, DefaultTes
self.tic()
def _checkRSS(self, response):
self.assertEqual(httplib.OK, response.getStatus())
self.assertEqual(six.moves.http_client.OK, response.getStatus())
rss = feedparser.parse(response.getBody())
self.assertEqual(rss['feed']['title'], "Support Requests")
self.assertEqual(len(rss.entries), 3)
......@@ -562,7 +562,7 @@ class TestSupportRequestRSSSNonVisibleSupportRequest(SupportRequestRSSTestCase,
self.tic()
def _checkRSS(self, response):
self.assertEqual(httplib.OK, response.getStatus())
self.assertEqual(six.moves.http_client.OK, response.getStatus())
rss = feedparser.parse(response.getBody())
item, = rss.entries
self.assertEqual(item['author'], self.user.getTitle())
......@@ -586,7 +586,7 @@ class TestSupportRequestRSSSNonVisibleSender(SupportRequestRSSTestCase, DefaultT
self.tic()
def _checkRSS(self, response):
self.assertEqual(httplib.OK, response.getStatus())
self.assertEqual(six.moves.http_client.OK, response.getStatus())
rss = feedparser.parse(response.getBody())
item, = rss.entries
# no author for this event, because sender could not be access
......@@ -605,7 +605,7 @@ class TestSupportRequestRSSSNonVisibleAttachment(SupportRequestRSSTestCase, Defa
self.tic()
def _checkRSS(self, response):
self.assertEqual(httplib.OK, response.getStatus())
self.assertEqual(six.moves.http_client.OK, response.getStatus())
rss = feedparser.parse(response.getBody())
item, = rss.entries
# no enclosure
......@@ -626,7 +626,7 @@ class TestIngestPostAsWebMessage(SupportRequestTestCase):
post = self.portal.post_module.newContent(
portal_type='HTML Post',
follow_up_value=support_request,
data="Hello"
data=b"Hello"
)
post.publish()
self.tic()
......
......@@ -25,7 +25,7 @@
#
##############################################################################
import urlparse
import six.moves.urllib.parse
from DateTime import DateTime
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
......@@ -343,17 +343,17 @@ class TestPaymentTransactionGroupPaymentSelection(AccountingTestCase):
self.tic()
ret = ptg1.PaymentTransactionGroup_selectPaymentTransactionLineList(select_mode='stopped_or_delivered')
self.assertEqual(
urlparse.parse_qs(urlparse.urlparse(ret).query)['portal_status_message'],
six.moves.urllib.parse.parse_qs(six.moves.urllib.parse.urlparse(ret).query)['portal_status_message'],
['Payment selection in progress.'])
self.commit()
ret = ptg1.PaymentTransactionGroup_selectPaymentTransactionLineList(select_mode='stopped_or_delivered')
self.assertEqual(
urlparse.parse_qs(urlparse.urlparse(ret).query)['portal_status_message'],
six.moves.urllib.parse.parse_qs(six.moves.urllib.parse.urlparse(ret).query)['portal_status_message'],
['Some payments are still beeing processed in the background, please retry later'])
self.commit()
# another PTG is same, because we also want to prevent things like two users selecting
# payments at the same time.
ret = ptg2.PaymentTransactionGroup_selectPaymentTransactionLineList(select_mode='stopped_or_delivered')
self.assertEqual(
urlparse.parse_qs(urlparse.urlparse(ret).query)['portal_status_message'],
six.moves.urllib.parse.parse_qs(six.moves.urllib.parse.urlparse(ret).query)['portal_status_message'],
['Some payments are still beeing processed in the background, please retry later'])
......@@ -25,7 +25,7 @@
#
##############################################################################
import urlparse
import six.moves.urllib.parse
import lxml.etree
from DateTime import DateTime
......@@ -140,7 +140,7 @@ class TestPaymentTransactionGroupPaymentSEPA(AccountingTestCase):
# this XML validates against the schema
xmlschema = lxml.etree.XMLSchema(
lxml.etree.fromstring(str(getattr(self.portal, 'pain.001.001.02.xsd').data)))
lxml.etree.fromstring(bytes(getattr(self.portal, 'pain.001.001.02.xsd').data)))
xmlschema.assertValid(pain)
self.assertEqual(
......@@ -234,7 +234,7 @@ class TestPaymentTransactionGroupPaymentSEPA(AccountingTestCase):
pain = lxml.etree.fromstring(
getattr(ptg, 'PaymentTransactionGroup_viewAsSEPACreditTransferPain.001.001.02')().encode('utf-8'))
xmlschema = lxml.etree.XMLSchema(
lxml.etree.fromstring(str(getattr(self.portal, 'pain.001.001.02.xsd').data)))
lxml.etree.fromstring(bytes(getattr(self.portal, 'pain.001.001.02.xsd').data)))
xmlschema.assertValid(pain)
self.assertEqual(
......@@ -251,7 +251,7 @@ class TestPaymentTransactionGroupPaymentSEPA(AccountingTestCase):
ret = ptg.PaymentTransactionGroup_generateSEPACreditTransferFile(
version='pain.001.001.02')
self.assertEqual(
urlparse.parse_qs(urlparse.urlparse(ret).query)['portal_status_message'],
six.moves.urllib.parse.parse_qs(six.moves.urllib.parse.urlparse(ret).query)['portal_status_message'],
['SEPA Credit Transfer File generated.'])
self.tic()
......@@ -263,7 +263,7 @@ class TestPaymentTransactionGroupPaymentSEPA(AccountingTestCase):
pain = lxml.etree.fromstring(f.getData())
xmlschema = lxml.etree.XMLSchema(
lxml.etree.fromstring(str(getattr(self.portal, 'pain.001.001.02.xsd').data)))
lxml.etree.fromstring(bytes(getattr(self.portal, 'pain.001.001.02.xsd').data)))
xmlschema.assertValid(pain)
self.assertEqual(
......@@ -279,7 +279,7 @@ class TestPaymentTransactionGroupPaymentSEPA(AccountingTestCase):
ret = ptg.PaymentTransactionGroup_generateSEPACreditTransferFile(
version='pain.001.001.02')
self.assertEqual(
urlparse.parse_qs(urlparse.urlparse(ret).query)['portal_status_message'],
six.moves.urllib.parse.parse_qs(six.moves.urllib.parse.urlparse(ret).query)['portal_status_message'],
['Some payments are still beeing processed in the background, please retry later'])
self.tic()
......
......@@ -28,7 +28,7 @@
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from DateTime import DateTime
from PIL import Image
import cStringIO
from io import BytesIO
import math
import os.path
from Products.Localizer.itools.i18n.accept import AcceptLanguage
......@@ -58,8 +58,8 @@ class TestSimplifiedPayslipReport(ERP5TypeTestCase):
# http://snipplr.com/view/757/compare-two-pil-images-in-python/
# http://effbot.org/zone/pil-comparing-images.htm
# http://effbot.org/imagingbook/image.htm
image1 = Image.open(cStringIO.StringIO(image_data_1))
image2 = Image.open(cStringIO.StringIO(image_data_2))
image1 = Image.open(BytesIO(image_data_1))
image2 = Image.open(BytesIO(image_data_2))
# image can be converted into greyscale without transparency
h1 = image1.histogram()
......@@ -181,7 +181,7 @@ class TestSimplifiedPayslipReport(ERP5TypeTestCase):
image_source_pdf_doc.setData(pdf_data)
self.tic()
_, png = image_source_pdf_doc.convert("png", frame=0, quality=100)
self.assertImageRenderingEquals(str(png), str(expected_image.getData()))
self.assertImageRenderingEquals(bytes(png), bytes(expected_image.getData()))
def test_03_payslip_holiday(self):
for i in self.portal.portal_catalog(
......@@ -257,5 +257,3 @@ class TestSimplifiedPayslipReport(ERP5TypeTestCase):
self.assertEqual(payslip_data["report_data"]["total_holiday_this_year"], 2)
self.assertEqual(payslip_data["report_data"]["taken_holiday"], 2)
self.assertEqual(payslip_data["report_data"]["total_holiday_year_before"], 0)
......@@ -115,19 +115,19 @@ class TestERP5PayzenSecurePayment(TestERP5PayzenSecurePaymentMixin):
def test_getSignature_dict_simple(self):
self.assertEqual(
self.service._getSignature({'key': 'value'}, ['key']),
sha1('value+' + self.service_password)
sha1(('value+' + self.service_password).encode())
)
def test_getSignature_dict_key_sort(self):
self.assertEqual(
self.service._getSignature({'key': 'value', 'key1': 'value1'}, ['key',
'key1']),
sha1('value+value1+' + self.service_password)
sha1(('value+value1+' + self.service_password).encode())
)
self.assertEqual(
self.service._getSignature({'key': 'value', 'key1': 'value1'}, ['key1',
'key']),
sha1('value1+value+' + self.service_password)
sha1(('value1+value+' + self.service_password).encode())
)
def test_getSignature_dict_date_as_datetime(self):
......@@ -135,7 +135,7 @@ class TestERP5PayzenSecurePayment(TestERP5PayzenSecurePaymentMixin):
d = {'key': now}
self.assertEqual(
self.service._getSignature(d, ['key']),
sha1(now.strftime('%Y%m%d') + '+' + self.service_password)
sha1((now.strftime('%Y%m%d') + '+' + self.service_password).encode())
)
# dict was updated
self.assertEqual(d['key'], now)
......@@ -158,8 +158,8 @@ class TestERP5PayzenSecurePayment(TestERP5PayzenSecurePaymentMixin):
self.portal.changeSkin(None)
try:
result = self.service.navigate(pt_id, {"key": 'value'})
signature = sha1('value+INTERACTIVE+ERP5+TEST+REGISTER+SINGLE+0123456+V2+'
+ self.service_password)
signature = sha1(('value+INTERACTIVE+ERP5+TEST+REGISTER+SINGLE+0123456+V2+'
+ self.service_password).encode())
self.assertEqual(result, """key=key value=value
key=signature value=%s
key=vads_action_mode value=INTERACTIVE
......
......@@ -26,14 +26,13 @@
##############################################################################
import warnings
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from erp5.component.test.testDms import makeFileUpload
from erp5.component.test.testDms import DocumentUploadTestCase
original_warnings_showwarnings = warnings.showwarning
class TestERP5PDFMerge(ERP5TypeTestCase):
class TestERP5PDFMerge(DocumentUploadTestCase):
def test_showwarning_issue(self):
"""
......@@ -42,7 +41,7 @@ class TestERP5PDFMerge(ERP5TypeTestCase):
"""
self.assertEqual(warnings.showwarning, original_warnings_showwarnings)
document = self.portal.portal_contributions.newContent(
file=makeFileUpload('REF-en-001.pdf'))
file=self.makeFileUpload('REF-en-001.pdf'))
merged_pdf_data = self.portal.ERP5Site_mergePDFList(
[document.getData(), document.getData()])
self.portal.document_module.newContent(
......@@ -53,7 +52,7 @@ class TestERP5PDFMerge(ERP5TypeTestCase):
def test_erp5_merge_pdf(self):
document = self.portal.portal_contributions.newContent(
file=makeFileUpload('REF-en-001.pdf'))
file=self.makeFileUpload('REF-en-001.pdf'))
merged_pdf_data = self.portal.ERP5Site_mergePDFList(
[document.getData(), document.getData()])
merged_document = self.portal.document_module.newContent(
......@@ -63,7 +62,7 @@ class TestERP5PDFMerge(ERP5TypeTestCase):
def test_erp5_merge_pdf_start_on_recto(self):
document = self.portal.portal_contributions.newContent(
file=makeFileUpload('REF-en-001.pdf'))
file=self.makeFileUpload('REF-en-001.pdf'))
merged_pdf_data = self.portal.ERP5Site_mergePDFList(
[document.getData(), document.getData()], start_on_recto=True)
merged_document = self.portal.document_module.newContent(
......
......@@ -135,8 +135,7 @@ class ResourceVariationTestCase(ERP5TypeTestCase):
colour.newContent(portal_type='Category', title='Red',
reference='l',id='red')
if not self.portal_categories.hasContent('individual_aspect'):
aspect = self.portal_categories.newContent(portal_type='Base Category',
aspect = self.portal_categories.newContent(portal_type='Base Category',
title='Individual Aspect',
reference='individual_aspect',
id='individual_aspect',
......
......@@ -45,11 +45,11 @@ class TestWorkflowPerformance(TestPerformanceMixin):
foo_list = []
foo_list_append = foo_list.append
range_10 = range(10)
range_10 = list(range(10))
portal_workflow = self.portal.portal_workflow
foo_count = 100
for x in xrange(foo_count):
for x in range(foo_count):
foo = self.foo_module.newContent()
foo_list_append(foo)
......
......@@ -47,8 +47,6 @@ class TestHTMLPostLogic(ERP5TypeTestCase):
# here, you can create the categories and objects your test will depend on
"""
pass
def test_01_sampleTest(self):
"""
......
......@@ -28,6 +28,7 @@
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from Products.ERP5Type.tests.Sequence import SequenceList
from DateTime import DateTime
import six
class TestRealTimeInventoryAccountingMixin:
def assertIterableLen(self, iterable, expected_len):
......@@ -163,7 +164,7 @@ class TestRealTimeInventoryAccountingMixin:
delivery_property_dict=None,
movement_property_dict_tuple=()):
if delivery_property_dict is not None:
for property_id, property_value in delivery_property_dict.iteritems():
for property_id, property_value in six.iteritems(delivery_property_dict):
self.assertEqual(delivery.getProperty(property_id), property_value)
if not movement_property_dict_tuple:
......
......@@ -26,12 +26,12 @@
#
##############################################################################
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from erp5.component.test.testDms import makeFileUpload
from erp5.component.test.testDms import DocumentUploadTestCase
from time import time
import base64
class TestRunMyDoc(ERP5TypeTestCase):
class TestRunMyDoc(DocumentUploadTestCase):
"""
Basic Test for internal implementation of RunMyDocs
"""
......@@ -96,7 +96,7 @@ class TestRunMyDoc(ERP5TypeTestCase):
Test Screeshot upload script used by Zelenium to
update screenshots of the documents.
"""
image_upload = makeFileUpload('TEST-en-002.png')
image_upload = self.makeFileUpload('TEST-en-002.png')
self.assertNotEqual(None, image_upload)
# Create a web page, and check if the content is not overwriten
......@@ -135,7 +135,7 @@ class TestRunMyDoc(ERP5TypeTestCase):
image_upload.seek(0)
self.assertEqual(image_page_2.getData(), base64.b64decode(image_upload.read()))
self.assertEqual(image_page_2.getFilename(), image_reference + '.png')
self.assertEqual(image_page.getData(), '')
self.assertEqual(image_page.getData(), b'')
def test_viewSeleniumTest(self):
"""
......
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
import transaction
from io import FileIO
from io import BytesIO
import os
class FileUpload(FileIO):
class FileUpload(BytesIO):
"""Act as an uploaded file.
"""
__allow_access_to_unprotected_subobjects__ = 1
def __init__(self, path, name):
self.filename = name
super(FileUpload, self).__init__(path)
with open(path, 'rb') as f:
super(FileUpload, self).__init__(f.read())
self.headers = {}
def makeFilePath(name):
#return os.path.join(os.path.dirname(__file__), 'tmp', name)
return name
def makeFileUpload(name, as_name=None):
if as_name is None:
as_name = name
path = makeFilePath(name)
return FileUpload(path, as_name)
class TestSafeImage(ERP5TypeTestCase):
def afterSetUp(self):
portal = self.getPortalObject()
......@@ -38,14 +29,21 @@ class TestSafeImage(ERP5TypeTestCase):
transaction.commit()
self.tic()
def makeFileUpload(self, name, as_name=None):
if as_name is None:
as_name = name
fu = FileUpload(name, as_name)
self.addCleanup(fu.close)
return fu
def _createImage(self):
portal = self.getPortalObject()
image = portal.restrictedTraverse('portal_skins/erp5_safeimage/img/image_unit_test.jpg')
path_image = "image_unit_test.jpg"
fd = os.open(path_image, os.O_CREAT | os.O_RDWR)
os.write(fd,str(image.data))
os.write(fd, bytes(image.data))
os.close(fd)
_image = makeFileUpload(path_image)
_image = self.makeFileUpload(path_image)
image = self.image_module.newContent(portal_type='Image',title='testImage',
id='testImage',file=_image,filename='testImage')
return image
......@@ -55,9 +53,9 @@ class TestSafeImage(ERP5TypeTestCase):
image = portal.restrictedTraverse('portal_skins/erp5_safeimage/img/image_unit_test.jpg')
path_image = "image_unit_test.jpg"
fd = os.open(path_image, os.O_CREAT | os.O_RDWR)
os.write(fd,str(image.data))
os.write(fd, bytes(image.data))
os.close(fd)
tile_image = makeFileUpload(path_image)
tile_image = self.makeFileUpload(path_image)
tile = self.image_module.newContent(portal_type='Image Tile',title='testTile',
id='testTile',file=tile_image,filename='testTile')
return tile
......@@ -67,9 +65,9 @@ class TestSafeImage(ERP5TypeTestCase):
image = portal.restrictedTraverse('portal_skins/erp5_safeimage/img/image_unit_test.jpg')
path_image = "image_unit_test.jpg"
fd = os.open(path_image, os.O_CREAT | os.O_RDWR)
os.write(fd,str(image.data))
os.write(fd, bytes(image.data))
os.close(fd)
tile_image_transformed = makeFileUpload(path_image)
tile_image_transformed = self.makeFileUpload(path_image)
tile_transformed = self.image_module.newContent(portal_type='Image Tile Transformed',
title='testTileTransformed',id='testTileTransformed',
file=tile_image_transformed,filename='testTileTransformed')
......@@ -97,7 +95,7 @@ class TestSafeImage(ERP5TypeTestCase):
self.assertNotEqual(tile,None)
image_property = getattr(tile, "ImageProperties.xml", None)
self.assertEqual(image_property.getData(),
"""<IMAGE_PROPERTIES WIDTH="660" HEIGHT="495" NUMTILES="9" NUMIMAGES="1" VERSION="1.8" TILESIZE="256" />""")
b"""<IMAGE_PROPERTIES WIDTH="660" HEIGHT="495" NUMTILES="9" NUMIMAGES="1" VERSION="1.8" TILESIZE="256" />""")
self.assertNotEqual(image_property, None)
self.assertEqual("Embedded File", image_property.getPortalType())
image_group = getattr(tile, "TileGroup0", None)
......@@ -127,7 +125,7 @@ class TestSafeImage(ERP5TypeTestCase):
self.assertNotEqual(tile_transformed,None)
image_property = getattr(tile_transformed, "ImageProperties.xml", None)
self.assertEqual(image_property.getData(),
"""<IMAGE_PROPERTIES WIDTH="660" HEIGHT="495" NUMTILES="9" NUMIMAGES="1" VERSION="1.8" TILESIZE="256" />""")
b"""<IMAGE_PROPERTIES WIDTH="660" HEIGHT="495" NUMTILES="9" NUMIMAGES="1" VERSION="1.8" TILESIZE="256" />""")
self.assertNotEqual(image_property, None)
self.assertEqual("Embedded File", image_property.getPortalType())
image_transform = getattr(tile_transformed, "TransformFile.txt", None)
......
......@@ -35,6 +35,7 @@ from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from DateTime import DateTime
from Products.ERP5Type.tests.utils import reindex
from Products.ERP5Type.tests.utils import todo_erp5
import six
class TestBPMMixin(ERP5TypeTestCase):
"""Skeletons for tests which depend on BPM"""
......@@ -172,7 +173,7 @@ class TestBPMMixin(ERP5TypeTestCase):
portal_type=self.trade_model_path_portal_type, **kw)
if criterion_property_dict:
trade_model_path._setCriterionPropertyList(tuple(criterion_property_dict))
for property_, identity in criterion_property_dict.iteritems():
for property_, identity in six.iteritems(criterion_property_dict):
trade_model_path.setCriterion(property_, identity)
reference = kw.get('reference', None)
if reference is not None:
......
......@@ -563,11 +563,11 @@ class TestInvoice(TestInvoiceMixin):
invoice.confirm()
self.tic()
odt = invoice.Invoice_viewAsODT()
import cStringIO
output = cStringIO.StringIO()
from io import BytesIO
output = BytesIO()
output.write(odt)
m = OpenDocumentTextFile(output)
text_content=m.toString().encode('ascii','replace')
text_content=m.toString().encode('ascii','replace').decode()
if text_content.find('Resource Tax') != -1 :
self.fail('fail to delete the tax line in product line')
if text_content.find('Tax Code') == -1 :
......@@ -640,7 +640,7 @@ class TestInvoice(TestInvoiceMixin):
parser = OOoParser()
parser.openFromBytes(odt)
style_xml = parser.oo_files['styles.xml']
self.assertNotIn('<draw:image', style_xml)
self.assertNotIn(b'<draw:image', style_xml)
def test_Invoice_viewAsODT_invalid_image(self):
resource = self.portal.getDefaultModule(
......
......@@ -60,7 +60,7 @@ class TestSimulationPerformance(TestTradeModelLineSale):
def perf_01_invoiceSimpleOrder(self, order_count=1):
start = time()
order = self.portal.unrestrictedTraverse(self._order)
order_list = [self.clone(order) for _ in xrange(order_count)]
order_list = [self.clone(order) for _ in range(order_count)]
for order in order_list:
for line in list(order.getMovementList()):
self.clone(line)
......
......@@ -80,6 +80,7 @@ from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from zLOG import LOG
from Products.ERP5Type.tests.utils import LogInterceptor
from Products.ERP5Type.tests.Sequence import SequenceList
from six.moves import range
# Define variable to chek if performance are good or not
# XXX These variable are specific to the testing environment
......@@ -398,7 +399,7 @@ class TestSimulationPerformance(ERP5TypeTestCase, LogInterceptor):
destination_decision = sequence.get('destination_decision')
destination_administration = sequence.get('destination_administration')
resource = sequence.get('resource')
for i in xrange(number_of_sale_orders):
for i in range(number_of_sale_orders):
start_date = base_date + i
stop_date = base_date + i + 1
order = module.newContent(
......@@ -410,7 +411,7 @@ class TestSimulationPerformance(ERP5TypeTestCase, LogInterceptor):
stop_date=stop_date)
# Set the rest through the trade condition.
order.SaleOrder_applySaleTradeCondition()
for _ in xrange(number_of_sale_order_lines):
for _ in range(number_of_sale_order_lines):
order.newContent(portal_type='Sale Order Line',
resource=resource, quantity=1.0)
......@@ -767,7 +768,7 @@ class TestSimulationPerformance(ERP5TypeTestCase, LogInterceptor):
number = sequence.get('number_of_additional_sale_packing_list_lines')
resource = sequence.get('resource')
for packing_list in module.contentValues(portal_type='Sale Packing List'):
for _ in xrange(number):
for _ in range(number):
packing_list.newContent(portal_type='Sale Packing List Line',
resource=resource, quantity=1.0)
......
......@@ -26,9 +26,8 @@
##############################################################################
import json
from six.moves import urllib
from io import BytesIO
from urlparse import parse_qs
from six.moves.urllib.parse import parse_qs
import requests
import responses
......@@ -461,8 +460,8 @@ class TestStripePaymentSession(ERP5TypeTestCase):
)
ret = self.publish(
"%s/ERP5Site_receiveStripeWebHook" % self.portal.getPath(),
stdin=BytesIO(urllib.parse.urlencode({
"BODY": json.dumps({
stdin=BytesIO(
json.dumps({
"url": "https://stripe.url",
"id": "evt_%s" % "abc321_expired",
"object": "event",
......@@ -474,9 +473,10 @@ class TestStripePaymentSession(ERP5TypeTestCase):
"object": "checkout.session"
}
}
})
}).encode()),
}).encode()
),
request_method="POST",
env={'CONTENT_TYPE': 'application/json'},
handle_errors=False)
self.assertEqual(200, ret.getStatus())
self.tic()
......@@ -658,8 +658,8 @@ class TestStripePaymentSession(ERP5TypeTestCase):
)
ret = self.publish(
"%s/ERP5Site_receiveStripeWebHook" % self.portal.getPath(),
stdin=BytesIO(urllib.parse.urlencode({
"BODY": json.dumps({
stdin=BytesIO(
json.dumps({
"id": "evt_%s" % session_id,
"object": "event",
"data": {
......@@ -670,9 +670,10 @@ class TestStripePaymentSession(ERP5TypeTestCase):
"object": "checkout.session"
}
}
})
}).encode()),
}).encode(),
),
request_method="POST",
env={'CONTENT_TYPE': 'application/json'},
handle_errors=False)
self.assertEqual(200, ret.getStatus())
self.tic()
......
......@@ -4,7 +4,8 @@ import json
from time import sleep
from DateTime import DateTime
import responses
import httplib
import six.moves.http_client
from six.moves import range
class TaskDistributionTestCase(ERP5TypeTestCase):
......@@ -146,6 +147,7 @@ class TaskDistributionTestCase(ERP5TypeTestCase):
"""start_count: number of test line to start
stop_count: number of test line to stop
"""
# pylint:disable=possibly-used-before-assignment
status_dict = {}
test_result_path, revision = self._createTestResult(revision=revision,
test_list=['testFoo', 'testBar'], test_title=test_title, node_title=node_title)
......@@ -167,6 +169,7 @@ class TaskDistributionTestCase(ERP5TypeTestCase):
self.assertEqual(test_result.getSimulationState(), "stopped")
else:
self.assertEqual(test_result.getSimulationState(), "started")
# pylint:enable=possibly-used-before-assignment
def _cleanupTestResult(self):
self.tic()
......@@ -1348,7 +1351,7 @@ class TestTaskDistribution(TaskDistributionTestCase):
zope_partition_dict += "{% endif %}\n"
cluster_configuration += zope_partition_dict + '\n}}'
# -Generate graph coordinate
graph_coordinate = range(1, len(node_list)+1)
graph_coordinate = list(range(1, len(node_list)+1))
# -Create the test suite
self._createTestSuite(quantity=1,priority=1, reference_correction=0,
specialise_value=self.scalability_distributor, portal_type="Scalability Test Suite",
......@@ -1675,7 +1678,7 @@ class TestGitlabRESTConnectorInterface(ERP5TypeTestCase):
self.assertEqual(
self.id(),
body['name'])
return (httplib.CREATED, {'content-type': 'application/json'}, '{}')
return (six.moves.http_client.CREATED, {'content-type': 'application/json'}, b'{}')
return _callback
def test_start_test(self):
......@@ -1692,7 +1695,7 @@ class TestGitlabRESTConnectorInterface(ERP5TypeTestCase):
rsps.add(
responses.POST,
self.post_commit_status_url,
{})
b'{}')
self.test_result.start()
self.tic()
......@@ -1830,7 +1833,7 @@ class TestGitlabRESTConnectorInterface(ERP5TypeTestCase):
self.assertEqual(
'https://erp5js.example.com/#%s' % self.test_result.getRelativeUrl(),
body['target_url'])
return (httplib.CREATED, {'content-type': 'application/json'}, '{}')
return (six.moves.http_client.CREATED, {'content-type': 'application/json'}, b'{}')
with responses.RequestsMock() as rsps:
rsps.add_callback(
......@@ -1912,7 +1915,7 @@ class TestGitlabRESTConnectorInterface(ERP5TypeTestCase):
responses.POST,
self.post_commit_status_url,
json={"message": 'Cannot transition status via :run from :running (Reason(s): Status cannot transition via "run")'},
status=httplib.BAD_REQUEST,
status=six.moves.http_client.BAD_REQUEST,
)
self.test_result.start()
self.tic()
......
......@@ -429,7 +429,7 @@ class TestInventory(TestOrderMixin, ERP5TypeTestCase):
sequence.edit(variation_2=cell_key)
quantity = 3
cell.edit(
quantity = quantity,
quantity = quantity, # pylint:disable=used-before-assignment
predicate_category_list = cell_key,
variation_category_list = cell_key,
mapped_value_property_list = ['quantity'],
......
......@@ -28,8 +28,9 @@
#
##############################################################################
import unittest
import functools
import os
import unittest
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from Products.ERP5Type.tests.utils import FileUpload,\
......@@ -464,7 +465,7 @@ class TestOrderMixin(SubcontentReindexingWrapper):
"""
Split a list and return tuple with the 2 half
"""
middle = len(l)/2 + len(l)%2
middle = int(len(l)//2 + len(l)%2)
return ( l[:middle] , l[middle:] )
def stepSetOrderLineHalfVCL(self,sequence=None, sequence_list=None, **kw):
......@@ -516,7 +517,7 @@ class TestOrderMixin(SubcontentReindexingWrapper):
self.assertEqual(len(cell_key_list), 0)
else:
len_range = [len(x) for x in cell_range]
self.assertEqual(len(cell_key_list), reduce(lambda x,y: x*y, len_range))
self.assertEqual(len(cell_key_list), functools.reduce(lambda x,y: x*y, len_range))
def stepCompleteOrderLineMatrix(self,sequence=None, sequence_list=None, \
**kw):
......@@ -2838,11 +2839,11 @@ class TestOrder(TestOrderMixin, ERP5TypeTestCase):
portal_type='Organisation', title='Client',
default_image_file=image)
from OFS.Image import Pdata
self.assertTrue(isinstance(client.getDefaultImageValue().data, Pdata))
self.assertIsInstance(client.getDefaultImageValue().data, Pdata)
vendor = self.portal.organisation_module.newContent(
portal_type='Organisation', title='Vendor',
default_image_file=image)
self.assertTrue(isinstance(vendor.getDefaultImageValue().data, Pdata))
self.assertIsInstance(vendor.getDefaultImageValue().data, Pdata)
order = self.portal.getDefaultModule(self.order_portal_type).newContent(
portal_type=self.order_portal_type,
specialise=self.business_process,
......
......@@ -27,11 +27,13 @@
#
##############################################################################
from Products.ERP5Type.Utils import ensure_list
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from DateTime import DateTime
from Products.ERP5Type.tests.Sequence import SequenceList
from erp5.component.test.testOrder import TestOrderMixin
from Products.ERP5.tests.testInventoryAPI import InventoryAPITestCase
import six
class TestOrderBuilderMixin(TestOrderMixin, InventoryAPITestCase):
......@@ -156,7 +158,7 @@ class TestOrderBuilderMixin(TestOrderMixin, InventoryAPITestCase):
order_line, = order.contentValues(portal_type=self.order_line_portal_type)
self.assertEqual(order_line.getResourceValue(), resource)
self.assertEqual(order_line.getTotalQuantity(),
sum(self.wanted_quantity_matrix.itervalues()))
sum(six.itervalues(self.wanted_quantity_matrix)))
quantity_matrix = {}
for cell in order_line.contentValues(portal_type=self.order_cell_portal_type):
......@@ -239,7 +241,7 @@ class TestOrderBuilderMixin(TestOrderMixin, InventoryAPITestCase):
self.wanted_quantity_matrix = self.decrease_quantity_matrix.copy()
packing_list_line.setVariationCategoryList(
self.decrease_quantity_matrix.keys(),
ensure_list(self.decrease_quantity_matrix.keys()),
)
self.tic()
......
......@@ -1799,7 +1799,7 @@ class TestPackingList(TestPackingListMixin, ERP5TypeTestCase) :
sale_packing_list2.getUid()))]
self.assertEqual({self.default_quantity-4, self.default_quantity-3},
set([x.getQuantity() for x in sale_packing_list1.getMovementList()]))
self.assertEqual({1, 1}, set([x.getQuantity() for x in sale_packing_list3.getMovementList()]))
self.assertEqual({1}, set([x.getQuantity() for x in sale_packing_list3.getMovementList()]))
self.assertEqual("solved", sale_packing_list3.getCausalityState())
self.assertEqual("solved", sale_packing_list1.getCausalityState())
def getSolverProcessStateList(delivery):
......@@ -1811,7 +1811,7 @@ class TestPackingList(TestPackingListMixin, ERP5TypeTestCase) :
self.assertEqual({self.default_quantity-4, self.default_quantity-3},
set([x.getQuantity() for x in sale_packing_list1.getMovementList()]))
self.assertEqual({1, 2}, set([x.getQuantity() for x in sale_packing_list2.getMovementList()]))
self.assertEqual({2, 2}, set([x.getQuantity() for x in sale_packing_list3.getMovementList()]))
self.assertEqual({2}, set([x.getQuantity() for x in sale_packing_list3.getMovementList()]))
self.assertEqual("solved", sale_packing_list1.getCausalityState())
self.assertEqual("solved", sale_packing_list2.getCausalityState())
self.assertEqual("solved", sale_packing_list3.getCausalityState())
......@@ -1822,7 +1822,7 @@ class TestPackingList(TestPackingListMixin, ERP5TypeTestCase) :
self.assertEqual({self.default_quantity-5, self.default_quantity-4},
set([x.getQuantity() for x in sale_packing_list1.getMovementList()]))
self.assertEqual({2, 3}, set([x.getQuantity() for x in sale_packing_list2.getMovementList()]))
self.assertEqual({2, 2}, set([x.getQuantity() for x in sale_packing_list3.getMovementList()]))
self.assertEqual({2}, set([x.getQuantity() for x in sale_packing_list3.getMovementList()]))
self.assertEqual("solved", sale_packing_list1.getCausalityState())
self.assertEqual("solved", sale_packing_list2.getCausalityState())
self.assertEqual("solved", sale_packing_list3.getCausalityState())
......@@ -1837,7 +1837,7 @@ class TestPackingList(TestPackingListMixin, ERP5TypeTestCase) :
self.assertEqual({self.default_quantity-6, self.default_quantity-5},
set([x.getQuantity() for x in sale_packing_list1.getMovementList()]))
self.assertEqual({2, 3}, set([x.getQuantity() for x in sale_packing_list2.getMovementList()]))
self.assertEqual({3, 3}, set([x.getQuantity() for x in sale_packing_list3.getMovementList()]))
self.assertEqual({3}, set([x.getQuantity() for x in sale_packing_list3.getMovementList()]))
self.assertEqual("solved", sale_packing_list1.getCausalityState())
self.assertEqual("solved", sale_packing_list2.getCausalityState())
self.assertEqual("solved", sale_packing_list3.getCausalityState())
......
......@@ -842,7 +842,7 @@ class TestResource(ERP5TypeTestCase):
destination_section_value=node)
supply.validate()
if 0:
if 0: # pylint:disable=using-constant-test
# XXX if both a supply line for the resource and a supply cell for
# the resource with the exact variation can be applied, one of them
# is choosen randomly. It looks like a bug, but I'm not sure we
......@@ -1412,6 +1412,7 @@ class TestResource(ERP5TypeTestCase):
self.assertEqual(resource.getInternalSupplyLineDestinationReference(),
'test_destination_reference_on_internal_supply_line')
@expectedFailure
def testQuantityUnitOnMovement(self):
"""Make sure that changing default quantity unit on resource does not
affect to movement.
......@@ -1464,7 +1465,8 @@ class TestResource(ERP5TypeTestCase):
# Check existing movement again and make sure that quantity
# unit is not changed.
expectedFailure(self.assertEqual)(
# XXX This is the expectedFailure
self.assertEqual(
sale_order_line.getQuantityUnitValue(),
self.quantity_unit_gram)
......
......@@ -33,6 +33,7 @@ from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5ReportTestCase
from Products.ERP5Type.tests.utils import reindex
from AccessControl.SecurityManagement import newSecurityManager
from DateTime import DateTime
import six
class TestTradeReports(ERP5ReportTestCase):
"""Test Trade reports
......@@ -275,7 +276,7 @@ class TestTradeReports(ERP5ReportTestCase):
if resource_dict is None:
resource_dict = {}
sale_order = self.sale_order_module.newContent(portal_type="Sale Order", **kw)
for product, values in resource_dict.iteritems():
for product, values in six.iteritems(resource_dict):
sale_order.newContent(portal_type="Sale Order Line",
resource=product,
quantity=values["quantity"],
......@@ -296,7 +297,7 @@ class TestTradeReports(ERP5ReportTestCase):
if resource_dict is None:
resource_dict = {}
sale_packing_list = self.portal.sale_packing_list_module.newContent(portal_type="Sale Packing List", **kw)
for product, values in resource_dict.iteritems():
for product, values in six.iteritems(resource_dict):
sale_packing_list.newContent(
portal_type="Sale Packing List Line",
resource=product,
......
......@@ -87,7 +87,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
cell_range = [['1', '2', '3'], ['a', 'b', 'c']]
kwd = {'base_id' : 'quantity'}
matrix.renameCellRange(*cell_range, **kwd)
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
i = 0
for place in cartesianProduct(cell_range):
......@@ -98,7 +98,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
cell_range = [['2', '3', '4'], ['b', 'c', 'd']]
matrix.renameCellRange(*cell_range, **kwd)
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
i = 0
for place in cartesianProduct(cell_range):
cell = matrix.getCell(*place, **kwd)
......@@ -109,7 +109,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
cell_range = [['1', '2', '3', '4'], ['a', 'b', 'c', 'd']]
value_list = (0, 1, 2, None, 3, 4, 5, None, 6, 7, 8, None, None, None, None, None)
matrix.renameCellRange(*cell_range, **kwd)
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
i = 0
for place in cartesianProduct(cell_range):
cell = matrix.getCell(*place, **kwd)
......@@ -123,7 +123,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
cell_range = [['1', '2'], ['a', 'b']]
value_list = (0, 1, 3, 4)
matrix.renameCellRange(*cell_range, **kwd)
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
i = 0
for place in cartesianProduct(cell_range):
cell = matrix.getCell(*place, **kwd)
......@@ -134,7 +134,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
cell_range = [['3'], ['a', 'b', 'c'], ['A', 'B', 'C']]
value_list = (0, None, None, 1, None, None, None, None, None)
matrix.renameCellRange(*cell_range, **kwd)
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
i = 0
for place in cartesianProduct(cell_range):
cell = matrix.getCell(*place, **kwd)
......@@ -148,7 +148,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
cell_range = [['1', '2'], ['A', 'B']]
value_list = (0, 1, None, None)
matrix.renameCellRange(*cell_range, **kwd)
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
i = 0
for place in cartesianProduct(cell_range):
cell = matrix.getCell(*place, **kwd)
......@@ -169,7 +169,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
cell_range = [['1', '2'], ['A', 'B'], ['a', 'b']]
value_list = (0, None, 1, None, 2, None, 3, None)
matrix.renameCellRange(*cell_range, **kwd)
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
i = 0
for place in cartesianProduct(cell_range):
cell = matrix.getCell(*place, **kwd)
......@@ -197,7 +197,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
cell_range = [['1', '2', '3'], ['a', 'b', 'c']]
kwd = {'base_id' : 'quantity'}
matrix.setCellRange(*cell_range, **kwd)
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
for place in cartesianProduct(cell_range):
matrix.newCell(portal_type="Purchase Order Cell",
......@@ -211,7 +211,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
matrix.setCellRange(*cell_range, **kwd)
# We must commit transaction in order to put cell reindexing in activity queue
self.commit()
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
next_cell_id_list = list(matrix.objectIds())
# the cells on coordinates 2b, 3b, 3b and 3c are kept
self.assertEqual(4, len(next_cell_id_list))
......@@ -231,7 +231,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
cell_range = [['0', '1'], ['a','b']]
matrix.setCellRange(*cell_range, **kwd)
self.commit()
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
next2_cell_id_list = list(matrix.objectIds())
removed_id_list = [x for x in next_cell_id_list if x not in next2_cell_id_list]
self.tic()
......@@ -244,7 +244,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
kwd = {'base_id' : 'movement'}
matrix.setCellRange(*cell_range, **kwd)
self.commit()
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
self.tic()
for id_ in next2_cell_id_list:
self.assertFalse(catalog.hasPath(url + '/' + id_))
......@@ -259,7 +259,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
# if we keep the same range, nothing happens
matrix.setCellRange(*cell_range, **kwd)
self.commit()
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
self.assertEqual(len(matrix.getCellValueList(**kwd)), 2)
self.tic()
......@@ -271,7 +271,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
cell_range = [['0', '2'], ['a', ], ['Z']]
matrix.setCellRange(*cell_range, **kwd)
self.commit()
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
self.tic()
# in this case, cells has been removed
......@@ -291,7 +291,7 @@ class TestXMLMatrix(ERP5TypeTestCase, LogInterceptor):
cell_range = [['1', '2'], ['a', ], ['X']]
matrix.setCellRange(*cell_range, **kwd)
self.commit()
self.assertEqual(map(set, matrix.getCellRange(**kwd)), map(set, cell_range))
self.assertEqual(list(map(set, matrix.getCellRange(**kwd))), list(map(set, cell_range)))
self.tic()
# in this case, cells has been removed
......
......@@ -38,7 +38,8 @@ from Testing import ZopeTestCase
from Products.ERP5Type.Globals import get_request
from Products.ERP5Type.tests.utils import createZODBPythonScript
from ZPublisher.HTTPRequest import FileUpload
from StringIO import StringIO
from six.moves import cStringIO as StringIO
import six
from Products.ERP5Form.Selection import Selection
from Products.Formulator.TALESField import TALESMethod
from Products.ERP5Form.ListBox import ListBoxHTMLRenderer
......@@ -242,12 +243,16 @@ class TestListBox(ERP5TypeTestCase):
# We create a script to use as a list method
list_method_id = 'ListBox_ParametersListMethod'
if six.PY2:
list_method_code = """return [context.asContext(alternate_title = u'\xe9lisa'.encode('utf8'))]"""
else:
list_method_code = """return [context.asContext(alternate_title = '\xe9lisa')]"""
createZODBPythonScript(
portal.portal_skins.custom,
list_method_id,
'selection=None, **kw',
"""return [context.asContext(alternate_title = u'\xe9lisa'.encode('utf8'))]""")
list_method_code,
)
# set the listbox to use this as list method
listbox = portal.FooModule_viewFooList.listbox
listbox.ListBox_setPropertyList(
......@@ -760,8 +765,9 @@ class TestListBox(ERP5TypeTestCase):
)
self.assertEqual(result.getStatus(), 500)
body = result.getBody()
self.assertIn('Error Type: TimeoutReachedError', body)
self.assertIn('Error Value: 1969: Query execution was interrupted (max_statement_time exceeded): SET STATEMENT', body)
self.assertIn(b'Error Type: TimeoutReachedError', body)
self.assertIn(b'Error Value: 1969: Query execution was interrupted (max_statement_time exceeded):', body)
self.assertIn(b'SET STATEMENT max_statement_time=', body)
def test_zodb_timeout(self):
portal = self.getPortal()
......@@ -798,7 +804,7 @@ return context.objectValues()
'%s:%s' % (self.manager_username, self.manager_password),
)
self.assertEqual(result.getStatus(), 500)
self.assertIn('Error Type: TimeoutReachedError', result.getBody())
self.assertIn(b'Error Type: TimeoutReachedError', result.getBody())
def test_suite():
suite = unittest.TestSuite()
......
......@@ -37,6 +37,7 @@ from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from zLOG import LOG
from Products.ERP5Type.tests.utils import LogInterceptor
import os
from six.moves import range
# Define variable to chek if performance are good or not
# XXX These variable are specific to the testing environment
......@@ -210,7 +211,7 @@ class TestPerformance(TestPerformanceMixin):
self.tic()
# Check performance
before_view = time()
for _ in xrange(100):
for _ in range(100):
# XXX: Note that we don't clean TransactionVariable cache and REQUEST
# before each call to 'view' requests. In reality, they would be
# always empty at the beginning of such requests.
......@@ -250,9 +251,9 @@ class TestPerformance(TestPerformanceMixin):
# call view once to fill caches
self.bar_module.BarModule_viewBarList()
# add object in bar module
for i in xrange(10):
for i in range(10):
def add():
for x in xrange(100):
for x in range(100):
self.bar_module.newContent(portal_type='Bar',
title='Bar Test',
quantity="%4d" %(x,))
......@@ -271,7 +272,7 @@ class TestPerformance(TestPerformanceMixin):
after_tic = time()
gc.collect()
before_form = time()
for _ in xrange(100):
for _ in range(100):
self.bar_module.BarModule_viewBarList()
after_form = time()
# store result
......@@ -333,7 +334,7 @@ class TestPerformance(TestPerformanceMixin):
self.tic()
# Check performance
before_view = time()
for _ in xrange(100):
for _ in range(100):
foo.Foo_viewProxyField()
after_view = time()
req_time = (after_view - before_view)/100.
......@@ -358,13 +359,13 @@ class TestPerformance(TestPerformanceMixin):
"""
foo = self.foo_module.newContent(portal_type='Foo',
title='Foo Test')
for i in xrange(100):
for i in range(100):
foo.newContent(portal_type='Foo Line',
title='Line %s' % i)
self.tic()
# Check performance
before_view = time()
for _ in xrange(100):
for _ in range(100):
foo.Foo_viewPerformance()
after_view = time()
req_time = (after_view - before_view)/100.
......@@ -403,7 +404,7 @@ class TestPropertyPerformance(TestPerformanceMixin):
def _benchmark(self, nb_iterations, min_time, max_time):
def decorated(f):
before = time()
for i in xrange(nb_iterations):
for i in range(nb_iterations):
f(i)
after = time()
total_time = (after - before) / 100.
......
......@@ -29,7 +29,7 @@
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from Products.ERP5Type.tests.Sequence import SequenceList
from Products.ERP5Type.Globals import get_request
from StringIO import StringIO
from six.moves import cStringIO as StringIO
from DateTime import DateTime
class DummyFieldStorage:
......
......@@ -32,8 +32,8 @@ import re
import requests
import time
from unittest import expectedFailure, skip
from StringIO import StringIO
from urllib import urlencode
from six.moves import cStringIO as StringIO
from six.moves.urllib.parse import urlencode
from AccessControl import Unauthorized
from Testing import ZopeTestCase
from DateTime import DateTime
......@@ -62,7 +62,7 @@ class WebTraversalHookTestMixin(object):
"""
self.assertEqual(1, len(self.web_section.__before_traverse__))
self.assertIsInstance(
self.web_section.__before_traverse__.values()[0],
list(self.web_section.__before_traverse__.values())[0],
self.traversal_hook_class)
def test_TraversalHook_on_clone(self):
......@@ -259,6 +259,7 @@ class TestERP5Web(ERP5TypeTestCase):
page.edit(text_content='<p>Hé Hé Hé!</p>', content_type='text/html')
self.tic()
self.assertEqual('Hé Hé Hé!', page.asText().strip())
self.assertIn('Hé Hé Hé!', page.getSearchableText())
def test_WebPageAsTextHTMLEntities(self):
"""Check if Web Page's asText() converts html entities properly
......@@ -511,7 +512,7 @@ Hé Hé Hé!""", page.asText().strip())
'15': dict(language='pt', version="2", reference="F"),
'16': dict(language='', version="1", reference="A"),
}
sequence_one = property_dict.keys()
sequence_one = list(property_dict.keys())
sequence_two = ['01', '13', '12', '09', '06', '15', '04', '11', '02',
'05', '03', '07', '10', '08', '14', '16']
sequence_three = ['05', '12', '13', '14', '06', '09', '10', '07',
......@@ -1023,12 +1024,10 @@ Hé Hé Hé!""", page.asText().strip())
web_section_portal_type = 'Web Section'
web_section = website.newContent(portal_type=web_section_portal_type)
content = '<p>initial text</p>'
new_content = '<p>modified text<p>'
document = portal.web_page_module.newContent(portal_type='Web Page',
id='document_cache',
reference='NXD-Document.Cache',
text_content=content)
text_content='<p>initial text</p>')
document.publish()
self.tic()
self.assertEqual(document.asText().strip(), 'initial text')
......@@ -1042,15 +1041,15 @@ Hé Hé Hé!""", page.asText().strip())
# Through the web_site.
path = website.absolute_url_path() + '/NXD-Document.Cache'
response = self.publish(path, self.credential)
self.assertNotEqual(response.getBody().find(content), -1)
self.assertIn(b'<p>initial text</p>', response.getBody())
# Through a web_section.
path = web_section.absolute_url_path() + '/NXD-Document.Cache'
response = self.publish(path, self.credential)
self.assertNotEqual(response.getBody().find(content), -1)
self.assertIn(b'<p>initial text</p>', response.getBody())
# modified the web_page content
document.edit(text_content=new_content)
document.edit(text_content='<p>modified text</p>')
self.assertEqual(document.asText().strip(), 'modified text')
self.tic()
......@@ -1058,12 +1057,12 @@ Hé Hé Hé!""", page.asText().strip())
# Through the web_site.
path = website.absolute_url_path() + '/NXD-Document.Cache'
response = self.publish(path, self.credential)
self.assertNotEqual(response.getBody().find(new_content), -1)
self.assertIn(b'<p>modified text</p>', response.getBody())
# Through a web_section.
path = web_section.absolute_url_path() + '/NXD-Document.Cache'
response = self.publish(path, self.credential)
self.assertNotEqual(response.getBody().find(new_content), -1)
self.assertIn(b'<p>modified text</p>', response.getBody())
def test_13a_DocumentMovedCache(self):
"""
......@@ -1114,12 +1113,10 @@ Hé Hé Hé!""", page.asText().strip())
web_section_portal_type = 'Web Section'
web_section = website.newContent(portal_type=web_section_portal_type)
content = '<p>initial text</p>'
new_content = '<p>modified text</p>'
document = portal.web_page_module.newContent(portal_type='Web Page',
id='document_cache',
reference='NXD-Document.Cache',
text_content=content)
text_content='<p>initial text</p>')
document.publish()
self.tic()
self.assertEqual(document.asText().strip(), 'initial text')
......@@ -1127,16 +1124,16 @@ Hé Hé Hé!""", page.asText().strip())
# Through the web_site.
path = website.absolute_url_path() + '/NXD-Document.Cache'
response = self.publish(path, self.credential)
self.assertNotEqual(response.getBody().find(content), -1)
self.assertIn(b'<p>initial text</p>', response.getBody())
# Through a web_section.
path = web_section.absolute_url_path() + '/NXD-Document.Cache'
response = self.publish(path, self.credential)
self.assertNotEqual(response.getBody().find(content), -1)
self.assertIn(b'<p>initial text</p>', response.getBody())
# Modify the web_page content
# Use unrestrictedTraverse (XXX-JPS reason unknown)
web_document = website.unrestrictedTraverse('web_page_module/%s' % document.getId())
web_document.edit(text_content=new_content)
web_document.edit(text_content='<p>modified text</p>')
# Make sure cached is emptied
self.assertFalse(web_document.hasConversion(format='txt'))
self.assertFalse(document.hasConversion(format='txt'))
......@@ -1161,14 +1158,14 @@ Hé Hé Hé!""", page.asText().strip())
self.assertEqual(web_document.asText().strip(), 'modified text')
path = web_section.absolute_url_path() + '/NXD-Document.Cache'
response = self.publish(path, self.credential)
self.assertNotEqual(response.getBody().find(new_content), -1)
self.assertIn(b'<p>modified text</p>', response.getBody())
# Through a web_site.
web_document = website.restrictedTraverse('NXD-Document.Cache')
self.assertEqual(web_document.asText().strip(), 'modified text')
path = website.absolute_url_path() + '/NXD-Document.Cache'
response = self.publish(path, self.credential)
self.assertNotEqual(response.getBody().find(new_content), -1)
self.assertIn(b'<p>modified text</p>', response.getBody())
def test_14_AccessWebSiteForWithDifferentUserPreferences(self):
"""Check that Ram Cache Manager do not mix websection
......@@ -1230,18 +1227,18 @@ Hé Hé Hé!""", page.asText().strip())
# connect as administrator and check that only developper_mode is enable
response = self.publish(websection_url, 'administrator:administrator')
self.assertIn('manage_main', response.getBody())
self.assertNotIn('manage_messages', response.getBody())
self.assertIn(b'manage_main', response.getBody())
self.assertNotIn(b'manage_messages', response.getBody())
# connect as webeditor and check that only translator_mode is enable
response = self.publish(websection_url, 'webeditor:webeditor')
self.assertNotIn('manage_main', response.getBody())
self.assertIn('manage_messages', response.getBody())
self.assertNotIn(b'manage_main', response.getBody())
self.assertIn(b'manage_messages', response.getBody())
# anonymous user doesn't exists, check anonymous access without preferences
response = self.publish(websection_url, 'anonymous:anonymous')
self.assertNotIn('manage_main', response.getBody())
self.assertNotIn('manage_messages', response.getBody())
self.assertNotIn(b'manage_main', response.getBody())
self.assertNotIn(b'manage_messages', response.getBody())
def test_15_Check_LastModified_Header(self):
"""Checks that Last-Modified header set by caching policy manager
......@@ -1337,6 +1334,7 @@ Hé Hé Hé!""", page.asText().strip())
conditional_get_response = requests.get(
web_section.absolute_url(),
headers={'If-Modified-Since': DateTime().utcdatetime().strftime('%a, %d %b %Y %H:%M:%S UTC')},
timeout=5,
)
self.assertEqual(conditional_get_response.status_code, 304)
self.assertIn('Cache-Control', conditional_get_response.headers)
......@@ -1409,7 +1407,7 @@ Hé Hé Hé!""", page.asText().strip())
self.assertEqual(HTTP_OK, response.getStatus())
self.assertEqual('text/html; charset=utf-8',
response.getHeader('content-type'))
self.assertIn("Data updated.", response.getBody())
self.assertIn(b"Data updated.", response.getBody())
self.tic()
......@@ -1467,7 +1465,7 @@ Hé Hé Hé!""", page.asText().strip())
self.assertEqual(HTTP_OK, response.getStatus())
self.assertEqual('text/html; charset=utf-8',
response.getHeader('content-type'))
self.assertIn("Data updated.", response.getBody())
self.assertIn(b"Data updated.", response.getBody())
self.tic()
......
......@@ -28,11 +28,11 @@ from __future__ import print_function
import unittest
from Products.ERP5Type.tests.ERP5TypeFunctionalTestCase import ERP5TypeFunctionalTestCase
from SimpleHTTPServer import SimpleHTTPRequestHandler
from six.moves.SimpleHTTPServer import SimpleHTTPRequestHandler
from threading import Thread
from datetime import datetime
import SocketServer
import six.moves.socketserver
import tempfile
import shutil
import time
......@@ -74,7 +74,7 @@ class TestZeleniumCore(ERP5TypeFunctionalTestCase):
root_title = "TEST Instance Tree"
def start_httpd_server(self, root_folder):
self.httpd = SocketServer.TCPServer(('localhost', 5378), CustomHTTPRequestHandler)
self.httpd = six.moves.socketserver.TCPServer(('localhost', 5378), CustomHTTPRequestHandler)
self.httpd.timeout = 2
os.chdir(root_folder)
#self.httpd.serve_forever()
......
......@@ -42,7 +42,7 @@ class TestRenderJSPortalType(ERP5TypeTestCase):
portal_type='Web Style',
reference='test_web_style.css'
)
web_style.setTextContent('/* cl\xc3\xa0sse */ .classe { background: red }')
web_style.setTextContent(b'/* cl\xc3\xa0sse */ .classe { background: red }'.decode('utf-8'))
web_style.publish()
self.tic()
self.assertEqual('text/css', web_style.getContentType())
......@@ -52,7 +52,7 @@ class TestRenderJSPortalType(ERP5TypeTestCase):
'%s/%s' % (self.web_site.getPath(), web_style.getReference())
)
self.assertEqual(
'/* cl\xc3\xa0sse */ .classe { background: red }',
b'/* cl\xc3\xa0sse */ .classe { background: red }',
response.getBody()
)
self.assertEqual(
......@@ -65,7 +65,7 @@ class TestRenderJSPortalType(ERP5TypeTestCase):
portal_type='Web Script',
reference='test_web_script.js'
)
web_script.setTextContent('alert("h\xc3\xa9h\xc3\xa9")')
web_script.setTextContent(b'alert("h\xc3\xa9h\xc3\xa9")'.decode('utf-8'))
web_script.publish()
self.tic()
self.assertEqual('application/javascript', web_script.getContentType())
......@@ -75,7 +75,7 @@ class TestRenderJSPortalType(ERP5TypeTestCase):
'%s/%s' % (self.web_site.getPath(), web_script.getReference())
)
self.assertEqual(
'alert("h\xc3\xa9h\xc3\xa9")',
b'alert("h\xc3\xa9h\xc3\xa9")',
response.getBody()
)
self.assertEqual(
......
......@@ -25,9 +25,9 @@
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
import StringIO
import textwrap
import time
import io
from Products.ERP5Type.tests.utils import createZODBPythonScript
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
......@@ -332,12 +332,12 @@ class TestRenderUpdateTranslationData(RenderJSUpgradeTestCase):
def test_WebSite_getTranslationDataTextContent_extract_from_file(self):
self.portal.portal_skins.custom.manage_addProduct['OFS'].manage_addFile(
'test_portal_skins_gadget.html',
file=StringIO.StringIO(textwrap.dedent('''
file=io.BytesIO(textwrap.dedent('''
<html>
<!--
data-i18n=Message from file
-->
</html>''')))
</html>''').encode()))
self.portal.changeSkin(None) # refresh skin cache
translation_data_text_content = self.web_site.WebSite_getTranslationDataTextContent()
self.assertIn('"Message from file":', translation_data_text_content)
......
......@@ -27,7 +27,7 @@
import os
import unittest
import urlparse
import six.moves.urllib.parse
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
......@@ -36,7 +36,7 @@ class TestSFTPConnection(ERP5TypeTestCase):
if os.environ.get("testSFTPConnection_SFTP_URL"):
def afterSetUp(self):
url = os.environ["testSFTPConnection_SFTP_URL"]
parsed_url = urlparse.urlparse(url)
parsed_url = six.moves.urllib.parse.urlparse(url)
self.connection = self.portal.portal_web_services.newContent(
portal_type='FTP Connector',
reference=self.id(),
......
......@@ -2,7 +2,7 @@
# Copyright (c) 2002-2015 Nexedi SA and Contributors. All Rights Reserved.
from json import dumps
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from httplib import HTTPSConnection
from six.moves.http_client import HTTPSConnection
from erp5.component.mixin.RESTAPIClientConnectorMixin import RESTAPIClientConnectorMixin
from ssl import SSLError
from Products.ERP5Type.Timeout import TimeoutReachedError
......@@ -71,11 +71,13 @@ class TestRESTAPIClientConnector(ERP5TypeTestCase):
with mock.patch(
'ssl.create_default_context',
) as mock_ssl_create_default_context, mock.patch(
'httplib.HTTPSConnection.request',
'six.moves.http_client.HTTPSConnection.request',
) as mock_https_connection_request, mock.patch(
'httplib.HTTPSConnection.getresponse',
'six.moves.http_client.HTTPSConnection.getresponse',
return_value=HTTPResponse_getresponse()
), mock.patch('httplib.HTTPSConnection', return_value=HTTPSConnection) as mock_https_connection:
), mock.patch(
'erp5.component.mixin.RESTAPIClientConnectorMixin.HTTPSConnection',
return_value=HTTPSConnection) as mock_https_connection:
header_dict, body_dict, status = self.rest_api_client_connection.call(
archive_resource=None,
method='POST',
......@@ -145,9 +147,9 @@ class TestRESTAPIClientConnector(ERP5TypeTestCase):
with mock.patch(
'ssl.create_default_context',
), mock.patch(
'httplib.HTTPSConnection.request',
'six.moves.http_client.HTTPSConnection.request',
), mock.patch(
'httplib.HTTPSConnection.getresponse',
'six.moves.http_client.HTTPSConnection.getresponse',
return_value=HTTPResponse_getresponse(498)
):
with self.assertRaises(RESTAPIError) as error:
......@@ -175,9 +177,9 @@ class TestRESTAPIClientConnector(ERP5TypeTestCase):
with mock.patch(
'ssl.create_default_context',
), mock.patch(
'httplib.HTTPSConnection.request',
'six.moves.http_client.HTTPSConnection.request',
), mock.patch(
'httplib.HTTPSConnection.getresponse',
'six.moves.http_client.HTTPSConnection.getresponse',
) as mock_https_connection_getresponse:
mock_https_connection_getresponse.side_effect = SSLError('The read operation timed out')
self.assertRaises(
......
......@@ -27,7 +27,6 @@
#
##############################################################################
import base64
import hashlib
import random
......@@ -55,5 +54,5 @@ class ShaCacheMixin(object):
}
self.shacache_url = self.shacache.absolute_url()
self.tic()
self.data = 'Random Content. %s' % str(random.random())
self.data = ('Random Content. %s' % random.random()).encode()
self.key = hashlib.sha512(self.data).hexdigest()
......@@ -28,8 +28,8 @@
##############################################################################
import httplib
import urlparse
import six.moves.http_client
import six.moves.urllib.parse
from unittest import expectedFailure
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from erp5.component.test.ShaCacheMixin import ShaCacheMixin
......@@ -49,8 +49,8 @@ class TestShaCache(ShaCacheMixin, ERP5TypeTestCase):
"""
Post the file
"""
parsed = urlparse.urlparse(self.shacache_url)
connection = httplib.HTTPConnection(parsed.hostname, parsed.port)
parsed = six.moves.urllib.parse.urlparse(self.shacache_url)
connection = six.moves.http_client.HTTPConnection(parsed.hostname, parsed.port)
try:
connection.request('POST', parsed.path, self.data, self.header_dict)
result = connection.getresponse()
......@@ -67,8 +67,8 @@ class TestShaCache(ShaCacheMixin, ERP5TypeTestCase):
if key is None:
key = self.key
parsed = urlparse.urlparse(self.shacache_url)
connection = httplib.HTTPConnection(parsed.hostname, parsed.port)
parsed = six.moves.urllib.parse.urlparse(self.shacache_url)
connection = six.moves.http_client.HTTPConnection(parsed.hostname, parsed.port)
try:
connection.request('GET', '/'.join([parsed.path, key]), None, {})
result = connection.getresponse()
......@@ -82,8 +82,8 @@ class TestShaCache(ShaCacheMixin, ERP5TypeTestCase):
Check if the PUT method is creating an object.
"""
result, data = self.postFile()
self.assertEqual(result, httplib.CREATED)
self.assertEqual(data, self.key)
self.assertEqual(result, six.moves.http_client.CREATED)
self.assertEqual(data, self.key.encode())
self.tic()
......@@ -100,8 +100,8 @@ class TestShaCache(ShaCacheMixin, ERP5TypeTestCase):
Check if the file returned is the correct.
"""
result, data = self.postFile()
self.assertEqual(result, httplib.CREATED)
self.assertEqual(data, self.key)
self.assertEqual(result, six.moves.http_client.CREATED)
self.assertEqual(data, self.key.encode())
self.tic()
......@@ -109,7 +109,7 @@ class TestShaCache(ShaCacheMixin, ERP5TypeTestCase):
self.assertNotEqual(None, document)
result, data = self.getFile()
self.assertEqual(result, httplib.OK)
self.assertEqual(result, six.moves.http_client.OK)
self.assertEqual(data, self.data)
def test_put_file_twice(self):
......@@ -131,6 +131,7 @@ class TestShaCache(ShaCacheMixin, ERP5TypeTestCase):
self.assertEqual('published', document2.getValidationState())
self.assertEqual('archived', document.getValidationState())
@expectedFailure
def test_put_file_twice_no_tic(self):
self.postFile()
self.commit()
......@@ -140,5 +141,7 @@ class TestShaCache(ShaCacheMixin, ERP5TypeTestCase):
document_list = self.portal.portal_catalog(reference=self.key)
self.assertEqual(2, len(document_list))
expectedFailure(self.assertEqual)(sorted(['archived', 'published']),
sorted(q.getValidationState() for q in document_list))
# this is the expected failure
self.assertEqual(
sorted(['archived', 'published']),
sorted(q.getValidationState() for q in document_list))
......@@ -45,10 +45,9 @@ class ShaDirMixin(object):
Initialize the ERP5 site.
"""
self.login()
self.portal = self.getPortal()
self.key = 'mykey' + str(random.random())
self.file_content = 'This is the content.'
self.file_content = b'This is the content.'
self.file_sha512sum = hashlib.sha512(self.file_content).hexdigest()
self.distribution = 'pypi'
self.creation_date = DateTime()
......@@ -63,9 +62,9 @@ class ShaDirMixin(object):
'expiration_date': str(self.expiration_date),
'distribution': self.distribution,
'architecture': self.architecture}),
b64encode("User SIGNATURE goes here.")]
b64encode(b"User SIGNATURE goes here.").decode()]
self.data = json.dumps(self.data_list)
self.data = json.dumps(self.data_list).encode()
self.sha512sum = hashlib.sha512(self.data).hexdigest()
self.header_dict = {
......
......@@ -27,15 +27,17 @@
#
##############################################################################
import six.moves.http_client
import six.moves.urllib.parse
import hashlib
import httplib
import urlparse
import json
import random
from base64 import b64encode
from unittest import expectedFailure
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from erp5.component.test.ShaDirMixin import ShaDirMixin
from Products.ERP5Type.Utils import bytes2str
class TestShaDir(ShaDirMixin, ERP5TypeTestCase):
"""
......@@ -53,8 +55,8 @@ class TestShaDir(ShaDirMixin, ERP5TypeTestCase):
Post the information calling the Python Script.
It simulates the real usage.
"""
parsed = urlparse.urlparse(self.shadir_url)
connection = httplib.HTTPConnection(parsed.hostname, parsed.port)
parsed = six.moves.urllib.parse.urlparse(self.shadir_url)
connection = six.moves.http_client.HTTPConnection(parsed.hostname, parsed.port)
try:
connection.request('PUT', '/'.join([parsed.path, key or self.key]),
data or self.data, self.header_dict)
......@@ -62,16 +64,16 @@ class TestShaDir(ShaDirMixin, ERP5TypeTestCase):
data = result.read()
finally:
connection.close()
self.assertEqual(result.status, httplib.CREATED)
self.assertEqual(data, '')
self.assertEqual(result.status, six.moves.http_client.CREATED)
self.assertEqual(data, b'')
def getInformation(self, key=None):
"""
Get the information calling the Python Script.
It simulates the real usage.
"""
parsed = urlparse.urlparse(self.shadir_url)
connection = httplib.HTTPConnection(parsed.hostname, parsed.port)
parsed = six.moves.urllib.parse.urlparse(self.shadir_url)
connection = six.moves.http_client.HTTPConnection(parsed.hostname, parsed.port)
try:
connection.request('GET', '/'.join([parsed.path, key or self.key]),
self.data, self.header_dict)
......@@ -103,19 +105,23 @@ class TestShaDir(ShaDirMixin, ERP5TypeTestCase):
data_set = self.portal.portal_catalog.getResultValue(
reference=self.key)
self.assertEqual(self.key, data_set.getReference())
self.assertNotEqual(self.key, data_set.getId())
self.assertEqual('published', data_set.getValidationState())
self.assertEqual(len(self.portal.data_set_module.contentValues()), 1)
# Asserting Document
document = self.portal.portal_catalog.getResultValue(
reference=self.sha512sum)
self.assertEqual(self.sha512sum, document.getTitle())
self.assertEqual(self.sha512sum, document.getReference())
self.assertNotEqual(self.sha512sum, document.getId())
self.assertEqual(self.data, document.getData())
self.assertEqual(data_set, document.getFollowUpValue())
self.assertEqual(str(self.expiration_date),
str(document.getExpirationDate()))
self.assertEqual('application/json', document.getContentType())
self.assertEqual('Published', document.getValidationStateTitle())
self.assertEqual(len(self.portal.document_module.contentValues()), 1)
def test_get_information(self):
"""
......@@ -126,12 +132,12 @@ class TestShaDir(ShaDirMixin, ERP5TypeTestCase):
self.tic()
result, data = self.getInformation()
self.assertEqual(result, httplib.OK)
self.assertEqual(result, six.moves.http_client.OK)
information_list = json.loads(data)
self.assertEqual(1, len(information_list))
self.assertEqual(json.dumps(information_list[0]), self.data)
self.assertEqual(json.dumps(information_list[0]), bytes2str(self.data))
def test_post_information_more_than_once(self):
"""
......@@ -158,12 +164,13 @@ class TestShaDir(ShaDirMixin, ERP5TypeTestCase):
q.getValidationState() for q in document_list]))
result, data = self.getInformation()
self.assertEqual(result, httplib.OK)
self.assertEqual(result, six.moves.http_client.OK)
information_list = json.loads(data)
self.assertEqual(1, len(information_list))
self.assertEqual(json.dumps(information_list[0]), self.data)
self.assertEqual(json.dumps(information_list[0]), bytes2str(self.data))
@expectedFailure
def test_post_information_more_than_once_no_tic(self):
"""
Check if posting information is working.
......@@ -174,7 +181,8 @@ class TestShaDir(ShaDirMixin, ERP5TypeTestCase):
self.postInformation()
self.tic()
expectedFailure(self.assertEqual)(1,
# XXX this is the expected failure
self.assertEqual(1,
self.portal.portal_catalog.countResults(reference=self.key)[0][0])
data_set = self.portal.portal_catalog.getResultValue(
reference=self.key)
......@@ -196,11 +204,11 @@ class TestShaDir(ShaDirMixin, ERP5TypeTestCase):
self.tic()
result, data = self.getInformation()
self.assertEqual(result, httplib.OK)
self.assertEqual(result, six.moves.http_client.OK)
information_list = json.loads(data)
self.assertEqual(1, len(information_list))
self.assertEqual(json.dumps(information_list[0]), self.data)
self.assertEqual(json.dumps(information_list[0]), bytes2str(self.data))
def test_get_information_from_different_data_set(self):
"""
......@@ -215,7 +223,7 @@ class TestShaDir(ShaDirMixin, ERP5TypeTestCase):
self.postInformation()
self.tic()
sha512_2 = hashlib.sha512(str(random.random())).hexdigest()
sha512_2 = hashlib.sha512(str(random.random()).encode()).hexdigest()
key_2 = 'another_key' + str(random.random())
data_list_2 = [json.dumps({
'sha512': sha512_2,
......@@ -223,7 +231,7 @@ class TestShaDir(ShaDirMixin, ERP5TypeTestCase):
'expiration_date': str(self.expiration_date),
'distribution': self.distribution,
'architecture': self.architecture}),
b64encode("User SIGNATURE goes here.")]
b64encode(b"User SIGNATURE goes here.").decode()]
data_2 = json.dumps(data_list_2)
self.postInformation(key_2, data_2)
self.tic()
......@@ -268,7 +276,7 @@ class TestShaDir(ShaDirMixin, ERP5TypeTestCase):
version="001",
language="en",
follow_up_value=person,
data="FILEDATA")
data=b"FILEDATA")
doc.publish()
self.tic()
......
......@@ -27,8 +27,8 @@
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
import urlparse
import httplib
import six.moves.urllib.parse
import six.moves.http_client
import unittest
import ssl
......@@ -73,7 +73,7 @@ class TestStaticWebSiteRedirection(ERP5TypeTestCase):
self.tic()
return website
def runTestRedirect(self, source_path, expected_failure=None,
def runTestRedirect(self, source_path,
use_moved_temporarily=None,
configuration_service_worker_url=None, **kw):
"""
......@@ -91,13 +91,13 @@ class TestStaticWebSiteRedirection(ERP5TypeTestCase):
if source_path.endswith("?"):
source_path = source_path[:-1]
api_scheme, api_netloc, api_path, _, _ = urlparse.urlsplit(absolute_url)
api_scheme, api_netloc, api_path, _, _ = six.moves.urllib.parse.urlsplit(absolute_url)
redirect_url = website.getLayoutProperty("redirect_domain")
redirect_location = "/".join([redirect_url, source_path])
status_to_assert = httplib.MOVED_PERMANENTLY
status_to_assert = six.moves.http_client.MOVED_PERMANENTLY
if use_moved_temporarily:
status_to_assert = httplib.FOUND
status_to_assert = six.moves.http_client.FOUND
api_netloc = '[ERP5_IPV6]:ERP5_PORT'
for url_to_check in [
......@@ -111,12 +111,12 @@ class TestStaticWebSiteRedirection(ERP5TypeTestCase):
# '%s://%s/VirtualHostBase/http/example.org:1234/erp5/web_site_module/VirtualHostRoot/%s/%s' % (api_scheme, api_netloc, website.getId(), source_path)
]:
scheme_to_check, netloc_to_check, _, _, _ = urlparse.urlsplit(url_to_check)
scheme_to_check, netloc_to_check, _, _, _ = six.moves.urllib.parse.urlsplit(url_to_check)
if (scheme_to_check == 'https'):
connection = httplib.HTTPSConnection(netloc_to_check, context=ssl._create_unverified_context(), timeout=10)
connection = six.moves.http_client.HTTPSConnection(netloc_to_check, context=ssl._create_unverified_context(), timeout=10)
else:
connection = httplib.HTTPConnection(netloc_to_check, timeout=10)
connection = six.moves.http_client.HTTPConnection(netloc_to_check, timeout=10)
self.addCleanup(connection.close)
connection.request(
method="GET",
......@@ -127,16 +127,15 @@ class TestStaticWebSiteRedirection(ERP5TypeTestCase):
if (source_path == configuration_service_worker_url):
# Test service worker URL
self.assertEqual(response.status, httplib.OK, '%s: %s' % (response.status, url_to_check))
self.assertEqual(response.status, six.moves.http_client.OK, '%s: %s' % (response.status, url_to_check))
self.assertEqual(response.getheader('Content-Type'), 'application/javascript')
self.assertTrue('self.registration.unregister()' in response_body,
response_body)
self.assertIn(b'self.registration.unregister()', response_body)
else:
self.assertEqual(response.status, status_to_assert, '%s: %s' % (response.status, url_to_check))
self.assertEqual(response.getheader(LOCATION), redirect_location)
self.assertEqual(response.getheader('Content-Type'), 'text/plain; charset=utf-8')
self.assertEqual(response_body, redirect_location)
self.assertEqual(response_body.decode('utf-8'), redirect_location)
##############################################################################
......
......@@ -90,17 +90,17 @@ class TestERP5WechatSecurePayment(TestERP5WechatSecurePaymentMixin):
def test_calculateSign_dict_simple(self):
self.assertEqual(
self.service.calculateSign({'key': 'value'}, 'mysecretkey'),
hashlib.md5("key=value&key=mysecretkey").hexdigest().upper()
hashlib.md5(b"key=value&key=mysecretkey").hexdigest().upper()
)
def test_calculateSign_dict_key_sort(self):
self.assertEqual(
self.service.calculateSign({'key0': 'value0', 'key1': 'value1'}, 'mysecretkey'),
hashlib.md5("key0=value0&key1=value1&key=mysecretkey").hexdigest().upper()
hashlib.md5(b"key0=value0&key1=value1&key=mysecretkey").hexdigest().upper()
)
self.assertEqual(
self.service.calculateSign({'key1': 'value1', 'key0': 'value0'}, 'mysecretkey'),
hashlib.md5("key0=value0&key1=value1&key=mysecretkey").hexdigest().upper()
hashlib.md5(b"key0=value0&key1=value1&key=mysecretkey").hexdigest().upper()
)
def test_navigate(self):
......
import urlparse
import six.moves.urllib.parse
import unittest
from erp5.component.mixin.TestWorkflowMixin import TestWorkflowMixin
from Products.ERP5Type.Core.Workflow import ValidationFailed
......@@ -444,7 +444,7 @@ class TestConvertedWorkflow(TestERP5WorkflowMixin):
self.tic()
ret = self.workflow.Workflow_updateSecurityRoles()
self.assertEqual(
urlparse.parse_qs(urlparse.urlparse(ret).query)['portal_status_message'],
six.moves.urllib.parse.parse_qs(six.moves.urllib.parse.urlparse(ret).query)['portal_status_message'],
["1 documents updated."])
self.tic()
self.assertEqual(text_document._View_Permission, ('Assignee', 'Assignor', 'Associate', 'Auditor', 'Author', 'Manager'))
......
......@@ -27,9 +27,13 @@
#
##############################################################################
import six
if six.PY2:
from base64 import encodestring as base64_encodebytes
else:
from base64 import encodebytes as base64_encodebytes
import base64
import httplib
import six.moves.http_client
from unittest import expectedFailure
from DateTime import DateTime
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
......@@ -59,8 +63,8 @@ class TestShaCacheExternal(ShaCacheMixin, ShaSecurityMixin, ERP5TypeTestCase):
# Define POST headers with Authentication
self.content_type = 'application/json'
authentication_string = 'lucas:lucas'
base64string = base64.encodestring(authentication_string).strip()
authentication_string = b'lucas:lucas'
base64string = base64_encodebytes(authentication_string).decode().strip()
self.header_dict = {'Authorization': 'Basic %s' % base64string,
'Content-Type': self.content_type}
......@@ -74,7 +78,7 @@ class TestShaCacheExternal(ShaCacheMixin, ShaSecurityMixin, ERP5TypeTestCase):
Test the external usage to POST information
"""
now = DateTime()
connection = httplib.HTTPConnection('%s:%s' % (self.host, self.port))
connection = six.moves.http_client.HTTPConnection('%s:%s' % (self.host, self.port))
try:
connection.request('POST', self.path, self.data, self.header_dict)
result = connection.getresponse()
......@@ -82,8 +86,8 @@ class TestShaCacheExternal(ShaCacheMixin, ShaSecurityMixin, ERP5TypeTestCase):
data = result.read()
finally:
connection.close()
self.assertEqual(self.key, data)
self.assertEqual(httplib.CREATED, result.status)
self.assertEqual(self.key, data.decode())
self.assertEqual(six.moves.http_client.CREATED, result.status)
# Check Document
document = self.portal.portal_catalog.getResultValue(portal_type='File',
......@@ -104,7 +108,7 @@ class TestShaCacheExternal(ShaCacheMixin, ShaSecurityMixin, ERP5TypeTestCase):
if not annonymous:
header_dict = self.header_dict
connection = httplib.HTTPConnection('%s:%s' % (self.host, self.port))
connection = six.moves.http_client.HTTPConnection('%s:%s' % (self.host, self.port))
try:
connection.request('GET', '/'.join([self.path, self.key]),
headers=header_dict)
......@@ -113,7 +117,7 @@ class TestShaCacheExternal(ShaCacheMixin, ShaSecurityMixin, ERP5TypeTestCase):
finally:
connection.close()
self.assertEqual(self.data, data)
self.assertEqual(httplib.OK, result.status)
self.assertEqual(six.moves.http_client.OK, result.status)
self.assertEqual(self.expected_content_type,
result.getheader("content-type"))
......@@ -129,7 +133,7 @@ class TestShaCacheExternal(ShaCacheMixin, ShaSecurityMixin, ERP5TypeTestCase):
"""
Anonymous should not be able to POST a file.
"""
connection = httplib.HTTPConnection('%s:%s' % (self.host, self.port))
connection = six.moves.http_client.HTTPConnection('%s:%s' % (self.host, self.port))
header_dict = {'Content-Type': self.content_type}
try:
connection.request('POST', self.path, self.data, header_dict)
......@@ -143,4 +147,4 @@ class TestShaCacheExternal(ShaCacheMixin, ShaSecurityMixin, ERP5TypeTestCase):
# Ref: http://www.w3.org/Protocols/rfc2616/rfc2616-sec10.html#sec10.4
# self.assertEqual(httplib.UNAUTHORIZED, result.status)
# FORBIDDEN seems more suitable for RESTful server...
self.assertEqual(httplib.FORBIDDEN, result.status)
self.assertEqual(six.moves.http_client.FORBIDDEN, result.status)
......@@ -202,7 +202,7 @@ class TestShaCacheSecurity(ShaCacheMixin, ShaSecurityMixin, SecurityTestCase):
contribution_tool)
document = self.portal.portal_contributions.newContent(
filename='test.txt',
data='test content',
data=b'test content',
reference='test-reference')
document()
document.view()
......
......@@ -31,7 +31,7 @@
import base64
import json
import os
import httplib
import six.moves.http_client
from DateTime import DateTime
from Products.ERP5Type.tests.ERP5TypeLiveTestCase import ERP5TypeTestCase
from erp5.component.test.ShaDirMixin import ShaDirMixin
......@@ -60,8 +60,8 @@ class TestShaDirExternal(ShaDirMixin, ShaSecurityMixin, ERP5TypeTestCase):
# Define POST headers with Authentication
self.content_type = 'application/json'
authentication_string = 'lucas:lucas'
base64string = base64.encodestring(authentication_string).strip()
authentication_string = b'lucas:lucas'
base64string = base64.b64encode(authentication_string).decode().strip()
self.header_dict = {'Authorization': 'Basic %s' % base64string,
'Content-Type': self.content_type}
......@@ -75,7 +75,7 @@ class TestShaDirExternal(ShaDirMixin, ShaSecurityMixin, ERP5TypeTestCase):
Test the external usage to POST information
"""
now = DateTime()
connection = httplib.HTTPConnection('%s:%s' % (self.host, self.port))
connection = six.moves.http_client.HTTPConnection('%s:%s' % (self.host, self.port))
try:
connection.request('PUT', self.path, self.data, self.header_dict)
result = connection.getresponse()
......@@ -83,7 +83,7 @@ class TestShaDirExternal(ShaDirMixin, ShaSecurityMixin, ERP5TypeTestCase):
data = result.read()
finally:
connection.close()
self.assertEqual('', data)
self.assertEqual(b'', data)
self.assertEqual(201, result.status)
# Check Data Set
......@@ -114,14 +114,14 @@ class TestShaDirExternal(ShaDirMixin, ShaSecurityMixin, ERP5TypeTestCase):
if not annonymous:
header_dict = self.header_dict
connection = httplib.HTTPConnection('%s:%s' % (self.host, self.port))
connection = six.moves.http_client.HTTPConnection('%s:%s' % (self.host, self.port))
try:
connection.request('GET', self.path, headers=header_dict)
result = connection.getresponse()
data = result.read()
finally:
connection.close()
self.assertEqual(json.dumps([json.loads(self.data)]), data)
self.assertEqual(json.dumps([json.loads(self.data)]), data.decode())
self.assertEqual(200, result.status)
self.assertEqual(self.content_type, result.getheader("content-type"))
......@@ -135,7 +135,7 @@ class TestShaDirExternal(ShaDirMixin, ShaSecurityMixin, ERP5TypeTestCase):
def test_external_post_anonymous(self):
"""
"""
connection = httplib.HTTPConnection('%s:%s' % (self.host, self.port))
connection = six.moves.http_client.HTTPConnection('%s:%s' % (self.host, self.port))
header_dict = {'Content-Type': self.content_type}
try:
connection.request('PUT', self.path, self.data, header_dict)
......@@ -153,7 +153,7 @@ class TestShaDirExternal(ShaDirMixin, ShaSecurityMixin, ERP5TypeTestCase):
data[0] = json.dumps(data[0])
data = json.dumps(data)
connection = httplib.HTTPConnection('%s:%s' % (self.host, self.port))
connection = six.moves.http_client.HTTPConnection('%s:%s' % (self.host, self.port))
try:
connection.request('PUT', self.path, data, self.header_dict)
result = connection.getresponse()
......
......@@ -216,7 +216,7 @@ class TestShaDirSecurity(ShaDirMixin, ShaSecurityMixin, SecurityTestCase):
data_set = self.portal.data_set_module.newContent(portal_type='Data Set')
document = self.portal.portal_contributions.newContent(
filename='test.txt',
data='test content',
data=b'test content',
reference='test-reference',
discover_metadata=False,
follow_up_list=[data_set.getRelativeUrl()])
......
......@@ -31,6 +31,7 @@ import transaction
import ZODB
from ZODB.DemoStorage import DemoStorage
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from six.moves import range
class TestType(unittest.TestCase):
......@@ -54,18 +55,18 @@ class TestType(unittest.TestCase):
self.tm2.begin()
x2 = self.conn2.root()['x']
x1.append(-1)
x2.extend(xrange(200))
x2.extend(range(200))
self.tm1.commit()
self.tm2.commit()
self.tm1.begin()
x1 += 401, 402
x2.extend(xrange(200, 400))
x2.extend(range(200, 400))
self.tm2.commit()
x2.append(400)
self.tm2.commit()
self.tm1.commit()
self.tm2.begin()
expected = range(-1, 403)
expected = list(range(-1, 403))
self.assertEqual(expected, list(x1))
self.assertEqual(expected, list(x2))
self.assertEqual(expected[::-1], list(reversed(x1)))
......@@ -112,7 +113,7 @@ class TestERP5(ERP5TypeTestCase):
# (see also Products.ERP5Type.patches.ZODBConnection)
for id in active_process.getRelativeUrl().split('/'):
remote = getattr(remote, id)
for x in xrange(100):
for x in range(100):
active_process.postResult(x)
remote.testActiveProcess_postResult(100)
try:
......@@ -120,7 +121,7 @@ class TestERP5(ERP5TypeTestCase):
except:
self.abort() # make failure more readable in case of regression
raise
self.assertEqual(sorted(active_process.getResultList()), range(101))
self.assertEqual(sorted(active_process.getResultList()), list(range(101)))
def test_suite():
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment