Commit e4d7edee authored by Sebastien Robin's avatar Sebastien Robin

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
parent 12591969
...@@ -194,6 +194,29 @@ class TestCMFActivity(ERP5TypeTestCase): ...@@ -194,6 +194,29 @@ class TestCMFActivity(ERP5TypeTestCase):
self.assertEquals(len(message_list),0) self.assertEquals(len(message_list),0)
self.assertEquals(2,organisation.getFoobar()) 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): def test_01_DeferedSetTitleSQLDict(self, quiet=0, run=run_all_test):
# Test if we can add a complete sales order # Test if we can add a complete sales order
if not run: return if not run: return
...@@ -302,6 +325,42 @@ class TestCMFActivity(ERP5TypeTestCase): ...@@ -302,6 +325,42 @@ class TestCMFActivity(ERP5TypeTestCase):
LOG('Testing... ',0,message) LOG('Testing... ',0,message)
self.CallOnceWithActivity('RAMQueue') 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')
......
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