From 47d1f6ddb6387e0687f613b3db5f4fd3c2d0375b Mon Sep 17 00:00:00 2001 From: Vincent Pelletier <vincent@nexedi.com> Date: Mon, 31 Dec 2007 16:27:01 +0000 Subject: [PATCH] Add a test to check request isolation from one activity to another. git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@18570 20353a03-c40f-0410-a6d1-a30d3c3de9de --- product/CMFActivity/tests/testCMFActivity.py | 37 ++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/product/CMFActivity/tests/testCMFActivity.py b/product/CMFActivity/tests/testCMFActivity.py index 89276244dc..7cf03ea38f 100644 --- a/product/CMFActivity/tests/testCMFActivity.py +++ b/product/CMFActivity/tests/testCMFActivity.py @@ -42,6 +42,7 @@ from ZODB.POSException import ConflictError from DateTime import DateTime import cPickle as pickle from Products.CMFActivity.ActivityTool import Message +import random try: from transaction import get as get_transaction @@ -2202,6 +2203,42 @@ class TestCMFActivity(ERP5TypeTestCase): finally: delattr(Organisation, 'appendToTitle') + def test_89_RequestIsolationInsideSameTic(self, quiet=0, run=run_all_test): + """ + Check that request information do not leak from one activity to another + inside the same TIC invocation. + This only apply to queues supporting batch processing: + - SQLQueue + """ + if not run: return + if not quiet: + message = '\nCheck request isolation between messages of the same batch' + ZopeTestCase._print(message) + LOG('Testing... ',0,message) + get_transaction().commit() + self.tic() + obj = self.getPortal().organisation_module.newContent(portal_type='Organisation', title='Pending') + marker_id = 'marker_%i' % (random.randint(1, 10), ) + def putMarkerValue(self, marker_id): + self.REQUEST.set(marker_id, 1) + def checkMarkerValue(self, marker_id): + if self.REQUEST.get(marker_id) is not None: + self.setTitle('Failed') + else: + self.setTitle('Success') + try: + Organisation.putMarkerValue = putMarkerValue + Organisation.checkMarkerValue = checkMarkerValue + obj.activate(activity='SQLQueue', tag='set_first').putMarkerValue(marker_id=marker_id) + obj.activate(activity='SQLQueue', after_tag='set_first').checkMarkerValue(marker_id=marker_id) + self.assertEqual(obj.getTitle(), 'Pending') + get_transaction().commit() + self.tic() + self.assertEqual(obj.getTitle(), 'Success') + finally: + delattr(Organisation, 'putMarkerValue') + delattr(Organisation, 'checkMarkerValue') + def test_suite(): suite = unittest.TestSuite() suite.addTest(unittest.makeSuite(TestCMFActivity)) -- 2.30.9