From e4d7edeeaeef1781adab92ac21dd33ad83d888ea Mon Sep 17 00:00:00 2001 From: Sebastien Robin <seb@nexedi.com> Date: Mon, 26 Apr 2004 17:41:08 +0000 Subject: [PATCH] added test so to check if message with errors are not deleted git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@748 20353a03-c40f-0410-a6d1-a30d3c3de9de --- product/CMFActivity/tests/testCMFActivity.py | 59 ++++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/product/CMFActivity/tests/testCMFActivity.py b/product/CMFActivity/tests/testCMFActivity.py index a81498ed7d..8c1bf4f65e 100755 --- a/product/CMFActivity/tests/testCMFActivity.py +++ b/product/CMFActivity/tests/testCMFActivity.py @@ -194,6 +194,29 @@ class TestCMFActivity(ERP5TypeTestCase): self.assertEquals(len(message_list),0) self.assertEquals(2,organisation.getFoobar()) + def TryMessageWithErrorOnActivity(self, activity): + portal = self.getPortal() + def crashThisActivity(self): + self.IWillCrach() + from Products.ERP5Type.Document.Organisation import Organisation + organisation = portal.organisation._getOb(self.company_id) + Organisation.crashThisActivity = crashThisActivity + organisation.activate(activity=activity).crashThisActivity() + # Needed so that the message are commited into the queue + get_transaction().commit() + message_list = portal.portal_activities.getMessageList() + self.assertEquals(len(message_list),1) + portal.portal_activities.distribute() + portal.portal_activities.tic() + # Test if there is still the message after it crashed + message_list = portal.portal_activities.getMessageList() + self.assertEquals(len(message_list),1) + portal.portal_activities.manageCancel(organisation.getPhysicalPath(),'crashThisActivity') + # Needed so that the message are commited into the queue + get_transaction().commit() + message_list = portal.portal_activities.getMessageList() + self.assertEquals(len(message_list),0) + def test_01_DeferedSetTitleSQLDict(self, quiet=0, run=run_all_test): # Test if we can add a complete sales order if not run: return @@ -302,6 +325,42 @@ class TestCMFActivity(ERP5TypeTestCase): LOG('Testing... ',0,message) self.CallOnceWithActivity('RAMQueue') + def test_13_TryMessageWithErrorOnSQLDict(self, quiet=0, run=run_all_test): + # Test if we call methods only once + if not run: return + if not quiet: + message = '\nTry Message With Error On SQL Dict ' + ZopeTestCase._print(message) + LOG('Testing... ',0,message) + self.TryMessageWithErrorOnActivity('SQLDict') + + def test_14_TryMessageWithErrorOnSQLQueue(self, quiet=0, run=run_all_test): + # Test if we call methods only once + if not run: return + if not quiet: + message = '\nTry Message With Error On SQL Queue ' + ZopeTestCase._print(message) + LOG('Testing... ',0,message) + self.TryMessageWithErrorOnActivity('SQLQueue') + + def test_15_TryMessageWithErrorOnRAMDict(self, quiet=0, run=run_all_test): + # Test if we call methods only once + if not run: return + if not quiet: + message = '\nTry Message With Error On RAM Dict ' + ZopeTestCase._print(message) + LOG('Testing... ',0,message) + self.TryMessageWithErrorOnActivity('RAMDict') + + def test_16_TryMessageWithErrorOnRAMQueue(self, quiet=0, run=run_all_test): + # Test if we call methods only once + if not run: return + if not quiet: + message = '\nTry Message With Error On RAM Queue ' + ZopeTestCase._print(message) + LOG('Testing... ',0,message) + self.TryMessageWithErrorOnActivity('RAMQueue') + -- 2.30.9