[Erp5-report] r18501 - /erp5/trunk/products/CMFActivity/tests/testCMFActivity.py
nobody at svn.erp5.org
nobody at svn.erp5.org
Mon Dec 24 20:46:10 CET 2007
Author: vincent
Date: Mon Dec 24 20:46:10 2007
New Revision: 18501
URL: http://svn.erp5.org?rev=18501&view=rev
Log:
Fix the 2 previously added tests, they were all wrong: test what happen when activity generates the error.
Add one more test for a bug found while coding above fix: object path is not a tuple when the path (string) is given to Message instance.
Add 2 more tests which test what was intended for the 2 first tests: test what happen when CMFActivity generates the error.
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=18501&r1=18500&r2=18501&view=diff
==============================================================================
--- erp5/trunk/products/CMFActivity/tests/testCMFActivity.py (original)
+++ erp5/trunk/products/CMFActivity/tests/testCMFActivity.py Mon Dec 24 20:46:10 2007
@@ -40,6 +40,8 @@
from zLOG import LOG
from ZODB.POSException import ConflictError
from DateTime import DateTime
+import cPickle as pickle
+from Products.CMFActivity.ActivityTool import Message
try:
from transaction import get as get_transaction
@@ -1927,60 +1929,218 @@
message_list = activity_tool.getMessageList()
self.assertEquals(len(message_list), 2)
- def activityModificationsViaCMFActivityConnectionRolledBackOnError(self, activity):
+ def test_83_ActivityModificationsViaCMFActivityConnectionRolledBackOnErrorSQLDict(self, quiet=0, run=run_all_test):
+ """
+ When an activity modifies tables through CMFActivity SQL connection and
+ raises, check that its changes are correctly rolled back.
+ """
+ if not run: return
+ if not quiet:
+ message = '\nCheck activity modifications via CMFActivity connection are rolled back on error (SQLDict)'
+ ZopeTestCase._print(message)
+ LOG('Testing... ',0,message)
+ get_transaction().commit()
+ self.tic()
activity_tool = self.getActivityTool()
- def modifySQLAndFail(self, connection_id):
- # Add a dumy activity which will not be executed
- # Modified table does not matter
- self.SQLDict_writeMessageList(
- uid_list=[0], # This uid is never automaticaly assigned (starts at 1)
- date_list=[DateTime().Date()],
- path_list=['dummy_activity'],
- method_id_list=['dummy_activity'],
- message_list=['dummy_message'],
- priority_list=[1],
- processing_node_list=[-4],
- group_method_id_list=[''],
- tag_list=[''],
- order_validation_text_list=['']
- )
+ def modifySQLAndFail(self, object_list, **kw):
+ # Only create the dummy activity if none is present: we would just
+ # generate missleading errors (duplicate uid).
+ if activity_tool.countMessage(method_id='dummy_activity') == 0:
+ # Add a dumy activity which will not be executed
+ # Modified table does not matter
+ method_id = 'dummy_activity'
+ path = '/'.join(self.getPhysicalPath())
+ message = Message(self, None, {}, method_id, (), {})
+ pickled_message = pickle.dumps(message)
+ self.SQLDict_writeMessageList(
+ uid_list=[0], # This uid is never automaticaly assigned (starts at 1)
+ date_list=[DateTime().Date()],
+ path_list=[path],
+ method_id_list=[method_id],
+ message_list=[pickled_message],
+ priority_list=[1],
+ processing_node_list=[-2],
+ group_method_id_list=[''],
+ tag_list=[''],
+ order_validation_text_list=['']
+ )
+ if len(object_list) == 2:
+ # Remove one entry from object list: this is understood by caller as a
+ # success for this entry.
+ object_list.pop()
+ Organisation.modifySQLAndFail = modifySQLAndFail
+ def dummy(self):
+ pass
+ Organisation.dummy = dummy
+ obj = self.getPortal().organisation_module.newContent(portal_type='Organisation')
+ group_method_id = '%s/modifySQLAndFail' % (obj.getPath(), )
+ obj.activate(activity='SQLDict', group_method_id=group_method_id).dummy()
+ obj2 = self.getPortal().organisation_module.newContent(portal_type='Organisation')
+ obj2.activate(activity='SQLDict', group_method_id=group_method_id).dummy()
+ get_transaction().commit()
+ self.flushAllActivities(silent=1, loop_size=100)
+ self.assertEquals(activity_tool.countMessage(method_id='dummy_activity'), 0)
+
+ def test_84_ActivityModificationsViaCMFActivityConnectionRolledBackOnErrorSQLQueue(self, quiet=0, run=run_all_test):
+ """
+ When an activity modifies tables through CMFActivity SQL connection and
+ raises, check that its changes are correctly rolled back.
+ """
+ if not run: return
+ if not quiet:
+ message = '\nCheck activity modifications via CMFActivity connection are rolled back on error (SQLQueue)'
+ ZopeTestCase._print(message)
+ LOG('Testing... ',0,message)
+ get_transaction().commit()
+ self.tic()
+ activity_tool = self.getActivityTool()
+ def modifySQLAndFail(self):
+ # Only create the dummy activity if none is present: we would just
+ # generate missleading errors (duplicate uid).
+ if activity_tool.countMessage(method_id='dummy_activity') == 0:
+ # Add a dumy activity which will not be executed
+ # Modified table does not matter
+ method_id = 'dummy_activity'
+ path = '/'.join(self.getPhysicalPath())
+ message = Message(self, None, {}, method_id, (), {})
+ pickled_message = pickle.dumps(message)
+ self.SQLDict_writeMessageList(
+ uid_list=[0], # This uid is never automaticaly assigned (starts at 1)
+ date_list=[DateTime().Date()],
+ path_list=[path],
+ method_id_list=[method_id],
+ message_list=[pickled_message],
+ priority_list=[1],
+ processing_node_list=[-2],
+ group_method_id_list=[''],
+ tag_list=[''],
+ order_validation_text_list=['']
+ )
# Fail
raise ValueError, 'This method always fail'
Organisation.modifySQLAndFail = modifySQLAndFail
obj = self.getPortal().organisation_module.newContent(portal_type='Organisation')
- obj.activate(activity=activity).modifySQLAndFail()
+ obj.activate(activity='SQLQueue').modifySQLAndFail()
get_transaction().commit()
self.flushAllActivities(silent=1, loop_size=100)
- self.assertEquals(activity_tool.countMessage(path='dummy_activity'), 0)
+ self.assertEquals(activity_tool.countMessage(method_id='dummy_activity'), 0)
+
+ def test_85_MessagePathMustBeATuple(self, quiet=0, run=run_all_test):
+ """
+ Message property 'object_path' must be a tuple, whatever it is generated from.
+ Possible path sources are:
+ - bare string
+ - object
+ """
+ if not run: return
+ if not quiet:
+ message = '\nCheck that message property \'object_path\' is a tuple, whatever it is generated from.'
+ ZopeTestCase._print(message)
+ LOG('Testing... ',0,message)
+ def check(value):
+ message = Message(value, None, {}, 'dummy', (), {})
+ self.assertTrue(isinstance(message.object_path, tuple))
+ # Bare string
+ check('/foo/bar')
+ # Object
+ check(self.getPortalObject().person_module)
+
+ def test_86_ActivityToolInvokeGroupFailureDoesNotCommitCMFActivitySQLConnectionSQLDict(self, quiet=0, run=run_all_test):
+ """
+ Check that CMFActivity SQL connection is rollback if activity_tool.invokeGroup raises.
+ """
+ if not run: return
+ if not quiet:
+ message = '\nCheck that activity modifications via CMFActivity connection are rolled back on ActivityTool error (SQLDict)'
+ ZopeTestCase._print(message)
+ LOG('Testing... ',0,message)
+ get_transaction().commit()
+ self.tic()
+ activity_tool = self.getActivityTool()
+ def modifySQLAndFail(self, *arg, **kw):
+ # Only create the dummy activity if none is present: we would just
+ # generate missleading errors (duplicate uid).
+ if activity_tool.countMessage(method_id='dummy_activity') == 0:
+ # Add a dumy activity which will not be executed
+ # Modified table does not matter
+ method_id = 'dummy_activity'
+ path = '/'.join(self.getPhysicalPath())
+ message = Message(self, None, {}, method_id, (), {})
+ pickled_message = pickle.dumps(message)
+ self.SQLDict_writeMessageList(
+ uid_list=[0], # This uid is never automaticaly assigned (starts at 1)
+ date_list=[DateTime().Date()],
+ path_list=[path],
+ method_id_list=[method_id],
+ message_list=[pickled_message],
+ priority_list=[1],
+ processing_node_list=[-2],
+ group_method_id_list=[''],
+ tag_list=[''],
+ order_validation_text_list=['']
+ )
+ # Fail
+ raise ValueError, 'This method always fail'
+ activity_tool.__class__.invoke = modifySQLAndFail
+ activity_tool.__class__.invokeGroup = modifySQLAndFail
+ def dummy(self):
+ pass
+ Organisation.dummy = dummy
+ obj = self.getPortal().organisation_module.newContent(portal_type='Organisation')
+ group_method_id = '%s/dummy' % (obj.getPath(), )
+ obj.activate(activity='SQLDict', group_method_id=group_method_id).dummy()
+ obj2 = self.getPortal().organisation_module.newContent(portal_type='Organisation')
+ obj2.activate(activity='SQLDict', group_method_id=group_method_id).dummy()
+ get_transaction().commit()
+ self.flushAllActivities(silent=1, loop_size=100)
+ self.assertEquals(activity_tool.countMessage(method_id='dummy_activity'), 0)
-
- def test_83_ActivityModificationsViaCMFActivityConnectionRolledBackOnErrorSQLDict(self, quiet=0, run=run_all_test):
- """
- When an activity modifies tables through CMFActivity SQL connection and
- raises, check that its changes are correctly rolled back.
- """
- if not run: return
- if not quiet:
- message = '\nCheck activity modifications via CMFActivity connection are rolled back on error (SQLDict)'
+ def test_87_ActivityToolInvokeFailureDoesNotCommitCMFActivitySQLConnectionSQLQueue(self, quiet=0, run=run_all_test):
+ """
+ Check that CMFActivity SQL connection is rollback if activity_tool.invoke raises.
+ """
+ if not run: return
+ if not quiet:
+ message = '\nCheck that activity modifications via CMFActivity connection are rolled back on ActivityTool error (SQLQueue)'
ZopeTestCase._print(message)
LOG('Testing... ',0,message)
get_transaction().commit()
self.tic()
- self.activityModificationsViaCMFActivityConnectionRolledBackOnError('SQLDict')
-
- def test_84_ActivityModificationsViaCMFActivityConnectionRolledBackOnErrorSQLQeue(self, quiet=0, run=run_all_test):
- """
- When an activity modifies tables through CMFActivity SQL connection and
- raises, check that its changes are correctly rolled back.
- """
- if not run: return
- if not quiet:
- message = '\nCheck activity modifications via CMFActivity connection are rolled back on error (SQLQueue)'
- ZopeTestCase._print(message)
- LOG('Testing... ',0,message)
- get_transaction().commit()
- self.tic()
- self.activityModificationsViaCMFActivityConnectionRolledBackOnError('SQLQueue')
+ activity_tool = self.getActivityTool()
+ def modifySQLAndFail(self, *args, **kw):
+ # Only create the dummy activity if none is present: we would just
+ # generate missleading errors (duplicate uid).
+ if activity_tool.countMessage(method_id='dummy_activity') == 0:
+ # Add a dumy activity which will not be executed
+ # Modified table does not matter
+ method_id = 'dummy_activity'
+ path = '/'.join(self.getPhysicalPath())
+ message = Message(self, None, {}, method_id, (), {})
+ pickled_message = pickle.dumps(message)
+ self.SQLDict_writeMessageList(
+ uid_list=[0], # This uid is never automaticaly assigned (starts at 1)
+ date_list=[DateTime().Date()],
+ path_list=[path],
+ method_id_list=[method_id],
+ message_list=[pickled_message],
+ priority_list=[1],
+ processing_node_list=[-2],
+ group_method_id_list=[''],
+ tag_list=[''],
+ order_validation_text_list=['']
+ )
+ # Fail
+ raise ValueError, 'This method always fail'
+ activity_tool.__class__.invoke = modifySQLAndFail
+ activity_tool.__class__.invokeGroup = modifySQLAndFail
+ def dummy(self):
+ pass
+ Organisation.dummy = dummy
+ obj = self.getPortal().organisation_module.newContent(portal_type='Organisation')
+ obj.activate(activity='SQLQueue').dummy()
+ get_transaction().commit()
+ self.flushAllActivities(silent=1, loop_size=100)
+ self.assertEquals(activity_tool.countMessage(method_id='dummy_activity'), 0)
def test_suite():
suite = unittest.TestSuite()
More information about the Erp5-report
mailing list