[Erp5-report] r19094 - /erp5/trunk/products/CMFActivity/tests/testCMFActivity.py
nobody at svn.erp5.org
nobody at svn.erp5.org
Wed Feb 6 11:40:46 CET 2008
Author: vincent
Date: Wed Feb 6 11:40:46 2008
New Revision: 19094
URL: http://svn.erp5.org?rev=19094&view=rev
Log:
Add a test to check that no actiivty is deleted when transaction commit raises.
Modified:
erp5/trunk/products/CMFActivity/tests/testCMFActivity.py
Modified: erp5/trunk/products/CMFActivity/tests/testCMFActivity.py
URL: http://svn.erp5.org/erp5/trunk/products/CMFActivity/tests/testCMFActivity.py?rev=19094&r1=19093&r2=19094&view=diff
==============================================================================
--- erp5/trunk/products/CMFActivity/tests/testCMFActivity.py (original)
+++ erp5/trunk/products/CMFActivity/tests/testCMFActivity.py Wed Feb 6 11:40:46 2008
@@ -2514,6 +2514,54 @@
LOG('Testing... ',0,message)
self.TryActivityRaiseInCommitDoesNotStallActivityConection('SQLQueue')
+ def TryActivityRaiseInCommitDoesNotLooseMessages(self, activity):
+ """
+ """
+ get_transaction().commit()
+ self.tic()
+ activity_tool = self.getActivityTool()
+ from Shared.DC.ZRDB.TM import TM
+ class dummy_tm(TM):
+ def tpc_vote(self, *ignored):
+ raise Exception, 'vote always raises'
+
+ def _finish(self):
+ pass
+
+ def _abort(self):
+ pass
+ dummy_tm_instance = dummy_tm()
+ def registerFailingTransactionManager(self, *args, **kw):
+ dummy_tm_instance._register()
+ try:
+ Organisation.registerFailingTransactionManager = registerFailingTransactionManager
+ obj = self.getPortal().organisation_module.newContent(portal_type='Organisation')
+ get_transaction().commit()
+ self.tic()
+ now = DateTime()
+ obj.activate(activity=activity).registerFailingTransactionManager()
+ get_transaction().commit()
+ self.flushAllActivities(silent=1, loop_size=100)
+ get_transaction().commit()
+ self.assertEquals(activity_tool.countMessage(method_id='registerFailingTransactionManager'), 1)
+ finally:
+ delattr(Organisation, 'registerFailingTransactionManager')
+
+ def test_98_ActivityRaiseInCommitDoesNotLooseMessagesSQLDict(self, quiet=0, run=run_all_test):
+ if not run: return
+ if not quiet:
+ message = '\nCheck that raising in commit does not loose messages (SQLDict)'
+ ZopeTestCase._print(message)
+ LOG('Testing... ',0,message)
+ self.TryActivityRaiseInCommitDoesNotLooseMessages('SQLDict')
+
+ def test_99_ActivityRaiseInCommitDoesNotLooseMessagesSQLQueue(self, quiet=0, run=run_all_test):
+ if not run: return
+ if not quiet:
+ message = '\nCheck that raising in commit does not loose messages (SQLQueue)'
+ ZopeTestCase._print(message)
+ LOG('Testing... ',0,message)
+ self.TryActivityRaiseInCommitDoesNotLooseMessages('SQLQueue')
def test_suite():
More information about the Erp5-report
mailing list